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;
///
/// 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 (rawvideo bgra, scaled to the target rect) and audio
/// (f32le -ac 2 -ar 48000, 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.
/// Looping is the media case: on natural EOF the whole pass restarts — a fresh
/// pair of processes, because a cannot be re-started. An alert clip
/// never loops, so its wrapper simply never sets .
/// 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.
///
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? FrameAvailable;
public event Action? AudioAvailable;
public event Action? Completed;
/// Restart the pass on natural EOF (media/BRB loops).
public bool Looping { get; set; }
private readonly string _path;
private readonly int _width;
private readonly int _height;
private readonly IFfmpegLocator _locator;
private readonly Func _processFactory;
private readonly IFrameRateProbe? _probe;
private readonly Func _delay;
private readonly bool _audio;
private readonly ChromaKeyOptions? _key;
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 processFactory,
IFrameRateProbe? frameRateProbe = null,
Func? delay = null,
bool audio = true,
ChromaKeyOptions? chromaKey = null)
{
_path = path;
_width = width;
_height = height;
_locator = locator;
_processFactory = processFactory;
_probe = frameRateProbe;
_delay = delay ?? ((t, ct) => Task.Delay(t, ct));
_audio = audio;
_key = chromaKey;
}
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 RunVideoAsync(string ffmpegPath, TimeSpan frameDuration, CancellationToken ct)
{
var args =
$"-hide_banner -loglevel error -i \"{_path}\" " +
$"-f rawvideo -pix_fmt bgra -vf {VideoFilterChain.BuildVideo(_width, _height, _key)} -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 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();
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();
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 */ }
}
}