diff --git a/crates/muninn-daemon/src/main.rs b/crates/muninn-daemon/src/main.rs index 7002bf3..35321dd 100644 --- a/crates/muninn-daemon/src/main.rs +++ b/crates/muninn-daemon/src/main.rs @@ -24,7 +24,7 @@ use odin_core::{ MUNINN_OBS_STREAM_CATALOG_SCHEMA, MuninnCaptureStreamCommandRecord, MuninnCaptureStreamRecord, MuninnCommandBoundaryCompatRecord, MuninnHidControllerStateRecord, MuninnMediaReceiverFeedbackRecord, - MuninnMoveControllerStateRecord, MuninnMoveHueProgramRecord, MuninnMoveIdentityRecord, + MuninnMoveControllerStateRecord, MuninnMoveEvidenceTransportHealthRecord, MuninnMoveHueProgramRecord, MuninnMoveIdentityRecord, MuninnMoveLightCommandRecord, MuninnMoveMarkerCandidateRecord, MuninnObsStreamCatalogRecord, MuninnQuestAccessRecord, MuninnTelemetrySurfaceRecord, MuninnTransportProfileCompatRecord, OdinDocuments, @@ -46,7 +46,7 @@ use std::net::{SocketAddr, UdpSocket}; use std::path::Path; use std::path::PathBuf; use std::process::{Child, Command, Stdio}; -use std::sync::{Arc, Mutex, mpsc}; +use std::sync::{Arc, Mutex, atomic::{AtomicU64, Ordering}, mpsc}; use std::thread; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -408,7 +408,10 @@ fn serve(options: Options) -> Result<()> { active_move_marker_camera_sources(&options, Arc::clone(&move_hue_program)); let mut move_marker_camera_reader = PlatformMoveMarkerCameraFrameReader::default(); let mut active_capture_streams = Vec::new(); - let move_evidence_rudp_sender = start_hid_controller_rudp_ingress(&options)?; + let move_evidence_rudp_sender = start_hid_controller_rudp_ingress( + &options, + move_evidence_stream.as_ref().map(|stream| Arc::clone(&stream.counters)), + )?; if let Some(stream) = move_evidence_stream.as_mut() { stream.rudp_sender = move_evidence_rudp_sender; } @@ -466,6 +469,7 @@ fn serve(options: Options) -> Result<()> { move_evidence_stream.as_mut(), )?; publish_move_tracker_health(&mut node, &mut active_move_marker_cameras)?; + publish_move_evidence_transport_health(&mut node, &options, move_evidence_stream.as_ref())?; publish_quest_access_if_requested(&mut node, &options)?; let state = if active_stream_ids.is_empty() { "idle" @@ -1653,6 +1657,15 @@ struct ActiveMoveEvidenceStream { frame_counter: u64, snapshot_path: Option, rudp_sender: Option>>>>, + counters: Arc, +} + +#[derive(Default)] +struct MoveEvidenceTransportCounters { + produced_frames: AtomicU64, + local_ring_admissions: AtomicU64, + remote_handoffs: AtomicU64, + remote_sends: AtomicU64, } fn activate(options: Options) -> Result<()> { @@ -3737,6 +3750,7 @@ fn create_move_evidence_stream(options: &Options) -> Result Result<()> { Ok(()) } +fn publish_move_evidence_transport_health( + node: &mut cultmesh_rs::CultMeshNode, + options: &Options, + stream: Option<&ActiveMoveEvidenceStream>, +) -> Result<()> { + let Some(stream) = stream else { return Ok(()); }; + let record = MuninnMoveEvidenceTransportHealthRecord { + health_id: format!("muninn:{}:move-evidence-transport-health", options.host_id), + host_id: options.host_id.clone(), + stream_id: stream.stream_id.clone(), + produced_frames: stream.counters.produced_frames.load(Ordering::Relaxed), + local_ring_admissions: stream.counters.local_ring_admissions.load(Ordering::Relaxed), + remote_handoffs: stream.counters.remote_handoffs.load(Ordering::Relaxed), + remote_sends: stream.counters.remote_sends.load(Ordering::Relaxed), + updated_at: timestamp()?, + }; + node.put(&record.health_id, &record)?; + Ok(()) +} + #[cfg(feature = "psmoveapi-tracker")] fn video_device_index(path: &Path) -> Option { path.file_name()?.to_str()?.strip_prefix("video")?.parse().ok() @@ -5208,6 +5245,7 @@ struct HidControllerRudpSubscription { fn start_hid_controller_rudp_ingress( options: &Options, + counters: Option>, ) -> Result>>>>> { let Some(bind_address) = options.hid_controller_rudp_bind else { return Ok(None); @@ -5225,7 +5263,7 @@ fn start_hid_controller_rudp_ingress( let runtime_options = options.clone(); thread::spawn(move || { if let Err(error) = - run_hid_controller_rudp_ingress(socket, runtime_options, ingress_evidence) + run_hid_controller_rudp_ingress(socket, runtime_options, ingress_evidence, counters) { eprintln!("Muninn HID controller RUDP stream stopped: {error:#}"); } @@ -5237,6 +5275,7 @@ fn run_hid_controller_rudp_ingress( socket: UdpSocket, options: Options, move_evidence_latest: Arc>>>, + counters: Option>, ) -> Result<()> { let local_addr = socket.local_addr()?; println!("Muninn HID controller RUDP stream listening at {local_addr}."); @@ -5327,6 +5366,8 @@ fn run_hid_controller_rudp_ingress( if let Some(payload) = payload { if let Err(error) = transport.send("move-evidence", payload) { eprintln!("Muninn Move evidence RUDP send warning: {error:#}"); + } else if let Some(counters) = counters.as_ref() { + counters.remote_sends.fetch_add(1, Ordering::Relaxed); } } } diff --git a/crates/odin-core/examples/dump_move_tracker_health.rs b/crates/odin-core/examples/dump_move_tracker_health.rs index b7c55fa..e8e9bc8 100644 --- a/crates/odin-core/examples/dump_move_tracker_health.rs +++ b/crates/odin-core/examples/dump_move_tracker_health.rs @@ -1,6 +1,6 @@ use anyhow::{Context, Result}; use cultmesh_rs::{CultMesh, CultMeshNodeOptions}; -use odin_core::{MuninnMoveTrackerHealthRecord, OdinDocuments}; +use odin_core::{MuninnMoveEvidenceTransportHealthRecord, MuninnMoveTrackerHealthRecord, OdinDocuments}; use std::{env, path::PathBuf}; fn main() -> Result<()> { @@ -47,5 +47,12 @@ fn main() -> Result<()> { record.detail, ); } + for record in node.cache().get_all::().unwrap_or_default() { + println!( + "stream={} produced={} local_admitted={} remote_handoffs={} remote_sends={} updated={}", + record.stream_id, record.produced_frames, record.local_ring_admissions, + record.remote_handoffs, record.remote_sends, record.updated_at, + ); + } Ok(()) } diff --git a/crates/odin-core/src/documents.rs b/crates/odin-core/src/documents.rs index 63a57f0..a3962ef 100644 --- a/crates/odin-core/src/documents.rs +++ b/crates/odin-core/src/documents.rs @@ -47,6 +47,7 @@ pub const MUNINN_MOVE_IDENTITY_SCHEMA: &str = "muninn.move_identity.v1"; pub const MUNINN_MOVE_LIGHT_COMMAND_SCHEMA: &str = "muninn.move_light_command.v1"; pub const MUNINN_MOVE_HUE_PROGRAM_SCHEMA: &str = "muninn.move_hue_program.v1"; pub const MUNINN_MOVE_TRACKER_HEALTH_SCHEMA: &str = "muninn.move_tracker_health.v1"; +pub const MUNINN_MOVE_EVIDENCE_TRANSPORT_HEALTH_SCHEMA: &str = "muninn.move_evidence_transport_health.v1"; pub const MUNINN_QUEST_ACCESS_SCHEMA: &str = "muninn.quest_access.v1"; pub const MUNINN_COMMAND_BOUNDARY_SCHEMA: &str = "muninn.command_boundary.v1"; pub const MUNINN_TRANSPORT_PROFILE_SCHEMA: &str = "muninn.transport_profile.v1"; @@ -1222,6 +1223,19 @@ pub struct MuninnMoveTrackerHealthRecord { #[cultcache(key = 20, default)] pub color_evidence_pixel_counts: Vec, } +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "muninn.move_evidence_transport_health", schema = "muninn.move_evidence_transport_health.v1")] +pub struct MuninnMoveEvidenceTransportHealthRecord { + #[cultcache(key = 0)] pub health_id: String, + #[cultcache(key = 1)] pub host_id: String, + #[cultcache(key = 2)] pub stream_id: String, + #[cultcache(key = 3)] pub produced_frames: u64, + #[cultcache(key = 4)] pub local_ring_admissions: u64, + #[cultcache(key = 5)] pub remote_handoffs: u64, + #[cultcache(key = 6)] pub remote_sends: u64, + #[cultcache(key = 7)] pub updated_at: String, +} + #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] #[cultcache(type = "muninn.quest_access", schema = "muninn.quest_access.v1")] pub struct MuninnQuestAccessRecord { @@ -1349,6 +1363,7 @@ cultmesh_rs::cultmesh_documents!(OdinDocuments { MuninnMoveLightCommandRecord => MUNINN_MOVE_LIGHT_COMMAND_SCHEMA, MuninnMoveHueProgramRecord => MUNINN_MOVE_HUE_PROGRAM_SCHEMA, MuninnMoveTrackerHealthRecord => MUNINN_MOVE_TRACKER_HEALTH_SCHEMA, + MuninnMoveEvidenceTransportHealthRecord => MUNINN_MOVE_EVIDENCE_TRANSPORT_HEALTH_SCHEMA, MuninnQuestAccessRecord => MUNINN_QUEST_ACCESS_SCHEMA, MuninnCommandBoundaryCompatRecord => MUNINN_COMMAND_BOUNDARY_SCHEMA, MuninnTransportProfileCompatRecord => MUNINN_TRANSPORT_PROFILE_SCHEMA,