jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Decode and play desktop live H264 and Opus streams

master
evgeny 7 days ago
parent
0a5b9b50f1
commit
10f197c817
2 changed files with 342 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 303
      tools/chatgui/src/streamplayer_window.cpp
  2. 39
      tools/chatgui/src/streamplayer_window.h

303
tools/chatgui/src/streamplayer_window.cpp
Unescape View File

@ -0,0 +1,303 @@
#include "streamplayer_window.h"
#include "videoplayer_window.h"
#include "sound_manager.h"
#include "../transport/stream_bridge.h"
#include <QVBoxLayout>
#include <QHBoxLayout>
#include <QLabel>
#include <QPushButton>
#include <QSlider>
#include <QCloseEvent>
#include <QKeyEvent>
#include <QTransform>
#include <atomic>
#include <thread>
#include <mutex>
#include <deque>
#include <array>
#include <chrono>
#include <algorithm>
#include <cstring>
extern "C" {
#include "../../../lib/debug_config.h"
#include "../../../lib/opus_codec.h"
#include "../../../src/stream/stream.h"
#ifdef HAVE_FFMPEG
#include <libavcodec/avcodec.h>
#include <libswscale/swscale.h>
#endif
}
namespace {
int64_t nowUs() {
return std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
}
std::atomic<uint64_t> nextId{0};
QString text(const char* value) { return QCoreApplication::translate("ChatGui", value); }
}
struct StreamPlayerWindow::Playback {
struct Video { QImage image; int64_t pts; };
struct Audio { std::array<int16_t, 960> pcm; int64_t pts; unsigned offset = 0; };
uint64_t id = ++nextId, group = 0, source = 0, epoch = 0;
std::atomic<bool> running{true}, audioAvailable{false};
std::atomic<float> volume{1.0f};
std::thread worker;
std::mutex lock;
std::deque<Video> videos;
std::deque<Audio> audios;
int64_t prefillUntil = 0, videoBase = -1, videoAt = 0, audioPts = -1, audioAt = 0;
uint64_t shown = 0, underruns = 0;
bool ended = false;
int64_t clock(int64_t now) const {
if (audioAvailable && audioPts >= 0 && now - audioAt < 200000) return audioPts + now - audioAt;
return videoBase >= 0 ? videoBase + now - videoAt : 0;
}
};
StreamPlayerWindow::StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t epoch, const QString& name, QWidget* parent)
: QWidget(parent), m_playback(std::make_unique<Playback>()) {
m_playback->group = group; m_playback->source = source; m_playback->epoch = epoch;
setWindowTitle(text("Live stream: %1").arg(name)); resize(800, 500);
auto* layout = new QVBoxLayout(this);
layout->addWidget(new QLabel(name, this));
m_status = new QLabel(text("Connecting…"), this); layout->addWidget(m_status);
m_render = new VideoRenderWidget(this); layout->addWidget(m_render, 1);
auto* controls = new QHBoxLayout;
auto* volume = new QSlider(Qt::Horizontal, this); volume->setRange(0, 100); volume->setValue(100);
volume->setAccessibleName(text("Stream volume")); controls->addWidget(new QLabel(text("Volume"), this)); controls->addWidget(volume);
auto* fullscreen = new QPushButton(text("Fullscreen"), this); controls->addWidget(fullscreen);
auto* close = new QPushButton(text("Close player"), this); controls->addWidget(close); layout->addLayout(controls);
connect(volume, &QSlider::valueChanged, this, [this](int value) { m_playback->volume = value / 100.0f; });
connect(fullscreen, &QPushButton::clicked, this, [this] { if (isFullScreen()) showNormal(); else showFullScreen(); });
connect(close, &QPushButton::clicked, this, &QWidget::close);
connect(this, &StreamPlayerWindow::statusChanged, m_status, &QLabel::setText, Qt::QueuedConnection);
connect(SoundManager::instance(), &SoundManager::contextAboutToReset, this, [this] {
m_audio.close(); m_playback->audioAvailable = false;
std::lock_guard<std::mutex> guard(m_playback->lock);
m_playback->audios.clear(); m_playback->audioPts = -1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream audio context reset id=%llu", (unsigned long long)m_playback->id);
});
connect(SoundManager::instance(), &SoundManager::contextRestored, this, &StreamPlayerWindow::openAudio);
connect(SoundManager::instance(), &SoundManager::devicesChanged, this, [this] { m_audio.close(); openAudio(); });
openAudio();
connect(&m_timer, &QTimer::timeout, this, &StreamPlayerWindow::tick); m_timer.start(15);
m_playback->worker = std::thread(&StreamPlayerWindow::decode, this);
}
StreamPlayerWindow::~StreamPlayerWindow() { stop(); }
/* Output принадлежит GUI и закрывается до shutdown общего miniaudio-контекста. */
void StreamPlayerWindow::openAudio() {
if (!m_playback->running || m_audio.initialized()) return;
ma_context* context = SoundManager::instance()->context();
if (!context) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop stream audio unavailable: no context"); return; }
auto config = ma_device_config_init(ma_device_type_playback);
config.sampleRate = 48000; config.playback.channels = 1; config.playback.format = ma_format_s16;
ma_device_id device;
if (SoundManager::deviceId(SoundManager::instance()->playbackDevice(), &device)) config.playback.pDeviceID = &device;
bool initialized = m_audio.init(context, config, "stream", m_playback->id, [this](const void*, void* output, unsigned frames) {
auto* pcm = static_cast<int16_t*>(output); std::fill(pcm, pcm + frames, 0);
auto& p = *m_playback; std::lock_guard<std::mutex> guard(p.lock);
int64_t now = nowUs();
if (!p.prefillUntil || now < p.prefillUntil) return;
unsigned written = 0; float volume = p.volume.load(); int64_t firstPts = -1;
while (written < frames && !p.audios.empty()) {
auto& packet = p.audios.front();
if (firstPts < 0) firstPts = packet.pts + packet.offset * 1000000LL / 48000 - written * 1000000LL / 48000;
unsigned count = std::min(frames - written, 960 - packet.offset);
for (unsigned i = 0; i < count; i++) pcm[written + i] = static_cast<int16_t>(packet.pcm[packet.offset + i] * volume);
written += count; packet.offset += count;
if (packet.offset == 960) p.audios.pop_front();
}
if (firstPts >= 0) { p.audioPts = firstPts; p.audioAt = m_audio.callbackTimeUs(); }
else if (p.audioPts >= 0) {
p.audioPts += m_audio.callbackTimeUs() - p.audioAt; p.audioAt = m_audio.callbackTimeUs();
}
if (written < frames && p.audioPts >= 0 && !p.ended) p.underruns++;
});
if (!initialized || !m_audio.start()) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream audio device failed id=%llu", (unsigned long long)m_playback->id);
m_audio.close(); m_playback->audioAvailable = false; return;
}
m_playback->audioAvailable = true;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream audio output ready id=%llu", (unsigned long long)m_playback->id);
}
/* GUI забирает последний готовый кадр по аппаратным часам, не накапливая Qt-сигналы с QImage. */
void StreamPlayerWindow::tick() {
auto& p = *m_playback;
if (m_audio.needsRecovery()) { m_audio.close(); p.audioAvailable = false; openAudio(); }
QImage image; bool finished = false;
{
std::lock_guard<std::mutex> guard(p.lock);
int64_t now = nowUs(), clock = p.clock(now);
if (now >= p.prefillUntil) while (!p.videos.empty() && p.videos.front().pts <= clock) {
image = std::move(p.videos.front().image); p.videos.pop_front();
}
finished = p.ended && p.videos.empty() && (p.audios.empty() || !p.audioAvailable);
}
if (!image.isNull()) {
m_render->setFrame(image);
if (!p.shown++) {
m_status->setText(text("Live"));
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop first frame id=%llu %dx%d", (unsigned long long)p.id, image.width(), image.height());
}
emit framePresented();
}
if (finished) { m_status->setText(text("Stream ended")); m_audio.close(); m_timer.stop(); }
}
/* Закрытие может происходить до выполнения open actor; session_id отменяет именно эту попытку. */
void StreamPlayerWindow::stop() {
if (!m_playback->running.exchange(false) && !m_playback->worker.joinable()) return;
m_timer.stop(); gui_stream_stop(m_playback->id);
if (m_playback->worker.joinable()) m_playback->worker.join();
m_audio.close(); m_playback->audioAvailable = false;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream player stopped id=%llu shown=%llu underruns=%llu",
(unsigned long long)m_playback->id, (unsigned long long)m_playback->shown, (unsigned long long)m_playback->underruns);
}
void StreamPlayerWindow::closeEvent(QCloseEvent* event) { stop(); QWidget::closeEvent(event); }
void StreamPlayerWindow::keyPressEvent(QKeyEvent* event) {
if (event->key() == Qt::Key_Escape) { if (isFullScreen()) showNormal(); else close(); }
else QWidget::keyPressEvent(event);
}
/* Worker владеет codec contexts. GAP/pressure сбрасывает refs и часы до следующего ключевого кадра. */
void StreamPlayerWindow::decode() {
auto& p = *m_playback;
#ifndef HAVE_FFMPEG
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop live decoder unavailable: FFmpeg disabled");
emit statusChanged(text("Live video playback requires FFmpeg"));
#else
AVCodecContext* decoder = nullptr; AVFrame* frame = av_frame_alloc(); AVPacket* packet = av_packet_alloc();
SwsContext* converter = nullptr;
opus_codec_decoder_t* opus = nullptr;
bool waitKey = true, opened = false;
int64_t deadline = nowUs() + 10000000, reportAt = nowUs(), expectedAudio = -1;
uint64_t decoded = 0, discarded = 0;
auto reset = [&] {
avcodec_free_context(&decoder);
opus_codec_decoder_destroy(opus); opus = nullptr; expectedAudio = -1; waitKey = true;
std::lock_guard<std::mutex> guard(p.lock);
p.videos.clear(); p.audios.clear(); p.prefillUntil = 0; p.videoBase = p.audioPts = -1;
emit statusChanged(text("Waiting for a key frame…"));
};
auto fail = [&](const QString& reason) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer failed id=%llu: %s", (unsigned long long)p.id, qPrintable(reason));
emit statusChanged(reason);
};
if (!frame || !packet) fail(text("Stream decoder allocation failed"));
else if (!gui_stream_open(p.id, p.group, p.source, p.epoch)) fail(text("Could not connect to the stream"));
else while (p.running) {
StreamRecord record;
if (!gui_stream_pull(p.id, record)) {
if (!gui_stream_ready()) { fail(text("Chat service is stopped")); break; }
if ((!opened || waitKey) && nowUs() > deadline) { fail(text("Stream connection timed out")); break; }
std::this_thread::sleep_for(std::chrono::milliseconds(5)); continue;
}
if (record.kind == StreamRecordKind::Opened) { opened = true; continue; }
if (record.kind == StreamRecordKind::Gap) { reset(); deadline = nowUs() + 10000000; continue; }
if (record.kind == StreamRecordKind::Closed) {
const char* reason = record.reason == STREAM_CLOSE_SOURCE_CHANGED ? "The stream restarted. Select a source again."
: record.reason == STREAM_CLOSE_MEMBER ? "The source is no longer a channel member"
: record.reason == STREAM_CLOSE_SERVICE ? "Chat service is stopped"
: record.reason == STREAM_CLOSE_GROUP ? "The channel is closed" : "Could not continue the stream";
fail(text(reason)); break;
}
if (record.kind == StreamRecordKind::End) { std::lock_guard<std::mutex> guard(p.lock); p.ended = true; break; }
if (record.pts > INT64_MAX) { fail(text("Invalid stream timestamp")); break; }
if (record.kind == StreamRecordKind::Audio) {
if (!p.audioAvailable) continue;
if (!opus) opus = opus_codec_decoder_create(48000, 1);
if (!opus) { fail(text("Stream audio decoder failed")); break; }
if (expectedAudio >= 0 && (static_cast<int64_t>(record.pts) < expectedAudio - 10000
|| static_cast<int64_t>(record.pts) > expectedAudio + 100000)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop audio discontinuity id=%llu expected=%lld pts=%llu",
(unsigned long long)p.id, (long long)expectedAudio, (unsigned long long)record.pts);
reset(); deadline = nowUs() + 10000000; continue;
}
int missing = expectedAudio < 0 ? 0 : static_cast<int>((static_cast<int64_t>(record.pts) - expectedAudio + 10000) / 20000);
bool failed = false, pressure = false;
for (int i = 0; i <= missing; i++) {
Playback::Audio audio{}; audio.pts = i == missing ? static_cast<int64_t>(record.pts) : expectedAudio + i * 20000;
int count = opus_codec_decode(opus, i == missing ? reinterpret_cast<const uint8_t*>(record.data.constData()) : nullptr,
i == missing ? record.data.size() : 0, audio.pcm.data(), 960);
if (count != 960) { fail(text("Stream audio decoder failed")); failed = true; break; }
std::lock_guard<std::mutex> guard(p.lock);
if (!p.prefillUntil) p.prefillUntil = nowUs() + 150000;
if (p.audios.size() >= 25) { pressure = true; break; }
p.audios.push_back(std::move(audio));
}
if (failed) break;
if (pressure) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop decoded audio pressure id=%llu; wait KEY", (unsigned long long)p.id);
reset(); deadline = nowUs() + 10000000;
} else expectedAudio = static_cast<int64_t>(record.pts) + 20000;
continue;
}
if (waitKey && !(record.flags & STREAM_MEDIA_KEY)) continue;
if (!decoder) {
const AVCodec* codec = avcodec_find_decoder(AV_CODEC_ID_H264);
decoder = codec ? avcodec_alloc_context3(codec) : nullptr;
if (!decoder || avcodec_open2(decoder, codec, nullptr) < 0) { fail(text("Stream video decoder failed")); break; }
decoder->pkt_timebase = AVRational{1, 1000000};
waitKey = false;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop H264 decoder started id=%llu", (unsigned long long)p.id);
}
av_packet_unref(packet);
if (av_new_packet(packet, record.data.size()) < 0) { fail(text("Stream decoder allocation failed")); break; }
std::memcpy(packet->data, record.data.constData(), static_cast<size_t>(record.data.size()));
packet->pts = packet->dts = static_cast<int64_t>(record.pts);
int sent = avcodec_send_packet(decoder, packet);
if (sent < 0) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop H264 packet rejected result=%d id=%llu", sent, (unsigned long long)p.id);
reset(); deadline = nowUs() + 10000000; continue;
}
int received;
while ((received = avcodec_receive_frame(decoder, frame)) == 0) {
decoded++;
if (frame->width <= 0 || frame->height <= 0 || frame->width > 4096 || frame->height > 4096) {
fail(text("Invalid stream video dimensions")); p.running = false; break;
}
int64_t pts = frame->best_effort_timestamp == AV_NOPTS_VALUE ? static_cast<int64_t>(record.pts) : frame->best_effort_timestamp;
{
std::lock_guard<std::mutex> guard(p.lock);
if (!p.prefillUntil) p.prefillUntil = nowUs() + 150000;
if (p.videoBase < 0) { p.videoBase = pts; p.videoAt = p.prefillUntil; }
if (pts < p.clock(nowUs()) - 100000) { discarded++; continue; }
}
converter = sws_getCachedContext(converter, frame->width, frame->height, static_cast<AVPixelFormat>(frame->format),
frame->width, frame->height, AV_PIX_FMT_RGB24, SWS_BILINEAR, nullptr, nullptr, nullptr);
QImage image(frame->width, frame->height, QImage::Format_RGB888);
if (!converter || image.isNull()) { fail(text("Stream decoder allocation failed")); p.running = false; break; }
uint8_t* destination[] = {image.bits(), nullptr, nullptr, nullptr}; int strides[] = {image.bytesPerLine(), 0, 0, 0};
if (sws_scale(converter, frame->data, frame->linesize, 0, frame->height, destination, strides) != frame->height) {
fail(text("Stream video conversion failed")); p.running = false; break;
}
if (record.format.rotation) image = image.transformed(QTransform().rotate(record.format.rotation));
std::lock_guard<std::mutex> guard(p.lock);
if (p.videos.size() == 6) { p.videos.pop_front(); discarded++; }
p.videos.push_back({std::move(image), pts});
}
if (received != AVERROR(EAGAIN) && received != AVERROR_EOF && p.running) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop H264 decode rejected result=%d id=%llu", received, (unsigned long long)p.id);
reset(); deadline = nowUs() + 10000000;
}
if (nowUs() - reportAt >= 1000000) {
reportAt = nowUs(); std::lock_guard<std::mutex> guard(p.lock);
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "desktop viewer id=%llu decoded=%llu discarded=%llu videoQ=%zu audioQ=%zu underruns=%llu",
(unsigned long long)p.id, (unsigned long long)decoded, (unsigned long long)discarded,
p.videos.size(), p.audios.size(), (unsigned long long)p.underruns);
}
}
avcodec_free_context(&decoder); av_frame_free(&frame); av_packet_free(&packet);
if (converter) sws_freeContext(converter);
opus_codec_decoder_destroy(opus);
#endif
gui_stream_stop(p.id);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop decoder released id=%llu", (unsigned long long)p.id);
}

39
tools/chatgui/src/streamplayer_window.h
Unescape View File

@ -0,0 +1,39 @@
/* Просмотр одной live-трансляции: FFmpeg/Opus в worker, ограниченные очереди PCM/кадров.
* GUI владеет output device и окном; audio callback публикует часы физического воспроизведения.
* Закрытие останавливает worker и снимает reader. Публикация и общий транспорт не закрываются. */
#pragma once
#include <QWidget>
#include <QTimer>
#include <memory>
#include <cstdint>
#include "audio_device.h"
class VideoRenderWidget;
class QLabel;
class QCloseEvent;
class QKeyEvent;
class StreamPlayerWindow : public QWidget {
Q_OBJECT
public:
explicit StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t epoch, const QString& name, QWidget* parent = nullptr);
~StreamPlayerWindow() override;
void stop();
signals:
void statusChanged(const QString& status);
void framePresented();
protected:
void closeEvent(QCloseEvent* event) override;
void keyPressEvent(QKeyEvent* event) override;
private:
struct Playback;
std::unique_ptr<Playback> m_playback;
AudioDevice m_audio;
QTimer m_timer;
VideoRenderWidget* m_render = nullptr;
QLabel* m_status = nullptr;
void openAudio();
void tick();
void decode();
};
Write Preview
Loading…
Cancel
Save