utun2/tools/chatgui/src/call_camera_worker.cpp

#include "call_camera_source.h"
#include "call_camera_protocol.h"
#include "call_camera_buffer.h"
#include "call_video_capture_clock.h"
#include "call_camera_rate.h"
#include "../../../src/call/call_video_playout.h"
#include "debug_config.h"
#include <QGuiApplication>
#include <QTimer>
#include <mutex>
#include <climits>
#include <algorithm>
#include <atomic>
#include <cstring>
#ifdef HAVE_STREAM_CAPTURE
#include <QCamera>
#include <QMediaDevices>
#include <QMediaCaptureSession>
#include <QVideoSink>

using namespace CallCameraProtocol;

class QtCallCameraSource : public CallCameraSource {
    std::unique_ptr<QCamera> camera;
    std::unique_ptr<QMediaCaptureSession> session;
    std::unique_ptr<QVideoSink> sink;
    CallCameraRate rate;
public:
    ~QtCallCameraSource() override { stop(); }
    CallCameraDevices devices() override {
        CallCameraDevices result;
        for (const auto& device : QMediaDevices::videoInputs()) result.append({device.description(), device.id()});
        return result;
    }
    bool start(const QByteArray& device, int width, int height) override {
        QCameraDevice selected;
        for (const auto& item : QMediaDevices::videoInputs()) if (item.id() == device) selected = item;
        if (selected.isNull()) { emit error(QStringLiteral("Camera is unavailable")); return false; }
        session = std::make_unique<QMediaCaptureSession>(); sink = std::make_unique<QVideoSink>();
        camera = std::make_unique<QCamera>(selected);
        session->setCamera(camera.get()); session->setVideoSink(sink.get());
        QCameraFormat best; int distance = INT_MAX;
        for (const auto& format : selected.videoFormats()) {
            if (format.maxFrameRate() < 29.9) continue;
            int score = std::abs(format.resolution().width() - width) + std::abs(format.resolution().height() - height);
            if (score < distance) { best = format; distance = score; }
        }
        if (!best.isNull()) camera->setCameraFormat(best);
        const auto format = camera->cameraFormat();
        DEBUG_INFO(DEBUG_CATEGORY_CALL, "camera selected device=%s size=%dx%d pixel=%d fps=%.3f..%.3f",
            device.constData(), format.resolution().width(), format.resolution().height(), format.pixelFormat(),
            format.minFrameRate(), format.maxFrameRate());
        connect(camera.get(), &QCamera::activeChanged, this, [this, device](bool active) {
            if (active) rate.activate(device); else rate.restore();
        });
        connect(camera.get(), &QCamera::errorOccurred, this, [this](QCamera::Error, const QString& message) { emit error(message); });
        connect(sink.get(), &QVideoSink::videoFrameChanged, this, &CallCameraSource::frame, Qt::DirectConnection);
        camera->start(); return true;
    }
    void stop() override {
        if (sink) disconnect(sink.get(), nullptr, this, nullptr);
        if (camera) { disconnect(camera.get(), nullptr, this, nullptr); camera->stop(); }
        rate.restore(); session.reset(); camera.reset(); sink.reset();
    }
};

class CameraWorker : public QObject {
    struct Pending {
        std::mutex lock;
        QVideoFrame raw;
        qint64 arrival = 0;
        bool posted = false;
    };
    QLocalSocket socket;
    CallCameraBuffer buffer;
    uchar* pixels = nullptr;
    Reader reader;
    std::unique_ptr<CallCameraSource> source;
    std::shared_ptr<Pending> pending;
    QMetaObject::Connection captureConnection, errorConnection;
    CallVideoCaptureClock clock;
    quint64 generation = 0, inFlight = 0;
    bool sending = false;
    int width = 640, height = 480;
    qint64 lastPts = -1, reportAt = 0, convertMax = 0;
    unsigned frames = 0;

    void send(const QByteArray& data) {
        if (socket.write(data) != data.size()) {
            DEBUG_ERROR(DEBUG_CATEGORY_CALL, "camera worker IPC write failed: %s", qPrintable(socket.errorString()));
            QCoreApplication::exit(2);
        }
        socket.flush(); // Стадия должна уйти до потенциально блокирующего вызова драйвера.
    }
    void stage(const QString& name) { send(packet(Stage, generation, [&](QDataStream& out) { out << name; })); }
    void stop() {
        stage(QStringLiteral("stop"));
        disconnect(captureConnection); disconnect(errorConnection); source->stop(); pending.reset();
        DEBUG_INFO(DEBUG_CATEGORY_CALL, "camera stopped generation=%llu", (unsigned long long)generation);
        stage(QStringLiteral("idle"));
    }
    void convert(std::shared_ptr<Pending> slot, quint64 revision) {
        if (slot != pending || revision != generation || sending) return;
        QVideoFrame raw; qint64 arrival;
        { std::lock_guard<std::mutex> guard(slot->lock);
          raw.swap(slot->raw); arrival = slot->arrival; slot->posted = false; }
        if (!raw.isValid()) return;
        const qint64 started = call_video_playout_now_us(), sourcePts = raw.startTime();
        if (clock.source >= 0 && sourcePts < clock.source)
            DEBUG_WARN(DEBUG_CATEGORY_CALL, "camera clock reset source=%lld previous=%lld", (long long)sourcePts, (long long)clock.source);
        const qint64 pts = clock.map(sourcePts, arrival);
        if (pts <= lastPts || started - pts > CallCameraMaxFrameAgeUs) {
            DEBUG_WARN(DEBUG_CATEGORY_CALL, "camera skips old frame age_us=%lld pts=%lld last=%lld",
                (long long)(started - pts), (long long)pts, (long long)lastPts); return;
        }
        stage(QStringLiteral("convert"));
        QImage image = raw.toImage();
        if (image.isNull()) {
            DEBUG_ERROR(DEBUG_CATEGORY_CALL, "camera frame conversion failed");
            send(packet(Error, generation, [](QDataStream& out) { out << QStringLiteral("Camera frame conversion failed"); })); return;
        }
        if (image.width() > width || image.height() > height) image = image.scaled(width, height, Qt::KeepAspectRatio, Qt::FastTransformation);
        image = image.convertToFormat(QImage::Format_RGB32);
        const qint64 converted = call_video_playout_now_us();
        if (converted - pts <= CallCameraMaxFrameAgeUs) {
            std::memcpy(pixels, image.constBits(), size_t(image.width()) * image.height() * 4);
            std::atomic_thread_fence(std::memory_order_release);
            auto bytes = packet(Frame, generation, [&](QDataStream& out) {
                out << pts << qint64(clock.correction) << qint32(image.width()) << qint32(image.height());
            });
            sending = true; inFlight = generation; lastPts = pts; send(bytes); ++frames;
        } else DEBUG_WARN(DEBUG_CATEGORY_CALL, "camera conversion exceeded frame age budget age_us=%lld", (long long)(converted - pts));
        stage(QStringLiteral("release")); image = {}; raw = {}; stage(QStringLiteral("running"));
        convertMax = std::max(convertMax, call_video_playout_now_us() - started);
        if (started - reportAt >= 1000000) {
            DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "camera flow frames=%u convert_max_us=%lld age_us=%lld clock_correction_us=%lld",
                frames, (long long)convertMax, (long long)(converted - pts), (long long)clock.correction);
            reportAt = started; convertMax = 0; frames = 0;
        }
    }
    bool receive(quint8 type, quint64 revision, QDataStream& in) {
        if (type == Buffer) {
            QString name; in >> name;
            if (pixels) return false;
            if (!buffer.attach(name)) {
                DEBUG_ERROR(DEBUG_CATEGORY_CALL, "camera frame mapping failed: %s", qPrintable(buffer.errorString())); return false;
            }
            pixels = buffer.data();
        } else if (type == Quit) { stop(); QCoreApplication::quit(); }
        else if (type == Ack) {
            if (sending && revision == inFlight) { sending = false; if (pending) convert(pending, generation); }
        } else if (type == Configure) {
            bool enabled = false; QByteArray device; qint32 w = 0, h = 0; in >> enabled >> device >> w >> h;
            if (in.status() != QDataStream::Ok || w < 1 || h < 1 || w > 1280 || h > 720 || !pixels) return false;
            stop(); generation = revision; width = w; height = h; clock = {}; lastPts = -1;
            if (!enabled) { stage(QStringLiteral("idle")); return true; }
            stage(QStringLiteral("start"));
            pending = std::make_shared<Pending>();
            errorConnection = connect(source.get(), &CallCameraSource::error, this, [this, revision](const QString& error) {
                if (revision != generation) return;
                DEBUG_ERROR(DEBUG_CATEGORY_CALL, "camera backend: %s", qPrintable(error));
                send(packet(Error, revision, [&](QDataStream& out) { out << error; }));
            }, Qt::QueuedConnection);
            captureConnection = connect(source.get(), &CallCameraSource::frame, this,
                [this, slot = pending, revision](const QVideoFrame& frame) {
                    if (!frame.isValid()) return;
                    const qint64 arrival = call_video_playout_now_us();
                    QVideoFrame previous; bool post;
                    { std::lock_guard<std::mutex> guard(slot->lock);
                      previous.swap(slot->raw); slot->raw = frame; slot->arrival = arrival;
                      post = !slot->posted; slot->posted = true; }
                    if (post) QMetaObject::invokeMethod(this, [this, slot, revision] { convert(slot, revision); }, Qt::QueuedConnection);
                }, Qt::DirectConnection);
            if (source->start(device, width, height)) stage(QStringLiteral("running"));
        } else return false;
        return true;
    }
public:
    explicit CameraWorker(std::unique_ptr<CallCameraSource> backend) : source(std::move(backend)) {
        connect(&socket, &QLocalSocket::connected, this, [this] {
            stage(QStringLiteral("enumerate"));
            const auto devices = source->devices();
            send(packet(Devices, 0, [&](QDataStream& out) {
                out << quint32(devices.size()); for (const auto& device : devices) out << device.first << device.second;
            }));
            stage(QStringLiteral("idle"));
        });
        connect(&socket, &QLocalSocket::readyRead, this, [this] {
            if (!reader.read(socket, [this](quint8 type, quint64 revision, QDataStream& in) { return receive(type, revision, in); })) {
                DEBUG_ERROR(DEBUG_CATEGORY_CALL, "camera worker invalid command"); QCoreApplication::exit(2);
            }
        });
        connect(&socket, &QLocalSocket::disconnected, qApp, &QCoreApplication::quit);
        connect(&socket, &QLocalSocket::errorOccurred, this, [this](QLocalSocket::LocalSocketError) {
            DEBUG_ERROR(DEBUG_CATEGORY_CALL, "camera worker IPC: %s", qPrintable(socket.errorString())); QCoreApplication::exit(2);
        });
    }
    ~CameraWorker() override { disconnect(captureConnection); source->stop(); }
    void connectTo(const QString& endpoint) { socket.connectToServer(endpoint); }
};

int call_camera_worker_run(const QString& endpoint, std::unique_ptr<CallCameraSource> source) {
    CameraWorker worker(std::move(source)); worker.connectTo(endpoint); return QCoreApplication::exec();
}
#endif

int call_camera_worker_main(int argc, char** argv) {
#ifdef HAVE_STREAM_CAPTURE
    QGuiApplication app(argc, argv); debug_config_init(); debug_set_level(DEBUG_LEVEL_INFO);
    debug_set_category_level(DEBUG_CATEGORY_CALL, DEBUG_LEVEL_DEBUG);
    if (argc != 3) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "camera worker endpoint missing"); return 2; }
    return call_camera_worker_run(QString::fromLocal8Bit(argv[2]), std::make_unique<QtCallCameraSource>());
#else
    Q_UNUSED(argc); Q_UNUSED(argv); return 2;
#endif
}