gramps
97ffc426fe
Take-two forensics: rawvideo carries no per-frame timestamps — ffmpeg stamps frames by ARRIVAL at the declared fps, so a 2fps producer yields a 30x time-lapse, 0.6s-long '60fps' file with zero errors anywhere (explains the creator's accelerated webcam motion + truncated length observations). The pump now logs frames-produced-vs-target per 5s with the avg render/submit split — take three names the slow stage instead of us theorizing. FramePumpTests 9/9. (Multi-commit working unit: remaining declared files land in the next commit.)
365 lines
14 KiB
C#
365 lines
14 KiB
C#
using System;
|
||
using System.Threading;
|
||
using System.Threading.Tasks;
|
||
using ytLive.Models;
|
||
using ytLive.Services.Compositor;
|
||
|
||
namespace ytLive.Services.Encoder;
|
||
|
||
/// <summary>
|
||
/// The live frame producer (TASK 4 ship step 5): the bridge between the capture
|
||
/// managers + compositor and the encoder. While live it snapshots the active
|
||
/// scene each tick, resolves every element to its latest frame, composites it
|
||
/// into the tier's output frame, and paces frames into the encoder at the tier's
|
||
/// FPS. All collaborators are constructor-injected seams (scene, resolver,
|
||
/// options, encoder factory, pacing delay) so the pump stays free of WPF and of
|
||
/// the capture managers and is fully hermetic in tests.
|
||
///
|
||
/// The RTMP URL comes from the options provider: until the live-stream create
|
||
/// flow lands (TASK 5) it yields null, so go-live runs the existing visual flow
|
||
/// without actually pushing.
|
||
/// </summary>
|
||
public sealed class FramePump : IDisposable
|
||
{
|
||
private readonly Func<Scene?> _sceneProvider;
|
||
private readonly Func<SceneElement, VideoFrame?> _frameResolver;
|
||
private readonly Func<CompositorOptions> _compositorOptions;
|
||
private readonly Func<EncoderOptions?> _encoderOptions;
|
||
private readonly Func<IFfmpegEncoder> _encoderFactory;
|
||
private readonly Action<string>? _log;
|
||
private readonly Func<TimeSpan, CancellationToken, Task> _pacingDelay;
|
||
private readonly Func<(VideoFrame? Frame, SocialBarPosition Position)>? _socialBar;
|
||
private readonly TransitionService? _transition;
|
||
private readonly Func<Scene?>? _fromSceneProvider;
|
||
private readonly SceneGraph? _sceneGraph;
|
||
private readonly SceneCompositor _compositor = new();
|
||
|
||
private readonly object _gate = new();
|
||
private IFfmpegEncoder? _encoder;
|
||
private CancellationTokenSource? _cts;
|
||
private Task? _pumpTask;
|
||
private bool _started;
|
||
|
||
/// <summary>Forwards the encoder's parsed health — ship step 6 binds this to the bottom bar.</summary>
|
||
public event EventHandler<StreamHealth>? HealthUpdated;
|
||
|
||
/// <summary>Raised when the encoder cannot start or dies mid-stream. The pump stops itself.</summary>
|
||
public event EventHandler<string>? Failed;
|
||
|
||
public FramePump(
|
||
Func<Scene?> sceneProvider,
|
||
Func<SceneElement, VideoFrame?> frameResolver,
|
||
Func<CompositorOptions> compositorOptions,
|
||
Func<EncoderOptions?> encoderOptions,
|
||
Func<IFfmpegEncoder> encoderFactory,
|
||
Action<string>? log = null,
|
||
Func<TimeSpan, CancellationToken, Task>? pacingDelay = null,
|
||
Func<(VideoFrame? Frame, SocialBarPosition Position)>? socialBar = null,
|
||
TransitionService? transition = null,
|
||
Func<Scene?>? fromSceneProvider = null,
|
||
SceneGraph? sceneGraph = null)
|
||
{
|
||
_sceneProvider = sceneProvider ?? throw new ArgumentNullException(nameof(sceneProvider));
|
||
_frameResolver = frameResolver ?? throw new ArgumentNullException(nameof(frameResolver));
|
||
_compositorOptions = compositorOptions ?? throw new ArgumentNullException(nameof(compositorOptions));
|
||
_encoderOptions = encoderOptions ?? throw new ArgumentNullException(nameof(encoderOptions));
|
||
_encoderFactory = encoderFactory ?? throw new ArgumentNullException(nameof(encoderFactory));
|
||
_log = log;
|
||
_pacingDelay = pacingDelay ?? ((delay, ct) => Task.Delay(delay, ct));
|
||
_socialBar = socialBar;
|
||
_transition = transition;
|
||
_fromSceneProvider = fromSceneProvider;
|
||
_sceneGraph = sceneGraph;
|
||
}
|
||
|
||
public bool IsRunning { get; private set; }
|
||
|
||
/// <summary>Never throws: failures are logged and surfaced via <see cref="Failed"/>,
|
||
/// so the VM can fire-and-forget it from a sync command handler.</summary>
|
||
public async Task StartAsync(CancellationToken cancellationToken = default)
|
||
{
|
||
lock (_gate)
|
||
{
|
||
if (_started) return;
|
||
_started = true;
|
||
}
|
||
|
||
IFfmpegEncoder? encoder = null;
|
||
try
|
||
{
|
||
var options = _encoderOptions();
|
||
if (options == null)
|
||
{
|
||
_log?.Invoke("FramePump: no output configured (neither streaming nor recording) — encoder skipped");
|
||
lock (_gate) _started = false;
|
||
return;
|
||
}
|
||
|
||
encoder = _encoderFactory();
|
||
encoder.HealthUpdated += OnHealthUpdated;
|
||
encoder.ProcessFailed += OnProcessFailed;
|
||
await encoder.StartAsync(options, cancellationToken);
|
||
|
||
lock (_gate)
|
||
{
|
||
_encoder = encoder;
|
||
}
|
||
|
||
// IsRunning must be true before the loop starts: the loop reads it on
|
||
// its first iteration, and with a completed-task delay it can run
|
||
// synchronously on this thread before PumpAsync even returns.
|
||
IsRunning = true;
|
||
_cts = new CancellationTokenSource();
|
||
_pumpTask = PumpAsync(options, _cts.Token);
|
||
_log?.Invoke($"FramePump started ({options.Width}×{options.Height} @ {options.Fps} fps)");
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_log?.Invoke($"FramePump: start failed: {ex.Message}");
|
||
if (encoder != null)
|
||
{
|
||
encoder.HealthUpdated -= OnHealthUpdated;
|
||
encoder.ProcessFailed -= OnProcessFailed;
|
||
try
|
||
{
|
||
encoder.Dispose();
|
||
}
|
||
catch (Exception disposeEx)
|
||
{
|
||
_log?.Invoke($"FramePump: disposing failed encoder: {disposeEx.Message}");
|
||
}
|
||
}
|
||
lock (_gate)
|
||
{
|
||
_started = false;
|
||
IsRunning = false;
|
||
}
|
||
Failed?.Invoke(this, ex.Message);
|
||
}
|
||
}
|
||
|
||
public async Task StopAsync(CancellationToken cancellationToken = default)
|
||
{
|
||
IFfmpegEncoder? encoder;
|
||
Task? pump;
|
||
lock (_gate)
|
||
{
|
||
if (!_started && _encoder == null) return;
|
||
_started = false;
|
||
IsRunning = false;
|
||
encoder = _encoder;
|
||
pump = _pumpTask;
|
||
_cts?.Cancel();
|
||
}
|
||
|
||
// Stop the encoder BEFORE awaiting the pump: closing its stdin unblocks a
|
||
// write stuck on pipe backpressure, otherwise the pump could await forever.
|
||
if (encoder != null)
|
||
{
|
||
try
|
||
{
|
||
await encoder.StopAsync(cancellationToken);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_log?.Invoke($"FramePump: encoder stop failed: {ex.Message}");
|
||
}
|
||
}
|
||
|
||
if (pump != null)
|
||
{
|
||
try
|
||
{
|
||
await pump;
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_log?.Invoke($"FramePump: pump loop faulted during stop: {ex.Message}");
|
||
}
|
||
}
|
||
|
||
if (encoder != null)
|
||
{
|
||
encoder.HealthUpdated -= OnHealthUpdated;
|
||
encoder.ProcessFailed -= OnProcessFailed;
|
||
try
|
||
{
|
||
encoder.Dispose();
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_log?.Invoke($"FramePump: encoder dispose failed: {ex.Message}");
|
||
}
|
||
}
|
||
|
||
lock (_gate)
|
||
{
|
||
_encoder = null;
|
||
_cts = null;
|
||
_pumpTask = null;
|
||
}
|
||
_log?.Invoke("FramePump stopped");
|
||
}
|
||
|
||
public void Dispose()
|
||
{
|
||
try
|
||
{
|
||
StopAsync().GetAwaiter().GetResult();
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_log?.Invoke($"FramePump: dispose failed: {ex.Message}");
|
||
}
|
||
}
|
||
|
||
private async Task PumpAsync(EncoderOptions options, CancellationToken ct)
|
||
{
|
||
var interval = TimeSpan.FromSeconds(1d / Math.Max(1, options.Fps));
|
||
var lastTick = System.Diagnostics.Stopwatch.StartNew();
|
||
|
||
// Stage timing (2026-09-01, take two): rawvideo carries no per-frame
|
||
// timestamps — ffmpeg stamps frames by ARRIVAL at the declared fps. A producer
|
||
// slower than the declared rate yields a time-lapsed, short file (observed:
|
||
// 39 frames in 27 wall-seconds ≈ 30x at 60fps) with no error anywhere.
|
||
// Log the render/submit split every 5s so the next take names the stage.
|
||
var renderSw = new System.Diagnostics.Stopwatch();
|
||
var submitSw = new System.Diagnostics.Stopwatch();
|
||
long renderTicks = 0, submitTicks = 0;
|
||
int statFrames = 0;
|
||
var statsNext = DateTime.UtcNow + TimeSpan.FromSeconds(5);
|
||
void ReportStats()
|
||
{
|
||
if (DateTime.UtcNow < statsNext) return;
|
||
var target = 5d / interval.TotalSeconds; // frames expected per window
|
||
_log?.Invoke(statFrames == 0
|
||
? "FramePump stats: NO frames produced in 5s (loop stalled?)"
|
||
: $"FramePump stats: {statFrames}/{target:F0} frames per 5s, " +
|
||
$"avg render {renderTicks / (double)System.Diagnostics.Stopwatch.Frequency * 1000 / statFrames:F1}ms, " +
|
||
$"avg submit {submitTicks / (double)System.Diagnostics.Stopwatch.Frequency * 1000 / statFrames:F1}ms");
|
||
renderTicks = submitTicks = 0;
|
||
statFrames = 0;
|
||
statsNext = DateTime.UtcNow + TimeSpan.FromSeconds(5);
|
||
}
|
||
|
||
try
|
||
{
|
||
while (!ct.IsCancellationRequested)
|
||
{
|
||
var scene = _sceneProvider();
|
||
if (scene != null)
|
||
{
|
||
var compositorOptions = _compositorOptions();
|
||
VideoFrame? socialBarFrame = null;
|
||
var socialBarTop = 0;
|
||
if (_socialBar != null)
|
||
{
|
||
var (barFrame, position) = _socialBar();
|
||
socialBarFrame = barFrame;
|
||
if (barFrame != null)
|
||
socialBarTop = position == SocialBarPosition.Top
|
||
? 0
|
||
: compositorOptions.SourceRectHeight - barFrame.Height;
|
||
}
|
||
|
||
VideoFrame frame;
|
||
renderSw.Restart();
|
||
if (_transition is { Active: true } transition && _fromSceneProvider != null)
|
||
{
|
||
var fromScene = _fromSceneProvider();
|
||
var fromFrame = fromScene != null
|
||
? RenderScene(fromScene, compositorOptions, socialBarFrame, socialBarTop)
|
||
: RenderScene(scene, compositorOptions, socialBarFrame, socialBarTop);
|
||
var toFrame = RenderScene(scene, compositorOptions, socialBarFrame, socialBarTop);
|
||
frame = transition.BlendFrame(toFrame);
|
||
transition.Tick(lastTick.Elapsed.TotalMilliseconds);
|
||
lastTick.Restart();
|
||
}
|
||
else
|
||
{
|
||
frame = RenderScene(scene, compositorOptions, socialBarFrame, socialBarTop);
|
||
lastTick.Restart();
|
||
}
|
||
renderSw.Stop();
|
||
renderTicks += renderSw.ElapsedTicks;
|
||
|
||
IFfmpegEncoder? encoder;
|
||
lock (_gate) encoder = _encoder;
|
||
if (encoder == null) break;
|
||
submitSw.Restart();
|
||
await encoder.SubmitFrameAsync(frame, ct);
|
||
submitSw.Stop();
|
||
submitTicks += submitSw.ElapsedTicks;
|
||
statFrames++;
|
||
ReportStats();
|
||
}
|
||
await _pacingDelay(interval, ct);
|
||
}
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
// normal stop
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
// A failure while the pump is supposed to run (encoder died under us,
|
||
// scene provider faulted, ...) stops the pump and surfaces once.
|
||
if (ct.IsCancellationRequested)
|
||
{
|
||
_log?.Invoke($"FramePump: pump exited during stop: {ex.Message}");
|
||
}
|
||
else
|
||
{
|
||
_log?.Invoke($"FramePump: pump loop faulted: {ex.Message}");
|
||
Failed?.Invoke(this, ex.Message);
|
||
}
|
||
}
|
||
finally
|
||
{
|
||
lock (_gate) IsRunning = false;
|
||
}
|
||
}
|
||
|
||
private void OnHealthUpdated(object? sender, StreamHealth health) => HealthUpdated?.Invoke(this, health);
|
||
|
||
/// <summary>Render a scene, using the baked-crust optimization when a <see cref="SceneGraph"/>
|
||
/// is wired in: bake/cache the static layers below the split point, then composite the
|
||
/// dynamic/above-split layers per frame. Without a SceneGraph, falls back to a full render
|
||
/// (identical output — see SceneCompositorTests).</summary>
|
||
private VideoFrame RenderScene(
|
||
Scene scene,
|
||
CompositorOptions options,
|
||
VideoFrame? socialBarFrame,
|
||
int socialBarTop)
|
||
{
|
||
if (_sceneGraph == null)
|
||
return _compositor.Render(scene, _frameResolver, null, options, socialBarFrame, socialBarTop);
|
||
|
||
var split = _sceneGraph.GetSplitPoint(scene);
|
||
if (split == scene.Elements.Count)
|
||
{
|
||
// Fully static scene: bake once, reuse.
|
||
var baked = _sceneGraph.GetBakedBase(scene, _frameResolver, _compositorOptions);
|
||
if (baked != null)
|
||
return StretchMath.BilinearScale(baked, options.OutputWidth, options.OutputHeight);
|
||
}
|
||
|
||
var baseFrame = _sceneGraph.GetBakedBase(scene, _frameResolver, _compositorOptions);
|
||
if (baseFrame != null)
|
||
{
|
||
return SceneCompositor.CompositeLayers(
|
||
baseFrame, scene, split, _frameResolver, options, socialBarFrame, socialBarTop);
|
||
}
|
||
|
||
// No static base (first layer is dynamic or empty scene) — full render.
|
||
return _compositor.Render(scene, _frameResolver, null, options, socialBarFrame, socialBarTop);
|
||
}
|
||
|
||
private void OnProcessFailed(object? sender, string message)
|
||
{
|
||
_log?.Invoke($"FramePump: encoder process failed: {message}");
|
||
Failed?.Invoke(this, message);
|
||
_ = StopAsync();
|
||
}
|
||
}
|