diff --git a/.gitignore b/.gitignore index a2d7ca1..5479e54 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,5 @@ calibration/runs/ tools/drivers/ tools/obs-spout2-plugin/ tools/src/ +target/ +native/**/target/ diff --git a/docs/implementation-plan.md b/docs/implementation-plan.md index c3d6336..6fae140 100644 --- a/docs/implementation-plan.md +++ b/docs/implementation-plan.md @@ -5,6 +5,7 @@ - Perfect Machine contract: - `docs/perfect-machine.md` defines the target native machine: one five-second spatiotemporal reservoir, Aquarium-owned dense visual fusion/material/brush rendering, Faust-owned hot audio DSP, OBS-owned broadcast controls, and LocalCastBridge-owned calibration/config/status. - `config/perfect-machine.example.json` declares the contract shape for six cameras, six microphones, reservoir rings, native authorities, outputs, and the demotion of bridge scripts to tooling. + - `native/reservoir` is the first native Rust crate. It implements shared-edge five-second reservoir rings and proves that all rings expire against the newest sample, not private per-source clocks. - Repo-local persistence machinery. - First architecture map. - Example config for one video source plus two audio sources. @@ -74,7 +75,7 @@ ## Next -1. Implement the native reservoir ABI declared by `config/perfect-machine.example.json`. +1. Add a C ABI and Aquarium-facing binding around `native/reservoir`. 2. Move camera ingest into native capture workers that append `camera_frame` samples to the reservoir. 3. Move feature extraction, flow, cross-view matching, LOD reconciliation, and material fitting into Aquarium GPU compute. 4. Move mic alignment, voice separation, room suppression, Ambisonic/HOA spatialization, and stem generation into Faust/native DSP. diff --git a/docs/perfect-machine.md b/docs/perfect-machine.md index a768302..ac334ae 100644 --- a/docs/perfect-machine.md +++ b/docs/perfect-machine.md @@ -147,3 +147,16 @@ Keep these as tooling: 6. Let Faust consume reservoir audio blocks and source controls directly. 7. Keep the old bridge only as a comparison and fallback surface. +## Native Work Started + +`native/reservoir` is the first native crate. It implements the shared-edge +five-second reservoir invariant: + +- append samples into named rings; +- advance the reservoir edge from the newest sample; +- evict every ring against the same edge; +- query the latest valid sample per sensor. + +This is intentionally below Aquarium and Faust. The crate owns retention and +sample-window semantics; Aquarium owns GPU interpretation and rendering, Faust +owns DSP interpretation. diff --git a/native/README.md b/native/README.md new file mode 100644 index 0000000..5bab682 --- /dev/null +++ b/native/README.md @@ -0,0 +1,13 @@ +# Native Runtime + +This directory is for the hot path. + +LocalCastBridge Python remains useful for calibration, launch, status, contract +tests, and offline analysis. It does not own dense live fusion or DSP. + +## Crates + +- `reservoir`: the first native five-second spatiotemporal reservoir core. It + owns shared-edge retention and typed sample rings. Aquarium/Faust integration + should build on this invariant instead of copying the deadline bridge cache. + diff --git a/native/reservoir/Cargo.lock b/native/reservoir/Cargo.lock new file mode 100644 index 0000000..8ce6767 --- /dev/null +++ b/native/reservoir/Cargo.lock @@ -0,0 +1,7 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "localcast-reservoir" +version = "0.1.0" diff --git a/native/reservoir/Cargo.toml b/native/reservoir/Cargo.toml new file mode 100644 index 0000000..13cd157 --- /dev/null +++ b/native/reservoir/Cargo.toml @@ -0,0 +1,10 @@ +[package] +name = "localcast-reservoir" +version = "0.1.0" +edition = "2024" +description = "Native five-second spatiotemporal reservoir core for LocalCastBridge" + +[lib] +crate-type = ["rlib"] + +[dependencies] diff --git a/native/reservoir/src/lib.rs b/native/reservoir/src/lib.rs new file mode 100644 index 0000000..2a7b204 --- /dev/null +++ b/native/reservoir/src/lib.rs @@ -0,0 +1,211 @@ +use std::collections::{BTreeMap, VecDeque}; + +pub const DEFAULT_RESERVOIR_NS: u64 = 5_000_000_000; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub enum RingKind { + CameraFrame, + CameraFeature, + SceneRay, + SurfaceClaim, + MaterialClaim, + AudioBlock, + PhaseClaim, + EventClaim, + RenderPacket, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct Sample { + pub sensor_id: String, + pub timestamp_ns: u64, + pub arrival_ns: u64, + pub sequence: u64, + pub payload: T, +} + +impl Sample { + pub fn new( + sensor_id: impl Into, + timestamp_ns: u64, + arrival_ns: u64, + sequence: u64, + payload: T, + ) -> Self { + Self { + sensor_id: sensor_id.into(), + timestamp_ns, + arrival_ns, + sequence, + payload, + } + } +} + +#[derive(Debug, Clone)] +pub struct ReservoirRing { + duration_ns: u64, + edge_ns: u64, + samples: VecDeque>, +} + +impl ReservoirRing { + pub fn new(duration_ns: u64) -> Self { + Self { + duration_ns, + edge_ns: 0, + samples: VecDeque::new(), + } + } + + pub fn push(&mut self, sample: Sample) { + self.edge_ns = self.edge_ns.max(sample.timestamp_ns); + self.samples.push_back(sample); + self.evict_expired(); + } + + pub fn set_edge(&mut self, edge_ns: u64) { + self.edge_ns = self.edge_ns.max(edge_ns); + self.evict_expired(); + } + + pub fn edge_ns(&self) -> u64 { + self.edge_ns + } + + pub fn window_start_ns(&self) -> u64 { + self.edge_ns.saturating_sub(self.duration_ns) + } + + pub fn len(&self) -> usize { + self.samples.len() + } + + pub fn is_empty(&self) -> bool { + self.samples.is_empty() + } + + pub fn iter(&self) -> impl Iterator> { + self.samples.iter() + } + + pub fn latest_for_sensor(&self, sensor_id: &str) -> Option<&Sample> { + self.samples + .iter() + .rev() + .find(|sample| sample.sensor_id == sensor_id) + } + + fn evict_expired(&mut self) { + let start_ns = self.window_start_ns(); + self.samples + .retain(|sample| sample.timestamp_ns >= start_ns && sample.timestamp_ns <= self.edge_ns); + } +} + +#[derive(Debug, Clone)] +pub struct Reservoir { + duration_ns: u64, + edge_ns: u64, + rings: BTreeMap>, +} + +impl Reservoir { + pub fn new(duration_ns: u64) -> Self { + Self { + duration_ns, + edge_ns: 0, + rings: BTreeMap::new(), + } + } + + pub fn push(&mut self, kind: RingKind, sample: Sample) { + self.edge_ns = self.edge_ns.max(sample.timestamp_ns); + for ring in self.rings.values_mut() { + ring.set_edge(self.edge_ns); + } + let ring = self + .rings + .entry(kind) + .or_insert_with(|| ReservoirRing::new(self.duration_ns)); + ring.push(sample); + ring.set_edge(self.edge_ns); + } + + pub fn set_edge(&mut self, edge_ns: u64) { + self.edge_ns = self.edge_ns.max(edge_ns); + for ring in self.rings.values_mut() { + ring.set_edge(self.edge_ns); + } + } + + pub fn ring(&self, kind: RingKind) -> Option<&ReservoirRing> { + self.rings.get(&kind) + } + + pub fn edge_ns(&self) -> u64 { + self.edge_ns + } + + pub fn window_start_ns(&self) -> u64 { + self.edge_ns.saturating_sub(self.duration_ns) + } +} + +impl Default for Reservoir { + fn default() -> Self { + Self::new(DEFAULT_RESERVOIR_NS) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ring_expires_from_newest_shared_edge() { + let mut ring = ReservoirRing::new(DEFAULT_RESERVOIR_NS); + ring.push(Sample::new("cam-a", 1_000_000_000, 1_000_000_010, 0, 1)); + ring.push(Sample::new("cam-b", 4_000_000_000, 4_000_000_010, 0, 2)); + ring.push(Sample::new("cam-a", 7_100_000_000, 7_100_000_010, 1, 3)); + + let timestamps = ring + .iter() + .map(|sample| sample.timestamp_ns) + .collect::>(); + + assert_eq!(vec![4_000_000_000, 7_100_000_000], timestamps); + assert_eq!(2_100_000_000, ring.window_start_ns()); + } + + #[test] + fn reservoir_edge_expires_all_rings() { + let mut reservoir = Reservoir::default(); + reservoir.push( + RingKind::CameraFrame, + Sample::new("cam-a", 1_000_000_000, 1_000_000_010, 0, "old-camera"), + ); + reservoir.push( + RingKind::AudioBlock, + Sample::new("audio", 7_000_000_001, 7_000_000_020, 0, "audio-edge"), + ); + + let camera_ring = reservoir.ring(RingKind::CameraFrame).unwrap(); + + assert!(camera_ring.is_empty()); + assert_eq!(2_000_000_001, reservoir.window_start_ns()); + } + + #[test] + fn latest_for_sensor_ignores_other_sources() { + let mut ring = ReservoirRing::new(DEFAULT_RESERVOIR_NS); + ring.push(Sample::new("cam-a", 10, 11, 0, "a0")); + ring.push(Sample::new("cam-b", 12, 13, 0, "b0")); + ring.push(Sample::new("cam-a", 14, 15, 1, "a1")); + + let latest = ring.latest_for_sensor("cam-a").unwrap(); + + assert_eq!("a1", latest.payload); + assert_eq!(1, latest.sequence); + } +} diff --git a/notes/current-system-map.md b/notes/current-system-map.md index 67dfd6c..eccef36 100644 --- a/notes/current-system-map.md +++ b/notes/current-system-map.md @@ -27,6 +27,7 @@ Ownership: - Faust owns hot audio DSP and program stem generation. - OBS owns broadcast controls. - Python remains tooling, not the stream hot path. +- `native/reservoir` is the first native crate. It owns shared-edge five-second sample retention for typed rings and is the seed of the Aquarium/Faust runtime boundary. Invariant: the five-second spatiotemporal reservoir is the live authority. Producers append, optimizers refine, Aquarium/Faust sample. No private history outlives the reservoir. diff --git a/state/map.yaml b/state/map.yaml index 8679004..3e6bdd7 100644 --- a/state/map.yaml +++ b/state/map.yaml @@ -33,6 +33,8 @@ invariants: - Local secrets remain untracked. accepted_design: + - `docs/perfect-machine.md` and `config/perfect-machine.example.json` define the native target: one five-second spatiotemporal reservoir, Aquarium GPU fusion/rendering, Faust DSP, OBS output controls, and LocalCastBridge tooling. + - `native/reservoir` is the first native Rust crate. It implements shared-edge reservoir rings so all sample types expire from the newest live sample rather than private source histories. - `config/localcast.example.json` defines receiver address, base port, video encode settings, and audio sources. - `scripts/sender-discover.ps1` lists FFmpeg devices, protocols, and encoders. - `scripts/sender-start.ps1` launches one FFmpeg process per enabled source.