LlamaCasty/Services/VideoClipPipe.cs

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using ytLive.Helpers;
using ytLive.Services.Audio;
using ytLive.Services.Encoder;

namespace ytLive.Services;

/// <summary>
/// One video FILE played through ffmpeg — the shared pipe behind every clip, so a media
/// layer (TASK 21) and an alert clip (TASK 47) decode identically. Owns two subprocesses
/// on the same file: video (<c>rawvideo bgra</c>, scaled to the target rect) and audio
/// (<c>f32le -ac 2 -ar 48000</c>, matching the mixer's native format so no resampler is
/// needed). Both are paced to real time, so the picture and the sound stay together even
/// though they arrive on separate pipes.
/// <para><b>Looping is the media case</b>: on natural EOF the whole pass restarts — a fresh
/// pair of processes, because a <see cref="Process"/> cannot be re-started. An alert clip
/// never loops, so its wrapper simply never sets <see cref="Looping"/>.</para>
/// <para>The seams (locator, process factory, frame-rate probe, delay) exist so tests drive
/// a fake decoder with no ffmpeg, no codec, and no audio device.</para>
/// </summary>
public sealed class VideoClipPipe : IDisposable
{
    public const int AudioSampleRate = 48000;
    public const int AudioChannels = 2;
    private const int AudioChunkFloats = 4800; // 50 ms of stereo

    public event Action<VideoFrame>? FrameAvailable;
    public event Action<AudioSample>? AudioAvailable;
    public event Action? Completed;

    /// <summary>Restart the pass on natural EOF (media/BRB loops).</summary>
    public bool Looping { get; set; }

    private readonly string _path;
    private readonly int _width;
    private readonly int _height;
    private readonly IFfmpegLocator _locator;
    private readonly Func<IDecodeProcess> _processFactory;
    private readonly IFrameRateProbe? _probe;
    private readonly Func<TimeSpan, CancellationToken, Task> _delay;
    private readonly bool _audio;
    private readonly CancellationTokenSource _cts = new();
    private IDisposable? _run;
    private IDecodeProcess? _activeVideo;
    private IDecodeProcess? _activeAudio;

    public VideoClipPipe(
        string path,
        int width,
        int height,
        IFfmpegLocator locator,
        Func<IDecodeProcess> processFactory,
        IFrameRateProbe? frameRateProbe = null,
        Func<TimeSpan, CancellationToken, Task>? delay = null,
        bool audio = true)
    {
        _path = path;
        _width = width;
        _height = height;
        _locator = locator;
        _processFactory = processFactory;
        _probe = frameRateProbe;
        _delay = delay ?? ((t, ct) => Task.Delay(t, ct));
        _audio = audio;
    }

    public Task StartAsync(CancellationToken cancellationToken = default)
    {
        if (_run != null) return Task.CompletedTask;
        _run = _cts.Token.Register(StopInternal);
        _ = Task.Run(RunAsync, CancellationToken.None);
        return Task.CompletedTask;
    }

    public Task StopAsync()
    {
        _cts.Cancel();
        _run = null;
        return Task.CompletedTask;
    }

    public void Dispose()
    {
        _cts.Cancel();
        _cts.Dispose();
        _activeVideo?.Dispose();
        _activeAudio?.Dispose();
    }

    private void StopInternal()
    {
        Kill(_activeVideo);
        Kill(_activeAudio);
    }

    private async Task RunAsync()
    {
        var ct = _cts.Token;
        var frames = 0;
        var chunks = 0;
        try
        {
            double? frameRate = null;
            if (_probe != null)
                frameRate = await _probe.ProbeAsync(_path, ct).ConfigureAwait(false);
            var frameDuration = frameRate is > 0 ? TimeSpan.FromSeconds(1.0 / frameRate.Value) : TimeSpan.Zero;

            var ffmpegPath = await _locator.LocateAsync(ct).ConfigureAwait(false);

            do
            {
                // Both pipes per pass, awaited together: a restart must begin a new
                // video AND a new audio stream, or the picture drifts from the sound
                // after the first loop.
                var video = RunVideoAsync(ffmpegPath, frameDuration, ct);
                var audio = _audio ? RunAudioAsync(ffmpegPath, ct) : Task.FromResult(0);
                frames += await video.ConfigureAwait(false);
                chunks += await audio.ConfigureAwait(false);
            }
            while (Looping && !ct.IsCancellationRequested);
        }
        catch (OperationCanceledException)
        {
            return;
        }
        catch (Exception ex)
        {
            // A failed decode is not fatal: the alert layer falls back to its animations,
            // and a media layer just holds its last frame.
            AppLog.Write($"Video clip decode failed ({Path.GetFileName(_path)}): {ex.Message}");
            return;
        }

        if (!ct.IsCancellationRequested)
        {
            AppLog.Write($"Video clip end: {Path.GetFileName(_path)} frames={frames} audioChunks={chunks}");
            Completed?.Invoke();
        }
    }

    private async Task<int> RunVideoAsync(string ffmpegPath, TimeSpan frameDuration, CancellationToken ct)
    {
        var args =
            $"-hide_banner -loglevel error -i \"{_path}\" " +
            $"-f rawvideo -pix_fmt bgra -vf scale={_width}:{_height} -an -";
        using var process = _processFactory();
        _activeVideo = process;
        process.Start(new ProcessStartInfo
        {
            FileName = ffmpegPath,
            Arguments = args,
            UseShellExecute = false,
            RedirectStandardOutput = true,
            RedirectStandardError = true,
            CreateNoWindow = true,
        });

        var reader = new RawVideoFrameReader(_width, _height);
        var buffer = new byte[65536];
        var stream = process.StandardOutput;
        var frames = 0;
        while (!ct.IsCancellationRequested && !process.HasExited)
        {
            var read = await stream.ReadAsync(buffer, 0, buffer.Length, ct).ConfigureAwait(false);
            if (read <= 0) break;
            var chunk = new byte[read];
            Buffer.BlockCopy(buffer, 0, chunk, 0, read);
            foreach (var frame in reader.Feed(chunk))
            {
                frames++;
                FrameAvailable?.Invoke(frame);
                if (frameDuration > TimeSpan.Zero)
                    await _delay(frameDuration, ct).ConfigureAwait(false);
            }
        }
        _activeVideo = null;
        return frames;
    }

    private async Task<int> RunAudioAsync(string ffmpegPath, CancellationToken ct)
    {
        var args =
            $"-hide_banner -loglevel error -i \"{_path}\" " +
            $"-f f32le -ac {AudioChannels} -ar {AudioSampleRate} -vn -";
        using var process = _processFactory();
        _activeAudio = process;
        process.Start(new ProcessStartInfo
        {
            FileName = ffmpegPath,
            Arguments = args,
            UseShellExecute = false,
            RedirectStandardOutput = true,
            RedirectStandardError = true,
            CreateNoWindow = true,
        });

        var stream = process.StandardOutput;
        var raw = new byte[65536];
        var carry = Array.Empty<byte>();
        var floatScratch = new float[AudioChunkFloats];
        var floatIndex = 0;
        var chunks = 0;
        // The pipe is byte-delimited only: a read may split a float, so leftover
        // bytes are carried into the next read instead of dropped (a dropped
        // half-float would desync the entire stream).
        while (!ct.IsCancellationRequested && !process.HasExited)
        {
            var read = await stream.ReadAsync(raw, 0, raw.Length, ct).ConfigureAwait(false);
            if (read <= 0) break;
            var total = carry.Length + read;
            var bytes = total <= raw.Length ? raw : new byte[total];
            if (carry.Length > 0)
            {
                Buffer.BlockCopy(carry, 0, bytes, 0, carry.Length);
                Buffer.BlockCopy(raw, 0, bytes, carry.Length, read);
            }
            var floatsTo = total - (total % 4);
            for (var i = 0; i < floatsTo && floatIndex < floatScratch.Length; i += 4)
                floatScratch[floatIndex++] = BitConverter.ToSingle(bytes, i);
            carry = total > floatsTo ? bytes[floatsTo..total] : Array.Empty<byte>();

            while (floatIndex == floatScratch.Length)
            {
                chunks++;
                AudioAvailable?.Invoke(new AudioSample(floatScratch, AudioSampleRate, AudioChannels));
                // Pace the NEXT chunk so the clip's audio lands at real-time cadence,
                // matching how the video pipe is paced by frame rate.
                await _delay(
                    TimeSpan.FromSeconds(AudioChunkFloats / (double)(AudioSampleRate * AudioChannels)), ct)
                    .ConfigureAwait(false);
                floatScratch = new float[AudioChunkFloats];
                floatIndex = 0;
            }
        }
        _activeAudio = null;
        return chunks;
    }

    private static void Kill(IDecodeProcess? process)
    {
        try { process?.Kill(); } catch { /* already exited */ }
    }
}