From a22fa0745cc192fcaad25df3b03d3d44dc5972b9 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 21 May 2026 18:27:39 +0100 Subject: [PATCH] Add native and process stream sources --- config/mimir-runtime.example.json | 38 +++++ docs/implementation-plan.md | 3 + docs/viable-stream-app.md | 9 ++ notes/current-system-map.md | 3 + notes/fresh-workspace-handoff.md | 4 +- src/Mimir.Runtime/MimirRuntime.cs | 7 +- .../MimirNativeIngestStreamSource.cs | 44 +++++ .../MimirProcessStreamSource.cs | 134 +++++++++++++++ .../MimirRuntimeConfiguration.cs | 153 ++++++++++++++++++ .../Synchronization/MimirStreamSample.cs | 3 +- state/evidence.jsonl | 1 + state/map.yaml | 8 + state/scratch.md | 3 + tests/test_mimir_app_contract.py | 6 + 14 files changed, 413 insertions(+), 3 deletions(-) create mode 100644 config/mimir-runtime.example.json create mode 100644 src/Mimir.Runtime/Synchronization/MimirNativeIngestStreamSource.cs create mode 100644 src/Mimir.Runtime/Synchronization/MimirProcessStreamSource.cs create mode 100644 src/Mimir.Runtime/Synchronization/MimirRuntimeConfiguration.cs diff --git a/config/mimir-runtime.example.json b/config/mimir-runtime.example.json new file mode 100644 index 0000000..6bcb3d3 --- /dev/null +++ b/config/mimir-runtime.example.json @@ -0,0 +1,38 @@ +{ + "bufferSeconds": 5.0, + "streams": [ + { + "sourceId": "kiyo-primary", + "kind": "Video", + "origin": "LocalDevice", + "enabled": true, + "adapter": "native" + }, + { + "sourceId": "host-focusrite", + "kind": "Audio", + "origin": "LocalDevice", + "enabled": true, + "adapter": "native" + }, + { + "sourceId": "neighbor-video-srt", + "kind": "Video", + "origin": "Network", + "enabled": false, + "adapter": "process", + "command": "ffmpeg", + "arguments": [ + "-hide_banner", + "-loglevel", + "warning", + "-i", + "srt://127.0.0.1:5100?mode=caller", + "-f", + "rawvideo", + "pipe:1" + ], + "chunkBytes": 65536 + } + ] +} diff --git a/docs/implementation-plan.md b/docs/implementation-plan.md index 9e1f389..2be4dd7 100644 --- a/docs/implementation-plan.md +++ b/docs/implementation-plan.md @@ -54,6 +54,9 @@ latency, and confidence. `MimirRollingStreamBuffer`, stream descriptors, and the `IMimirStreamSource` adapter seam. It initializes one bounded rolling buffer per configured audio/video stream and polls registered local/network sources each frame. + Local devices should feed through native push adapters; process-backed + sources are compatibility edges for network/diagnostic command ingest, not + the six-camera capture foundation. - `config/perfect-machine.example.json` declares the contract shape for six cameras, six microphones, reservoir typed views, native authorities, outputs, and the demotion of bridge scripts to tooling. - `native/reservoir` is the first native Rust crate. It implements one shared-edge five-second rolling buffer with typed views and proves that diff --git a/docs/viable-stream-app.md b/docs/viable-stream-app.md index 7493c83..1acdcf5 100644 --- a/docs/viable-stream-app.md +++ b/docs/viable-stream-app.md @@ -94,6 +94,10 @@ The first cut should be deliberately boring: - `MimirRollingStreamBuffer` owns one stream's short history and evicts samples outside the configured window. - `IMimirStreamSource` is the adapter seam for local devices and network feeds. +- `MimirNativeIngestStreamSource` is the low-overhead push seam for native + camera/audio adapters that already own buffers or handles. +- `MimirProcessStreamSource` is a compatibility adapter for network or + diagnostic command sources. It is not the six-camera local capture foundation. The default buffer duration is five seconds. It can be overridden with `MIMIR_SYNC_BUFFER_SECONDS` or `MIMIR_RESERVOIR_SECONDS`. @@ -109,6 +113,11 @@ environment variables: These create buffers immediately. Capture adapters still need to register `IMimirStreamSource` implementations to feed live samples into those buffers. +For local cameras, prefer native adapters that push sample handles directly +through `MimirNativeIngestStreamSource`. Process-backed FFmpeg ingest is allowed +for remote SRT/diagnostic edges, but local six-camera ingest should stay close +to the driver, GPU, and capture APIs. + ## Cut Line Do not add a new registry, bridge daemon, JSON live store, OBS plugin, or diff --git a/notes/current-system-map.md b/notes/current-system-map.md index 8362a47..2a8f03f 100644 --- a/notes/current-system-map.md +++ b/notes/current-system-map.md @@ -68,6 +68,9 @@ and D3D12 bridge, while `src/Mimir.Runtime` provides the Aquarium client runtime `Mimir.Runtime` owns the first synchronization hub: one five-second default rolling buffer per configured audio/video stream, source polling through `IMimirStreamSource`, and runtime UI readouts for stream/buffer health. +Local six-camera ingest should push native handles through +`MimirNativeIngestStreamSource`; process-backed ingest remains a compatibility +edge for network SRT or diagnostics. ```mermaid flowchart TD diff --git a/notes/fresh-workspace-handoff.md b/notes/fresh-workspace-handoff.md index de697e4..c93fc39 100644 --- a/notes/fresh-workspace-handoff.md +++ b/notes/fresh-workspace-handoff.md @@ -49,7 +49,9 @@ Get-Content .\state\evidence.jsonl -Tail 8 buffer per audio/video stream, defaulting to five seconds. Stream ids can be declared with `MIMIR_LOCAL_VIDEO_STREAMS`, `MIMIR_LOCAL_AUDIO_STREAMS`, `MIMIR_NETWORK_VIDEO_STREAMS`, and `MIMIR_NETWORK_AUDIO_STREAMS`; real capture - adapters feed buffers through `IMimirStreamSource`. + adapters feed buffers through `IMimirStreamSource`. Local six-camera ingest + should use native push adapters such as `MimirNativeIngestStreamSource`, not + FFmpeg stdout/process capture except as a network or diagnostic edge. - The reservoir is now one native time-ordered rolling buffer with typed indexes/views. The previous typed-ring crate is gone. - Native reservoir samples carry provenance flags; diagnostic/fallback samples diff --git a/src/Mimir.Runtime/MimirRuntime.cs b/src/Mimir.Runtime/MimirRuntime.cs index 8abcb4c..b5cb4c2 100644 --- a/src/Mimir.Runtime/MimirRuntime.cs +++ b/src/Mimir.Runtime/MimirRuntime.cs @@ -16,7 +16,12 @@ public sealed class MimirRuntime : IAquariumRuntime private int lastPollCount; public MimirRuntime(AquariumRuntimeOptions options) - : this(options, MimirSynchronizationSettings.FromEnvironment()) + : this(options, MimirRuntimeConfiguration.Load()) + { + } + + public MimirRuntime(AquariumRuntimeOptions options, MimirRuntimeConfiguration configuration) + : this(options, configuration.Settings, configuration.Sources) { } diff --git a/src/Mimir.Runtime/Synchronization/MimirNativeIngestStreamSource.cs b/src/Mimir.Runtime/Synchronization/MimirNativeIngestStreamSource.cs new file mode 100644 index 0000000..8982eac --- /dev/null +++ b/src/Mimir.Runtime/Synchronization/MimirNativeIngestStreamSource.cs @@ -0,0 +1,44 @@ +using System.Collections.Concurrent; + +namespace Mimir.Runtime.Synchronization; + +public sealed class MimirNativeIngestStreamSource : IMimirStreamSource +{ + private readonly ConcurrentQueue samples = new(); + private ulong sequence; + + public MimirNativeIngestStreamSource(MimirStreamDescriptor descriptor) + { + Descriptor = descriptor; + } + + public MimirStreamDescriptor Descriptor { get; } + + public void Push( + long timestampNs, + long arrivalNs, + ulong payloadHandle, + int byteLength = 0, + ReadOnlyMemory data = default) + { + samples.Enqueue(new MimirStreamSample( + Descriptor.SourceId, + Descriptor.Kind, + Descriptor.Origin, + timestampNs, + arrivalNs, + sequence++, + payloadHandle, + byteLength, + data)); + } + + public bool TryRead(out MimirStreamSample sample) + { + return samples.TryDequeue(out sample); + } + + public void Dispose() + { + } +} diff --git a/src/Mimir.Runtime/Synchronization/MimirProcessStreamSource.cs b/src/Mimir.Runtime/Synchronization/MimirProcessStreamSource.cs new file mode 100644 index 0000000..c5144e6 --- /dev/null +++ b/src/Mimir.Runtime/Synchronization/MimirProcessStreamSource.cs @@ -0,0 +1,134 @@ +using System.Collections.Concurrent; +using System.Diagnostics; + +namespace Mimir.Runtime.Synchronization; + +public sealed record MimirProcessStreamSourceOptions( + string Command, + IReadOnlyList Arguments, + int ChunkBytes = 65_536); + +public sealed class MimirProcessStreamSource : IMimirStreamSource +{ + private readonly ConcurrentQueue samples = new(); + private readonly CancellationTokenSource cancellation = new(); + private readonly Process process; + private readonly Task stdoutTask; + private readonly Task stderrTask; + private ulong sequence; + private bool disposed; + + public MimirProcessStreamSource( + MimirStreamDescriptor descriptor, + MimirProcessStreamSourceOptions options) + { + Descriptor = descriptor; + Options = options; + process = StartProcess(options); + stdoutTask = Task.Run(() => ReadStdout(options.ChunkBytes, cancellation.Token)); + stderrTask = Task.Run(() => DrainStderr(cancellation.Token)); + } + + public MimirStreamDescriptor Descriptor { get; } + + public MimirProcessStreamSourceOptions Options { get; } + + public bool TryRead(out MimirStreamSample sample) + { + return samples.TryDequeue(out sample); + } + + private Process StartProcess(MimirProcessStreamSourceOptions options) + { + var startInfo = new ProcessStartInfo(options.Command) + { + UseShellExecute = false, + RedirectStandardOutput = true, + RedirectStandardError = true, + CreateNoWindow = true, + }; + + foreach (var argument in options.Arguments) + { + startInfo.ArgumentList.Add(argument); + } + + return Process.Start(startInfo) + ?? throw new InvalidOperationException($"Could not start stream source process: {options.Command}"); + } + + private async Task ReadStdout(int chunkBytes, CancellationToken token) + { + var stream = process.StandardOutput.BaseStream; + while (!token.IsCancellationRequested) + { + var buffer = new byte[chunkBytes]; + var read = await stream.ReadAsync(buffer.AsMemory(0, buffer.Length), token).ConfigureAwait(false); + if (read <= 0) + { + break; + } + + if (read != buffer.Length) + { + Array.Resize(ref buffer, read); + } + + var now = Stopwatch.GetTimestamp(); + var timestampNs = StopwatchTicksToNs(now); + samples.Enqueue(new MimirStreamSample( + Descriptor.SourceId, + Descriptor.Kind, + Descriptor.Origin, + timestampNs, + timestampNs, + sequence++, + 0, + buffer.Length, + buffer)); + } + } + + private async Task DrainStderr(CancellationToken token) + { + while (!token.IsCancellationRequested) + { + var line = await process.StandardError.ReadLineAsync(token).ConfigureAwait(false); + if (line == null) + { + break; + } + } + } + + private static long StopwatchTicksToNs(long ticks) + { + return checked((long)(ticks * (1_000_000_000.0 / Stopwatch.Frequency))); + } + + public void Dispose() + { + if (disposed) + { + return; + } + + disposed = true; + cancellation.Cancel(); + try + { + if (!process.HasExited) + { + process.Kill(entireProcessTree: true); + } + } + catch + { + } + + process.Dispose(); + cancellation.Dispose(); + GC.SuppressFinalize(stdoutTask); + GC.SuppressFinalize(stderrTask); + } +} diff --git a/src/Mimir.Runtime/Synchronization/MimirRuntimeConfiguration.cs b/src/Mimir.Runtime/Synchronization/MimirRuntimeConfiguration.cs new file mode 100644 index 0000000..58019be --- /dev/null +++ b/src/Mimir.Runtime/Synchronization/MimirRuntimeConfiguration.cs @@ -0,0 +1,153 @@ +using System.Text.Json; + +namespace Mimir.Runtime.Synchronization; + +public sealed class MimirRuntimeConfiguration +{ + public MimirSynchronizationSettings Settings { get; init; } = new(); + + public IReadOnlyList Sources { get; init; } = []; + + public static MimirRuntimeConfiguration Load() + { + var path = ResolveConfigPath(); + if (path == null) + { + return new MimirRuntimeConfiguration + { + Settings = MimirSynchronizationSettings.FromEnvironment(), + }; + } + + var model = JsonSerializer.Deserialize( + File.ReadAllText(path), + new JsonSerializerOptions + { + PropertyNameCaseInsensitive = true, + }) ?? new MimirRuntimeConfigFile(); + + var bufferDuration = model.BufferSeconds > 0.0 + ? TimeSpan.FromSeconds(Math.Clamp(model.BufferSeconds, 0.25, 60.0)) + : MimirSynchronizationSettings.FromEnvironment().BufferDuration; + + var sourceModels = model.Streams + .Where(stream => stream.Enabled) + .ToArray(); + var streams = sourceModels + .Select(ToDescriptor) + .ToArray(); + var sources = sourceModels + .Select(TryCreateSource) + .Where(source => source != null) + .Cast() + .ToArray(); + + return new MimirRuntimeConfiguration + { + Settings = new MimirSynchronizationSettings + { + BufferDuration = bufferDuration, + Streams = streams, + }, + Sources = sources, + }; + } + + private static string? ResolveConfigPath() + { + var configured = Environment.GetEnvironmentVariable("MIMIR_RUNTIME_CONFIG"); + if (!string.IsNullOrWhiteSpace(configured) && File.Exists(configured)) + { + return configured; + } + + var current = AppContext.BaseDirectory; + for (var index = 0; index < 8; index++) + { + var candidate = Path.Combine(current, "config", "mimir-runtime.json"); + if (File.Exists(candidate)) + { + return candidate; + } + + var parent = Directory.GetParent(current); + if (parent == null) + { + break; + } + + current = parent.FullName; + } + + return null; + } + + private static MimirStreamDescriptor ToDescriptor(MimirStreamConfig stream) + { + return new MimirStreamDescriptor( + stream.SourceId, + ParseKind(stream.Kind), + ParseOrigin(stream.Origin), + stream.Enabled); + } + + private static IMimirStreamSource? TryCreateSource(MimirStreamConfig stream) + { + if (!string.Equals(stream.Adapter, "process", StringComparison.OrdinalIgnoreCase)) + { + return null; + } + + if (string.IsNullOrWhiteSpace(stream.Command)) + { + return null; + } + + return new MimirProcessStreamSource( + ToDescriptor(stream), + new MimirProcessStreamSourceOptions( + stream.Command, + stream.Arguments, + Math.Max(1024, stream.ChunkBytes))); + } + + private static MimirStreamKind ParseKind(string value) + { + return Enum.TryParse(value, ignoreCase: true, out var kind) + ? kind + : throw new InvalidOperationException($"Unknown Mimir stream kind: {value}"); + } + + private static MimirStreamOrigin ParseOrigin(string value) + { + return Enum.TryParse(value, ignoreCase: true, out var origin) + ? origin + : throw new InvalidOperationException($"Unknown Mimir stream origin: {value}"); + } +} + +public sealed class MimirRuntimeConfigFile +{ + public double BufferSeconds { get; set; } = 5.0; + + public List Streams { get; set; } = []; +} + +public sealed class MimirStreamConfig +{ + public string SourceId { get; set; } = ""; + + public string Kind { get; set; } = "Video"; + + public string Origin { get; set; } = "LocalDevice"; + + public bool Enabled { get; set; } = true; + + public string Adapter { get; set; } = "process"; + + public string Command { get; set; } = ""; + + public string[] Arguments { get; set; } = []; + + public int ChunkBytes { get; set; } = 65_536; +} diff --git a/src/Mimir.Runtime/Synchronization/MimirStreamSample.cs b/src/Mimir.Runtime/Synchronization/MimirStreamSample.cs index 48ceef7..a59f745 100644 --- a/src/Mimir.Runtime/Synchronization/MimirStreamSample.cs +++ b/src/Mimir.Runtime/Synchronization/MimirStreamSample.cs @@ -8,4 +8,5 @@ public readonly record struct MimirStreamSample( long ArrivalNs, ulong Sequence, ulong PayloadHandle, - int ByteLength = 0); + int ByteLength = 0, + ReadOnlyMemory Data = default); diff --git a/state/evidence.jsonl b/state/evidence.jsonl index c6c6e5b..2a817ff 100644 --- a/state/evidence.jsonl +++ b/state/evidence.jsonl @@ -98,3 +98,4 @@ {"ts": "2026-05-21T12:09:51+00:00", "type": "decision", "status": "ok", "note": "Viable stream app cut pinned: Aquarium hosts one five-second in-memory LocalcastRuntime, local/networked feeds append as producers, Aquarium owns runtime UI/debug/settings/output management, Faust/native audio builds synchronized stems, and OBS receives program outputs rather than raw feed sync."} {"ts": "2026-05-21T12:14:26+00:00", "type": "implementation", "status": "ok", "note": "Mimir now owns a C# app solution: Mimir.slnx contains src/Mimir.App, which launches AquariumHost from Aquarium Engine, and src/Mimir.Runtime, which provides the Aquarium runtime factory over Aquarium.LocalCast. Aquarium Engine remains the windowing/rendering/D3D12 library reference."} {"ts": "2026-05-21T12:33:01+00:00", "type": "implementation", "status": "ok", "note": "Mimir.Runtime now owns the first synchronization spine: MimirRuntime wraps Aquarium LocalCast rendering, initializes per-stream rolling buffers through MimirSynchronizationHub, polls registered IMimirStreamSource adapters each update, and defaults buffer duration to five seconds with env overrides for local/network audio/video stream ids."} +{"ts": "2026-05-21T17:27:12+00:00", "type": "decision", "status": "ok", "note": "For six local cameras, Mimir ingest should stay close to driver/GPU capture surfaces. MimirNativeIngestStreamSource is the preferred local adapter seam carrying payload handles and optional memory into rolling buffers; MimirProcessStreamSource remains a network/diagnostic compatibility edge, not the local camera foundation."} diff --git a/state/map.yaml b/state/map.yaml index 08afb39..68882b6 100644 --- a/state/map.yaml +++ b/state/map.yaml @@ -76,6 +76,11 @@ accepted_design: and `IMimirStreamSource`. It initializes one bounded rolling buffer per configured audio/video stream and drains registered local/network sources into those buffers each Aquarium update. + - Local six-camera ingest should use native push sources such as + `MimirNativeIngestStreamSource`, carrying native payload handles and + optional memory spans into Mimir buffers. `MimirProcessStreamSource` is a + compatibility path for network SRT or diagnostics, not the local camera + foundation. - `native/reservoir` is the first native Rust crate. It now uses one time-ordered rolling buffer with typed views, so retention has exactly one owner while Aquarium/Faust still get kind-scoped queries. @@ -119,6 +124,9 @@ active_subgoals: expose runtime UI, and emit synchronized OBS program outputs. - Implement concrete local device and network `IMimirStreamSource` adapters that feed the Mimir runtime buffers with live audio/video samples. + - Build the local camera adapter close to the driver/GPU capture surface; + avoid FFmpeg stdout for the six local camera streams unless used only as a + temporary diagnostic witness. - Bind Aquarium/Faust to the rolling-buffer `LocalcastRuntime` and `LocalcastProducer` C ABI described by `native/reservoir/include/localcast_reservoir.h`. diff --git a/state/scratch.md b/state/scratch.md index d80f379..5b5805a 100644 --- a/state/scratch.md +++ b/state/scratch.md @@ -14,5 +14,8 @@ No active scratch subgoal. - Synchronization ownership started moving into `Mimir.Runtime`: hub, stream descriptors, per-stream rolling buffers, settings, and source adapter seam are in C# now; concrete capture adapters are next. +- Local six-camera ingest should stay close to the metal: native push sources + carry payload handles into buffers; process sources are network/diagnostic + compatibility, not the main local camera path. - Keep scratch short and disposable. - Promote only durable lessons into `state/evidence.jsonl` or the map. diff --git a/tests/test_mimir_app_contract.py b/tests/test_mimir_app_contract.py index 7bcb511..cf861fb 100644 --- a/tests/test_mimir_app_contract.py +++ b/tests/test_mimir_app_contract.py @@ -47,6 +47,12 @@ class MimirAppContractTests(unittest.TestCase): self.assertIn("TimeSpan.FromSeconds(5)", settings) self.assertIn("MIMIR_NETWORK_VIDEO_STREAMS", settings) self.assertIn("MIMIR_LOCAL_AUDIO_STREAMS", settings) + native_source = (ROOT / "src" / "Mimir.Runtime" / "Synchronization" / "MimirNativeIngestStreamSource.cs").read_text(encoding="utf-8") + process_source = (ROOT / "src" / "Mimir.Runtime" / "Synchronization" / "MimirProcessStreamSource.cs").read_text(encoding="utf-8") + + self.assertIn("payloadHandle", native_source) + self.assertIn("ReadOnlyMemory", native_source) + self.assertIn("RedirectStandardOutput", process_source) if __name__ == "__main__":