LlamaCasty/Services/MediaVideoSource.cs

using System.Diagnostics;
using System.IO;
using ytLive.Services.Encoder;

namespace ytLive.Services;

/// <summary>
/// TASK 21: decodes a local media file (any format ffmpeg handles) into
/// <see cref="VideoFrame"/>s for a MediaSource. Implements
/// <see cref="IMediaFrameSource"/>: raises <see cref="FrameAvailable"/> per
/// frame and <see cref="IMediaFrameSource.Completed"/> on natural EOF (unless
/// <see cref="IMediaFrameSource.Looping"/>). Spawns ffmpeg with
/// <c>-f rawvideo -pix_fmt bgra</c> and drains the raw BGRA pipe via
/// <see cref="RawVideoFrameReader"/>. The decoder process and emitters are seams
/// so tests run without a real codec or audio device.
/// </summary>
public sealed class MediaVideoSource : IMediaFrameSource
{
    public event Action<VideoFrame>? FrameAvailable;
    public event Action? Completed;

    public string Key => _path;

    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 CancellationTokenSource _cts = new();
    private IDisposable? _run;
    private IDecodeProcess? _active;

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

    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();
        _active?.Dispose();
    }

    private void StopInternal() => KillProcess();

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

            var ffmpegPath = await _locator.LocateAsync(_cts.Token).ConfigureAwait(false);
            var args =
                $"-hide_banner -loglevel error -i \"{_path}\" " +
                $"-f rawvideo -pix_fmt bgra -vf scale={_width}:{_height} -an -";
            var startInfo = new ProcessStartInfo
            {
                FileName = ffmpegPath,
                Arguments = args,
                UseShellExecute = false,
                RedirectStandardOutput = true,
                RedirectStandardError = true,
                CreateNoWindow = true,
            };

            do
            {
                using (var process = _processFactory())
                {
                    _active = process;
                    process.Start(startInfo);

                    var reader = new RawVideoFrameReader(_width, _height);
                    var buffer = new byte[65536];
                    var stream = process.StandardOutput;
                    while (!_cts.IsCancellationRequested && !process.HasExited)
                    {
                        var read = await stream.ReadAsync(buffer, 0, buffer.Length, _cts.Token).ConfigureAwait(false);
                        if (read <= 0) break;
                        foreach (var frame in reader.Feed(buffer.AsSpan(0, read).ToArray()))
                        {
                            FrameAvailable?.Invoke(frame);
                            if (frameDuration > TimeSpan.Zero)
                                await _delay(frameDuration, _cts.Token).ConfigureAwait(false);
                        }
                    }
                }
                _active = null;
            }
            while (Looping && !_cts.IsCancellationRequested);
        }
        catch (OperationCanceledException)
        {
        }
        catch (Exception ex)
        {
            Fail($"Media decode failed: {ex.Message}");
            return;
        }

        if (!_cts.IsCancellationRequested)
            Completed?.Invoke();
    }

    private void Fail(string message) => Console.Error.WriteLine(message);

    private void KillProcess()
    {
        try { _active?.Kill(); } catch { /* already exited */ }
    }
}