diff --git a/.gitignore b/.gitignore index 71d0fc9..8fdf455 100644 --- a/.gitignore +++ b/.gitignore @@ -44,4 +44,7 @@ secluso-v* *.p12 __pycache__ *.wic -metadata.json \ No newline at end of file +metadata.json + +# Prevent un-necessary Cargo.lock files from being introduced. +Cargo.lock \ No newline at end of file diff --git a/camera_hub/Cargo.lock b/camera_hub/Cargo.lock index 7663255..5e74de7 100644 --- a/camera_hub/Cargo.lock +++ b/camera_hub/Cargo.lock @@ -3554,7 +3554,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b" dependencies = [ "aws-lc-rs", + "log", "once_cell", + "ring", "rustls-pki-types", "rustls-webpki", "subtle", @@ -3738,6 +3740,7 @@ dependencies = [ "base64-url", "bincode", "image", + "libc", "log", "openmls", "openmls_basic_credential", @@ -3748,6 +3751,7 @@ dependencies = [ "qrcode", "rand 0.9.4", "reqwest", + "rustls", "serde", "serde_derive", "serde_json", @@ -3768,6 +3772,7 @@ name = "secluso-motion-ai" version = "1.0.2" dependencies = [ "anyhow", + "cfg-if", "chrono", "crossbeam-channel", "fast_image_resize", diff --git a/camera_hub/Cargo.toml b/camera_hub/Cargo.toml index fe625e2..9109506 100644 --- a/camera_hub/Cargo.toml +++ b/camera_hub/Cargo.toml @@ -5,10 +5,14 @@ edition = "2021" authors = ["Ardalan Amiri Sani "] [features] -default = ["logging"] +default = ["logging", "crypto-aws-lc"] logging = ["log"] +# crypto-ring on armv6 (Pi Zero W) +crypto-aws-lc = ["secluso-client-lib/crypto-aws-lc"] +crypto-ring = ["secluso-client-lib/crypto-ring"] ip = ["dep:rpassword", "dep:reqwest", "dep:http-auth", "dep:linfa", "dep:linfa-clustering", "dep:retina", "dep:serde_yaml2", "dep:ndarray", "dep:futures", "secluso-client-lib/camera_secret_qrcode"] raspberry = ["dep:secluso-motion-ai"] +use_ai = ["secluso-motion-ai/ai"] manual = [] telemetry = [] # todo: dep on the motion_ai crate test = [] diff --git a/camera_hub/src/config.rs b/camera_hub/src/config.rs index cf79cec..a293df8 100644 --- a/camera_hub/src/config.rs +++ b/camera_hub/src/config.rs @@ -2,17 +2,20 @@ //! //! SPDX-License-Identifier: GPL-3.0-or-later -use crate::DeliveryMonitor; use crate::pairing::get_names; use crate::version::camera_version_info; +use crate::DeliveryMonitor; use secluso_client_lib::config::{ - Heartbeat, HeartbeatRequest, OPCODE_HEARTBEAT_REQUEST, OPCODE_HEARTBEAT_RESPONSE, - AddAppRequest, AddAppResponseCommon, AddAppResponseDedicated, OPCODE_ADD_APP_REQUEST, OPCODE_ADD_APP_RESPONSE, + AddAppRequest, AddAppResponseCommon, AddAppResponseDedicated, Heartbeat, HeartbeatRequest, + OPCODE_ADD_APP_REQUEST, OPCODE_ADD_APP_RESPONSE, OPCODE_HEARTBEAT_REQUEST, + OPCODE_HEARTBEAT_RESPONSE, }; use secluso_client_lib::http_client::HttpClient; -use secluso_client_lib::mls_clients::{NUM_MLS_CLIENTS, MlsClientsCommon, MlsClientsDedicated, CONFIG_DED, - NUM_DEDICATED_MLS_CLIENTS, NUM_COMMON_MLS_CLIENTS}; -use secluso_client_lib::mls_client::{MlsClient, ClientType}; +use secluso_client_lib::mls_client::{ClientType, MlsClient}; +use secluso_client_lib::mls_clients::{ + MlsClientsCommon, MlsClientsDedicated, CONFIG_DED, NUM_COMMON_MLS_CLIENTS, + NUM_DEDICATED_MLS_CLIENTS, NUM_MLS_CLIENTS, +}; use std::io; pub fn process_config_command( @@ -39,7 +42,7 @@ pub fn process_config_command( delivery_monitor_opt, )?; Ok(None) - }, + } OPCODE_ADD_APP_REQUEST => { if primary_app { if !second_app_already_paired { @@ -58,11 +61,11 @@ pub fn process_config_command( error!("Error: Secondary app cannot add other apps!"); Ok(None) } - }, + } _ => { error!("Error: Unknown config command opcode!"); Ok(None) - }, + } } } Err(e) => { @@ -98,7 +101,12 @@ fn handle_heartbeat_request( ); } - send_heartbeat_response(clients_com, clients_ded, heartbeat_request.timestamp, http_client)?; + send_heartbeat_response( + clients_com, + clients_ded, + heartbeat_request.timestamp, + http_client, + )?; Ok(()) } @@ -109,7 +117,8 @@ fn send_heartbeat_response( timestamp: u64, http_client: &HttpClient, ) -> io::Result<()> { - let heartbeat = Heartbeat::generate(clients_com, clients_ded, timestamp, camera_version_info()?)?; + let heartbeat = + Heartbeat::generate(clients_com, clients_ded, timestamp, camera_version_info()?)?; let mut config_msg = vec![OPCODE_HEARTBEAT_RESPONSE]; config_msg.extend(bincode::serialize(&heartbeat).unwrap()); @@ -117,7 +126,10 @@ fn send_heartbeat_response( let config_msg_enc = clients_ded[CONFIG_DED].encrypt(&config_msg)?; clients_ded[CONFIG_DED].save_group_state().unwrap(); - http_client.config_response(&clients_ded[CONFIG_DED].get_group_name().unwrap(), config_msg_enc)?; + http_client.config_response( + &clients_ded[CONFIG_DED].get_group_name().unwrap(), + config_msg_enc, + )?; Ok(()) } @@ -131,52 +143,49 @@ fn handle_add_app_request( let add_app_requests: [AddAppRequest; NUM_MLS_CLIENTS] = bincode::deserialize(command_bytes) .map_err(|e| io::Error::other(format!("Failed to deserialize add_app msg - {e}")))?; - let add_app_resps_com: [AddAppResponseCommon; NUM_COMMON_MLS_CLIENTS] = std::array::from_fn(|i| { - println!("handle_add_app_request [1]"); - let camera_key_package = clients_com[i].key_package(); - - // FIXME: "app2" is hardcoded. - println!("handle_add_app_request [2]"); - let camera_contact = - MlsClient::create_contact("app2", add_app_requests[i].new_app_key_package.clone()) + let add_app_resps_com: [AddAppResponseCommon; NUM_COMMON_MLS_CLIENTS] = + std::array::from_fn(|i| { + println!("handle_add_app_request [1]"); + let camera_key_package = clients_com[i].key_package(); + + // FIXME: "app2" is hardcoded. + println!("handle_add_app_request [2]"); + let camera_contact = + MlsClient::create_contact("app2", add_app_requests[i].new_app_key_package.clone()) + .unwrap(); + + println!("handle_add_app_request [3]"); + // FIXME: Use a different secret per channel + let (welcome_msg_vec, psk_proposal_vec, commit_msg_vec) = clients_com[i] + .invite_with_secret(&camera_contact, add_app_requests[i].secret.clone()) .unwrap(); - println!("handle_add_app_request [3]"); - // FIXME: Use a different secret per channel - let (welcome_msg_vec, psk_proposal_vec, commit_msg_vec) = clients_com[i] - .invite_with_secret(&camera_contact, add_app_requests[i].secret.clone()) - .unwrap(); - - println!("handle_add_app_request [4]"); - clients_com[i].save_group_state().unwrap(); + println!("handle_add_app_request [4]"); + clients_com[i].save_group_state().unwrap(); - println!("handle_add_app_request [5]"); - AddAppResponseCommon { - camera_key_package, - welcome_msg_vec, - psk_proposal_vec, - commit_msg_vec, - } - }); + println!("handle_add_app_request [5]"); + AddAppResponseCommon { + camera_key_package, + welcome_msg_vec, + psk_proposal_vec, + commit_msg_vec, + } + }); - let [(client_l, resp_l), (client_c, resp_c)]: [(MlsClient, AddAppResponseDedicated); NUM_DEDICATED_MLS_CLIENTS] = - std::array::from_fn(|i| { + let [(client_l, resp_l), (client_c, resp_c)]: [(MlsClient, AddAppResponseDedicated); + NUM_DEDICATED_MLS_CLIENTS] = std::array::from_fn(|i| { // This part of code has a lot in common with initialize_mls_clients() in main.rs // Initialize mls_client // FIXME - let tag = if i == 0 { - "livestream2" - } else { - "config2" - }; + let tag = if i == 0 { "livestream2" } else { "config2" }; let (camera_name, group_name) = get_names( - clients_ded[CONFIG_DED].get_file_dir(), // Could use either of the clients - true, - format!("camera_{}_name", tag), - format!("group_{}_name", tag), - ); + clients_ded[CONFIG_DED].get_file_dir(), // Could use either of the clients + true, + format!("camera_{}_name", tag), + format!("group_{}_name", tag), + ); let mut client = MlsClient::new( camera_name, true, @@ -193,15 +202,21 @@ fn handle_add_app_request( // Now invite let camera_key_package = client.key_package(); - let app_key_package = add_app_requests[i + NUM_COMMON_MLS_CLIENTS].new_app_key_package.clone(); + let app_key_package = add_app_requests[i + NUM_COMMON_MLS_CLIENTS] + .new_app_key_package + .clone(); let app_contact = MlsClient::create_contact("app", app_key_package).unwrap(); info!("Added contact."); let (welcome_msg_vec, _, _) = client - .invite_with_secret(&app_contact, add_app_requests[i + NUM_COMMON_MLS_CLIENTS].secret.clone()) + .invite_with_secret( + &app_contact, + add_app_requests[i + NUM_COMMON_MLS_CLIENTS].secret.clone(), + ) .inspect_err(|_| { error!("invite() returned error:"); - }).unwrap(); + }) + .unwrap(); client.save_group_state().unwrap(); info!("App invited to the group."); @@ -230,7 +245,10 @@ fn handle_add_app_request( println!("[1]: config_msg_enc len = {:?}", config_msg_enc.len()); clients_ded[CONFIG_DED].save_group_state().unwrap(); - http_client.config_response(&clients_ded[CONFIG_DED].get_group_name().unwrap(), config_msg_enc)?; + http_client.config_response( + &clients_ded[CONFIG_DED].get_group_name().unwrap(), + config_msg_enc, + )?; Ok(Some(new_clients_ded)) } diff --git a/camera_hub/src/delivery_monitor.rs b/camera_hub/src/delivery_monitor.rs index daca634..b3bf59a 100644 --- a/camera_hub/src/delivery_monitor.rs +++ b/camera_hub/src/delivery_monitor.rs @@ -124,8 +124,7 @@ impl DeliveryMonitor { file.sync_all().unwrap(); //delete old state files - let d_files = - Self::get_state_files_sorted(&self.state_dir, "delivery_monitor_").unwrap(); + let d_files = Self::get_state_files_sorted(&self.state_dir, "delivery_monitor_").unwrap(); assert!(d_files[0] == "delivery_monitor_".to_owned() + ¤t_timestamp.to_string()); for f in &d_files[1..] { let _ = fs::remove_file(self.state_dir.clone() + "/" + f); @@ -278,7 +277,9 @@ impl DeliveryMonitor { pub fn get_thumbnail_file_path(&self, info: &ThumbnailMetaInfo) -> PathBuf { let video_dir_path = Path::new(&self.thumbnail_dir); - video_dir_path.join(ThumbnailMetaInfo::get_filename_from_timestamp(info.timestamp)) + video_dir_path.join(ThumbnailMetaInfo::get_filename_from_timestamp( + info.timestamp, + )) } pub fn get_enc_video_file_path(&self, info: &VideoInfo) -> PathBuf { diff --git a/camera_hub/src/fmp4.rs b/camera_hub/src/fmp4.rs index 2ea69c1..3705166 100644 --- a/camera_hub/src/fmp4.rs +++ b/camera_hub/src/fmp4.rs @@ -101,7 +101,7 @@ impl TrakTracker { // Writes tfdt + trun. Returns the moof-relative byte position of the i32 data_offset. fn write_fragment(&self, buf: &mut BytesMut) -> Result, Error> { write_box!(buf, b"tfdt", { - buf.put_u32(1 << 24); // version=1, flags=0 + buf.put_u32(1 << 24); // version=1, flags=0 buf.put_u64(self.fragment_start_time); }); @@ -113,7 +113,8 @@ impl TrakTracker { const TRUN_SAMPLE_SIZE: u32 = 0x000200; write_box!(buf, b"trun", { - let flags = TRUN_DATA_OFFSET | TRUN_FIRST_SAMPLE_FL | TRUN_SAMPLE_DURATION | TRUN_SAMPLE_SIZE; + let flags = + TRUN_DATA_OFFSET | TRUN_FIRST_SAMPLE_FL | TRUN_SAMPLE_DURATION | TRUN_SAMPLE_SIZE; buf.put_u32(flags); buf.put_u32(self.core.samples); @@ -150,7 +151,6 @@ impl TrakTracker { } } - /// Writes fragmented `.mp4` data to a sink. pub struct Fmp4Writer { core: Mp4WriterCore, @@ -168,12 +168,12 @@ impl Fmp4Writer Result { let mut buf = BytesMut::new(); write_box!(&mut buf, b"ftyp", { - buf.extend_from_slice(b"isom"); // major_brand - buf.extend_from_slice(&0x00000200u32.to_be_bytes()); // minor_version - buf.extend_from_slice(b"isom"); // compat[0] - buf.extend_from_slice(b"iso6"); // compat[1] - buf.extend_from_slice(b"avc1"); // compat[2] - buf.extend_from_slice(b"mp41"); // compat[3] + buf.extend_from_slice(b"isom"); // major_brand + buf.extend_from_slice(&0x00000200u32.to_be_bytes()); // minor_version + buf.extend_from_slice(b"isom"); // compat[0] + buf.extend_from_slice(b"iso6"); // compat[1] + buf.extend_from_slice(b"avc1"); // compat[2] + buf.extend_from_slice(b"mp41"); // compat[3] }); inner.write_all(&buf).await?; @@ -201,42 +201,45 @@ impl Fmp4Writer Fmp4Writer Fmp4Writer Fmp4Writer Mp4 for Fmp4Writer { async fn video(&mut self, frame: &[u8], ts: u64, is_rap: bool) -> Result<(), Error> { let size = u32::try_from(frame.len())?; self.video_trak.add_sample(size, ts, is_rap)?; - self.core.mdat_pos = self.core.mdat_pos.checked_add(size).ok_or_else(|| anyhow!("mdat_pos overflow"))?; + self.core.mdat_pos = self + .core + .mdat_pos + .checked_add(size) + .ok_or_else(|| anyhow!("mdat_pos overflow"))?; self.fbuf_video.extend_from_slice(frame); Ok(()) } @@ -298,7 +304,11 @@ impl Mp4 for Fmp4 async fn audio(&mut self, frame: &[u8], ts: u64) -> Result<(), Error> { let size = u32::try_from(frame.len())?; self.audio_trak.add_sample(size, ts, false)?; - self.core.mdat_pos = self.core.mdat_pos.checked_add(size).ok_or_else(|| anyhow!("mdat_pos overflow"))?; + self.core.mdat_pos = self + .core + .mdat_pos + .checked_add(size) + .ok_or_else(|| anyhow!("mdat_pos overflow"))?; self.fbuf_audio.extend_from_slice(frame); Ok(()) } diff --git a/camera_hub/src/ip/mod.rs b/camera_hub/src/ip/mod.rs index 4b091ca..2919382 100644 --- a/camera_hub/src/ip/mod.rs +++ b/camera_hub/src/ip/mod.rs @@ -1,4 +1,4 @@ //! SPDX-License-Identifier: GPL-3.0-or-later pub(crate) mod ip_camera; -pub(crate) mod ip_motion_detection; \ No newline at end of file +pub(crate) mod ip_motion_detection; diff --git a/camera_hub/src/livestream.rs b/camera_hub/src/livestream.rs index 28f7269..b22b841 100644 --- a/camera_hub/src/livestream.rs +++ b/camera_hub/src/livestream.rs @@ -74,7 +74,12 @@ pub fn livestream( } // Update MLS epoch + let update_start = Instant::now(); let (commit_msg, _epoch) = mls_client.update()?; + debug!( + "Livestream: mls_client.update() took {}ms", + update_start.elapsed().as_millis() + ); mls_client.save_group_state().unwrap(); let group_name = mls_client.get_group_name().unwrap(); @@ -99,6 +104,7 @@ pub fn livestream( camera.launch_livestream(livestream_writer)?; let mut chunk_number: u64 = 1; + let mut last_recv = Instant::now(); loop { // We include the chunk number in the chunk itself (and check it in the app) @@ -107,6 +113,12 @@ pub fn livestream( info!("Ending livestream because the camera backend stopped producing fragments."); break; }; + debug!( + "Livestream: {}ms since previous fragment for chunk {}", + last_recv.elapsed().as_millis(), + chunk_number + ); + last_recv = Instant::now(); let mut data: Vec = chunk_number.to_be_bytes().to_vec(); data.extend(fragment); @@ -143,6 +155,11 @@ pub fn livestream( chunk_number - 1, curr_epoch_ms ); + debug!( + "Livestream: {} pending chunk(s) on server after chunk {}", + num_pending_files, + chunk_number - 1 + ); // The server returns 0 when the app has explicitly ended livestream if num_pending_files == 0 || num_pending_files > MAX_NUM_PENDING_LIVESTREAM_CHUNKS { @@ -151,7 +168,12 @@ pub fn livestream( } } + let save_start = Instant::now(); mls_client.save_group_state().unwrap(); + debug!( + "Livestream: final save_group_state() took {}ms", + save_start.elapsed().as_millis() + ); Ok(()) } diff --git a/camera_hub/src/main.rs b/camera_hub/src/main.rs index 762b0c2..26f5db1 100644 --- a/camera_hub/src/main.rs +++ b/camera_hub/src/main.rs @@ -13,9 +13,8 @@ use docopt::Docopt; use secluso_client_lib::http_client::HttpClient; use secluso_client_lib::mls_client::{ClientType, MlsClient}; use secluso_client_lib::mls_clients::{ - MlsClients, FCM, MLS_CLIENT_TAGS, MOTION, NUM_MLS_CLIENTS, - THUMBNAIL, LIVESTREAM_DED, CONFIG_DED, - MlsClientsCommon, MlsClientsDedicated, + MlsClients, MlsClientsCommon, MlsClientsDedicated, CONFIG_DED, FCM, LIVESTREAM_DED, + MLS_CLIENT_TAGS, MOTION, NUM_MLS_CLIENTS, THUMBNAIL, }; use secluso_client_lib::thumbnail_meta_info::ThumbnailMetaInfo; use std::array; @@ -38,8 +37,8 @@ use crate::delivery_monitor::{DeliveryMonitor, VideoInfo}; mod motion; use crate::motion::{ - prepare_motion_thumbnail, prepare_motion_video, - upload_pending_enc_thumbnails, upload_pending_enc_videos, + prepare_motion_thumbnail, prepare_motion_video, upload_pending_enc_thumbnails, + upload_pending_enc_videos, }; mod livestream; @@ -132,10 +131,21 @@ struct Args { flag_save_all: bool, } -fn main() -> io::Result<()> { +fn main() -> anyhow::Result<()> { let version = env!("CARGO_PKG_NAME").to_string() + ", version: " + env!("CARGO_PKG_VERSION"); env_logger::init(); + // ring TLS backend (armv6/Pi Zero W) needs provider installed before any HTTPS request + #[cfg(feature = "crypto-ring")] + secluso_client_lib::http_client::install_crypto_provider(); + + // SECLUSO_CRYPTO_SELFTEST=1 RUST_LOG=info /usr/bin/secluso-camera-hub + // Runs MLS in hot loop on the otherwise idle process, then exits without starting the camera pipeline + if std::env::var_os("SECLUSO_CRYPTO_SELFTEST").is_some() { + secluso_client_lib::crypto_selftest::run()?; + return Ok(()); + } + let args: Args = Docopt::new(USAGE) .map(|d| d.help(true)) .map(|d| d.version(Some(version))) @@ -412,6 +422,9 @@ fn core( println!("[{}] Running...", camera_name); + // We start it after we've paired. It's off by default. + camera.set_motion_active(true); + let (server_username, server_password, server_addr) = read_parse_full_credentials(); let http_client = HttpClient::new(server_addr, server_username, server_password); @@ -425,7 +438,9 @@ fn core( DeliveryMonitor::from_file_or_new(video_dir, thumbnail_dir, state_dir.clone()); let livestream_request = Arc::new(Mutex::new((false, true))); let livestream_request_clone = Arc::clone(&livestream_request); - let group_livestream_name_clone = clients_ded_primary[LIVESTREAM_DED].get_group_name().unwrap(); + let group_livestream_name_clone = clients_ded_primary[LIVESTREAM_DED] + .get_group_name() + .unwrap(); let http_client_clone = http_client.clone(); let group_config_name_clone = clients_ded_primary[CONFIG_DED].get_group_name().unwrap(); let http_client_clone_2 = http_client.clone(); @@ -440,7 +455,7 @@ fn core( { println!("Livestream1 detected"); let mut check = livestream_request_clone.lock().unwrap(); - *check = (true, true); // second true -> livestream command from the primary app + *check = (true, true); // second true -> livestream command from the primary app } else { sleep(Duration::from_secs(1)); } @@ -476,12 +491,11 @@ fn core( let motion_timestamp = video_info.timestamp; println!("Detected motion."); + // We do not want motion to continue running after we've started recording. We'll resume it again after we're done. + camera.set_motion_active(false); + let clients_ded_sec_opt = clients_ded_secondary.lock().unwrap(); - let num_apps = if clients_ded_sec_opt.is_some() { - 2 - } else { - 1 - }; + let num_apps = if clients_ded_sec_opt.is_some() { 2 } else { 1 }; // We send the thumbnail BEFORE the FCM notification, to ensure that when the mobile app receives it, it can download it. if let Some(thumbnail_image) = motion_event.thumbnail { @@ -548,8 +562,8 @@ fn core( platform_label ); let notification_timestamp: u64 = 0; - let notification_msg = clients_com[FCM] - .encrypt(&bincode::serialize(¬ification_timestamp).unwrap())?; + let notification_msg = + clients_com[FCM].encrypt(&bincode::serialize(¬ification_timestamp).unwrap())?; clients_com[FCM].save_group_state().unwrap(); match send_notification(state_dir_ref, &http_client, notification_msg) { Ok(_) => {} @@ -558,6 +572,8 @@ fn core( } } + // Resume motion detection again after we're done recording. + camera.set_motion_active(true); locked_motion_check_time = Some(Instant::now().add(Duration::from_secs(60))); } @@ -571,6 +587,9 @@ fn core( let primary_app = check.1; if check.0 { info!("Livestream start detected"); + // We do not want motion to continue running after we've started recording. We'll resume it again afterwards. + camera.set_motion_active(false); + *check = (false, false); if primary_app { livestream( @@ -581,7 +600,8 @@ fn core( )?; } else { let mut clients_ded_sec_opt = clients_ded_secondary.lock().unwrap(); - if let Some(ref mut clients_ded_sec) = *clients_ded_sec_opt { // Should always be the case if we get here + if let Some(ref mut clients_ded_sec) = *clients_ded_sec_opt { + // Should always be the case if we get here livestream( &mut clients_ded_sec[LIVESTREAM_DED], camera, @@ -593,6 +613,8 @@ fn core( } } + // Resume motion detection again after we're done recording. + camera.set_motion_active(true); locked_livestream_check_time = Some(Instant::now().add(Duration::from_millis(100))); } @@ -601,11 +623,7 @@ fn core( || locked_delivery_check_time.unwrap().le(&Instant::now()) { let clients_ded_sec_opt = clients_ded_secondary.lock().unwrap(); - let num_apps = if clients_ded_sec_opt.is_some() { - 2 - } else { - 1 - }; + let num_apps = if clients_ded_sec_opt.is_some() { 2 } else { 1 }; if upload_pending_enc_videos( &clients_com[MOTION].get_group_name().unwrap(), @@ -667,10 +685,12 @@ fn core( if let Some(ref clients_ded_sec) = *clients_ded_sec_opt { println!("Launching threads for the second app."); - let group_livestream2_name_clone = clients_ded_sec[LIVESTREAM_DED].get_group_name()?; + let group_livestream2_name_clone = + clients_ded_sec[LIVESTREAM_DED].get_group_name()?; let livestream_request_clone_2 = Arc::clone(&livestream_request); let http_client_clone_3 = http_client.clone(); - let group_config2_name_clone = clients_ded_sec[CONFIG_DED].get_group_name()?; + let group_config2_name_clone = + clients_ded_sec[CONFIG_DED].get_group_name()?; let http_client_clone_4 = http_client.clone(); let config_enc_commands_clone_2 = Arc::clone(&config_enc_commands); @@ -688,9 +708,13 @@ fn core( }); thread::spawn(move || loop { - if let Ok(enc_command) = http_client_clone_4.config_check(&group_config2_name_clone) { - let mut config_enc_commands = config_enc_commands_clone_2.lock().unwrap(); - config_enc_commands.push((enc_command, false)); // false -> config command from the secondary app + if let Ok(enc_command) = + http_client_clone_4.config_check(&group_config2_name_clone) + { + let mut config_enc_commands = + config_enc_commands_clone_2.lock().unwrap(); + config_enc_commands.push((enc_command, false)); + // false -> config command from the secondary app } else { error!("Error in receiving config command"); sleep(Duration::from_secs(1)); diff --git a/camera_hub/src/manual.rs b/camera_hub/src/manual.rs index a211006..da87d79 100644 --- a/camera_hub/src/manual.rs +++ b/camera_hub/src/manual.rs @@ -134,16 +134,16 @@ impl ManualCamera { let video_path = PathBuf::from(&tokens[1]); if !video_path.exists() { - return Err(format!("Video file does not exist: {}", video_path.display())); + return Err(format!( + "Video file does not exist: {}", + video_path.display() + )); } let thumbnail_path = if let Some(thumbnail_token) = tokens.get(2) { let path = PathBuf::from(thumbnail_token); if !path.exists() { - return Err(format!( - "Thumbnail file does not exist: {}", - path.display() - )); + return Err(format!("Thumbnail file does not exist: {}", path.display())); } Some(path) @@ -247,11 +247,12 @@ impl ManualCamera { } fn default_webcam_command() -> String { - let device = std::env::var("SECLUSO_MANUAL_WEBCAM_DEVICE").unwrap_or_else(|_| "FaceTime HD Camera".to_string()); + let device = std::env::var("SECLUSO_MANUAL_WEBCAM_DEVICE") + .unwrap_or_else(|_| "FaceTime HD Camera".to_string()); let frame_rate = std::env::var("SECLUSO_MANUAL_WEBCAM_FPS").unwrap_or_else(|_| "30".to_string()); - let video_size = std::env::var("SECLUSO_MANUAL_WEBCAM_SIZE") - .unwrap_or_else(|_| "640x480".to_string()); + let video_size = + std::env::var("SECLUSO_MANUAL_WEBCAM_SIZE").unwrap_or_else(|_| "640x480".to_string()); let input_pixel_format = std::env::var("SECLUSO_MANUAL_WEBCAM_PIXEL_FORMAT") .unwrap_or_else(|_| "nv12".to_string()); diff --git a/camera_hub/src/motion.rs b/camera_hub/src/motion.rs index 6f73a7b..2574353 100644 --- a/camera_hub/src/motion.rs +++ b/camera_hub/src/motion.rs @@ -6,7 +6,7 @@ use crate::delivery_monitor::{DeliveryMonitor, VideoInfo}; use image::RgbImage; use secluso_client_lib::http_client::HttpClient; use secluso_client_lib::mls_client::MlsClient; -use secluso_client_lib::mls_clients::{MAX_OFFLINE_WINDOW}; +use secluso_client_lib::mls_clients::MAX_OFFLINE_WINDOW; use secluso_client_lib::thumbnail_meta_info::{GeneralDetectionType, ThumbnailMetaInfo}; use secluso_client_lib::video::{encrypt_thumbnail_file, encrypt_video_file}; use std::io; diff --git a/camera_hub/src/pairing.rs b/camera_hub/src/pairing.rs index a28432d..32e88aa 100644 --- a/camera_hub/src/pairing.rs +++ b/camera_hub/src/pairing.rs @@ -18,6 +18,7 @@ use serde_json::Value; use std::fs; use std::fs::File; use std::io; +use std::io::ErrorKind; use std::io::{BufRead, BufReader, Read, Write}; use std::net::{TcpListener, TcpStream, ToSocketAddrs}; use std::path::Path; @@ -26,7 +27,6 @@ use std::process::{Command, Stdio}; use std::sync::{Mutex, OnceLock}; use std::{thread, time::Duration}; use url::Url; -use std::io::ErrorKind; // Used to generate random names. // With 16 alphanumeric characters, the probability of collision is very low. @@ -147,7 +147,8 @@ pub fn get_input_wifi_password() -> String { Ok("1") => "/provision/wifi_password", _ => "./wifi_password", }; - let contents = fs::read_to_string(pathname).expect("Failed to read from \"wifi_password\" file. You can generate this in config tool"); + let contents = fs::read_to_string(pathname) + .expect("Failed to read from \"wifi_password\" file. You can generate this in config tool"); return contents; } @@ -263,7 +264,14 @@ fn ensure_command_success(output: Output, context: &str) -> io::Result { fn active_connections() -> io::Result> { // Read the active profile list once so readiness checks reason about the same NM view. - let output = nmcli_stdout(&["-t", "-f", "NAME,TYPE,DEVICE", "connection", "show", "--active"])?; + let output = nmcli_stdout(&[ + "-t", + "-f", + "NAME,TYPE,DEVICE", + "connection", + "show", + "--active", + ])?; let mut active = Vec::new(); for line in output.lines() { let mut parts = line.splitn(3, ':'); @@ -398,7 +406,10 @@ fn wait_for_wifi_readiness(ssid: &str, server_addr: &str, timeout: Duration) -> // what layer is lagging: wrong active connection, no device yet, no DHCP yet, no route yet, or no relay reachability yet... let active_names = active_connection_names()?; if !active_names.iter().any(|name| name == ssid) { - last_reason = format!("active connections are {:?}, expected {}", active_names, ssid); + last_reason = format!( + "active connections are {:?}, expected {}", + active_names, ssid + ); thread::sleep(Duration::from_millis(500)); continue; } @@ -458,7 +469,9 @@ fn wait_for_hotspot_shutdown(timeout: Duration) -> io::Result<()> { thread::sleep(Duration::from_millis(500)); } - Err(io::Error::other("Timed out waiting for hotspot to shut down")) + Err(io::Error::other( + "Timed out waiting for hotspot to shut down", + )) } fn wait_for_ssid_visibility(ssid: &str, timeout: Duration) -> io::Result<()> { diff --git a/camera_hub/src/raspberry_pi/mod.rs b/camera_hub/src/raspberry_pi/mod.rs index f9d32a7..9207d8d 100644 --- a/camera_hub/src/raspberry_pi/mod.rs +++ b/camera_hub/src/raspberry_pi/mod.rs @@ -1,4 +1,4 @@ //! SPDX-License-Identifier: GPL-3.0-or-later pub(crate) mod rpi_camera; -mod rpi_dual_stream; \ No newline at end of file +mod rpi_dual_stream; diff --git a/camera_hub/src/raspberry_pi/rpi_camera.rs b/camera_hub/src/raspberry_pi/rpi_camera.rs index 0b8d28b..efdf51f 100644 --- a/camera_hub/src/raspberry_pi/rpi_camera.rs +++ b/camera_hub/src/raspberry_pi/rpi_camera.rs @@ -7,7 +7,10 @@ use std::{ collections::VecDeque, io, process::Command, - sync::{Arc, Mutex}, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, Mutex, + }, thread, time::Duration, }; @@ -29,6 +32,8 @@ use crossbeam_channel::unbounded; use image::RgbImage; use secluso_client_lib::thumbnail_meta_info::GeneralDetectionType; use secluso_motion_ai::logic::pipeline::PipelineController; + +#[cfg(feature = "use_ai")] use secluso_motion_ai::ml::models::DetectionType; use secluso_motion_ai::pipeline; use tokio::runtime::Runtime; @@ -79,6 +84,9 @@ pub struct RaspberryPiCamera { sps_frame: Frame, pps_frame: Frame, motion_detection: Arc>, + // Mirrors the pipeline's run_detections. + // When false (while recording), the raw-frame reader thread closes the secondary-stream socket, so rpicam stops sending raw frames + motion_active: Arc, resolution: CameraResolution, } @@ -102,7 +110,9 @@ impl RaspberryPiCamera { // Start motion detection using raw frames from the shared stream. let pipeline = pipeline![ secluso_motion_ai::logic::stages::MotionStage, + #[cfg(feature = "use_ai")] secluso_motion_ai::logic::stages::InferenceStage, + secluso_motion_ai::logic::stages::EmitFrameIfDetected, ]; let write_logs = cfg!(feature = "telemetry"); @@ -141,7 +151,10 @@ impl RaspberryPiCamera { debug!("Exited controller tick loop"); }); - let resolution: CameraResolution = Self::fetch_resolution().expect("A supported camera module was not found"); + let resolution: CameraResolution = + Self::fetch_resolution().expect("A supported camera module was not found"); + + let motion_active = Arc::new(AtomicBool::new(false)); // Start the new shared stream. rpi_dual_stream::start( @@ -153,8 +166,9 @@ impl RaspberryPiCamera { Arc::clone(&frame_queue), ps_tx, motion_fps as u8, + Arc::clone(&motion_active), ) - .expect("Failed to start shared stream"); + .expect("Failed to start shared stream"); rpi_dual_stream::start_audio(Arc::clone(&frame_queue)) .expect("Failed to start audio stream"); @@ -189,6 +203,7 @@ impl RaspberryPiCamera { sps_frame, pps_frame, motion_detection, + motion_active, resolution, } } @@ -315,7 +330,7 @@ impl RaspberryPiCamera { RpiCameraAudioParameters::new(), file, ) - .await?; + .await?; // Process the rest of the frames, writing both to the MP4 writer and to the raw file. Self::copy(&mut mp4, Some(duration), frame_queue).await?; @@ -435,7 +450,7 @@ impl RaspberryPiCamera { RpiCameraAudioParameters::new(), livestream_writer, ) - .await?; + .await?; fmp4.finish_header(None).await?; Self::copy(&mut fmp4, None, frame_queue).await?; @@ -556,7 +571,7 @@ impl RaspberryPiCamera { }) } else { None - } + }; } } @@ -570,15 +585,26 @@ impl Camera for RaspberryPiCamera { let img = RgbImage::from_raw(frame.width as u32, frame.height as u32, data) .expect("Failed to convert RGB data into RgbImage"); - // TODO: We have to manually map these until we connect the IP camera to motion_ai - let mut detections: Vec = Vec::new(); - for detection in pipeline_result.detections { - if detection == DetectionType::Animal { - detections.push(GeneralDetectionType::Pet); - } else if detection == DetectionType::Human { - detections.push(GeneralDetectionType::Human); - } else if detection == DetectionType::Car { - detections.push(GeneralDetectionType::Car); + cfg_if::cfg_if! { + if #[cfg(feature = "use_ai")] { + // TODO: We have to manually map these until we connect the IP camera to motion_ai + let mut detections: Vec = Vec::new(); + + // If AI feature is off, the Vec will be empty. + // That means no AI detections, which fits here. + // Although, will the app send notifications for just motion? + // TODO: Should the app be notified if the camera is AI capable? + for detection in pipeline_result.detections { + if detection == DetectionType::Animal { + detections.push(GeneralDetectionType::Pet); + } else if detection == DetectionType::Human { + detections.push(GeneralDetectionType::Human); + } else if detection == DetectionType::Car { + detections.push(GeneralDetectionType::Car); + } + } + } else { + let detections: Vec = Vec::new(); } } @@ -596,6 +622,11 @@ impl Camera for RaspberryPiCamera { }) } + fn set_motion_active(&self, active: bool) { + self.motion_active.store(active, Ordering::Relaxed); + self.motion_detection.lock().unwrap().set_pipeline_active(active); + } + fn record_motion_video(&self, info: &VideoInfo, duration: u64) -> io::Result<()> { let rt = Runtime::new()?; @@ -608,7 +639,7 @@ impl Camera for RaspberryPiCamera { Arc::clone(&self.frame_queue), self.sps_frame.clone(), self.pps_frame.clone(), - self.resolution.clone() + self.resolution.clone(), ); rt.block_on(future).unwrap(); @@ -634,7 +665,7 @@ impl Camera for RaspberryPiCamera { frame_queue_clone, sps_frame_clone, pps_frame_clone, - resolution_clone + resolution_clone, ); if let Err(e) = rt.block_on(future) { eprintln!("[Livestream] write_fmp4 error: {e:?}"); @@ -669,7 +700,11 @@ struct RpiCameraVideoParameters { impl RpiCameraVideoParameters { pub fn new(sps: Vec, pps: Vec, dimensions: CameraResolution) -> Self { - Self { sps, pps, dimensions } + Self { + sps, + pps, + dimensions, + } } } @@ -757,7 +792,10 @@ impl CodecParameters for RpiCameraVideoParameters { } fn get_dimensions(&self) -> (u32, u32) { - ((self.dimensions.width as u32) << 16, (self.dimensions.height as u32) << 16) + ( + (self.dimensions.width as u32) << 16, + (self.dimensions.height as u32) << 16, + ) } } diff --git a/camera_hub/src/raspberry_pi/rpi_dual_stream.rs b/camera_hub/src/raspberry_pi/rpi_dual_stream.rs index 7b3b50b..6143c32 100644 --- a/camera_hub/src/raspberry_pi/rpi_dual_stream.rs +++ b/camera_hub/src/raspberry_pi/rpi_dual_stream.rs @@ -3,18 +3,19 @@ //! //! SPDX-License-Identifier: GPL-3.0-or-later +use bytes::Buf; use std::collections::VecDeque; use std::os::unix::net::UnixStream; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::thread::sleep; -use std::time::{SystemTime}; +use std::time::SystemTime; use std::{ io::{BufReader, Read, Write}, process::{Command, Stdio}, thread, time::Duration, }; -use bytes::Buf; use crate::raspberry_pi::rpi_camera::{Frame, FrameKind}; use anyhow::anyhow; @@ -34,6 +35,7 @@ pub fn start( frame_queue: Arc>>, ps_tx: Sender, motion_fps: u8, + motion_active: Arc, ) -> Result<(), Box> { // For 8-bit yuv420p, frame size = width * height * 3/2 bytes. // However, we need to take into account how the width is padded to 64-bytes. @@ -93,12 +95,12 @@ pub fn start( } } Err(e) => { - println!("Got error {:?}", e); + println!("Got error {e:?}"); } } } Err(e) => { - eprintln!("Error reading rpicam stdout: {:?}", e); + eprintln!("Error reading rpicam stdout: {e:?}"); break; } } @@ -107,37 +109,37 @@ pub fn start( } // Spawn a thread that will continuously read full frames from a UNIX domain socket in the modified rpicam-vid + // Only holds the connection open while detection is active. + // Re-connect after motion/livestream event is sent out. { - thread::spawn(move || { - let stream_attempt: Option = connect_to_socket(); - if stream_attempt.is_none() { - panic!("Was unable to connect to the rpicam-vid socket. Are you using the built rpicam-apps secluso fork?"); + thread::spawn(move || loop { + // Park (cheaply) until detection is active. + while !motion_active.load(Ordering::Relaxed) { + sleep(Duration::from_millis(100)); } - let mut stream = stream_attempt.unwrap(); // Unwrap will work since we checked is_none() - - // Write the motion_fps we want the output to synchronize to for maximum efficiency. + // (Re)connect and request our motion FPS. + let Some(mut stream) = connect_to_socket() else { + panic!("Was unable to connect to the rpicam-vid socket. Are you using the built rpicam-apps secluso fork?"); + }; if let Err(e) = stream.write(&[motion_fps]) { - panic!("Failed to write Motion FPS to rpicam-vid: {:?}", e); + panic!("Failed to write Motion FPS to rpicam-vid: {e:?}"); } - // Continuously read in frames from the secondary stream - loop { + // Continuously read in frames from the secondary stream (while connection stays active) + while motion_active.load(Ordering::Relaxed) { let mut buffer = vec![0u8; yuv_frame_size]; - match stream.read_exact(&mut buffer) { - Ok(_) => { + Ok(()) => { let raw_frame = RawFrame::create_from_buffer(buffer, width, height); - { - let mut lock = pipeline_controller.lock().unwrap(); - lock.push_frame(raw_frame); - } + let mut lock = pipeline_controller.lock().unwrap(); + lock.push_frame(raw_frame); } Err(e) => { - panic!( - "Error reading from UNIX domain socket from secondary stream: {:?}", - e + eprintln!( + "Error reading from secondary stream socket: {e:?}; will reconnect.", ); + break; } } } @@ -275,9 +277,13 @@ fn extract_h264_frame(buffer: &mut BytesMut) -> anyhow::Result> { } fn adts_frame_len(header: &[u8]) -> Option { - if header.len() < 7 { return None; } + if header.len() < 7 { + return None; + } // syncword 0xFFF - if header[0] != 0xFF || (header[1] & 0xF0) != 0xF0 { return None; } + if header[0] != 0xFF || (header[1] & 0xF0) != 0xF0 { + return None; + } let protection_absent = header[1] & 0x01; let hdr_len = if protection_absent == 1 { 7 } else { 9 }; @@ -285,23 +291,30 @@ fn adts_frame_len(header: &[u8]) -> Option { | ((header[4] as usize) << 3) | (((header[5] & 0xE0) as usize) >> 5); - if frame_length < hdr_len { return None; } + if frame_length < hdr_len { + return None; + } Some(frame_length) } fn strip_adts(frame: &[u8]) -> Option<&[u8]> { - if frame.len() < 7 { return None; } - if frame[0] != 0xFF || (frame[1] & 0xF0) != 0xF0 { return None; } + if frame.len() < 7 { + return None; + } + if frame[0] != 0xFF || (frame[1] & 0xF0) != 0xF0 { + return None; + } let protection_absent = frame[1] & 0x01; let hdr_len = if protection_absent == 1 { 7 } else { 9 }; - if frame.len() < hdr_len { return None; } + if frame.len() < hdr_len { + return None; + } Some(&frame[hdr_len..]) } pub fn start_audio( frame_queue: Arc>>, ) -> Result<(), Box> { - let cmd = "\ arecord -D plughw:0,0 -f S16_LE -r 48000 -c 1 -t raw | \ sox -t raw -b 16 -e signed-integer -r 48000 -c 1 - \ @@ -335,7 +348,9 @@ pub fn start_audio( buf.extend_from_slice(&tmp[..n]); loop { - if buf.len() < 7 { break; } + if buf.len() < 7 { + break; + } let len = match adts_frame_len(&buf[..7]) { Some(l) => l, None => { @@ -344,7 +359,9 @@ pub fn start_audio( continue; } }; - if buf.len() < len { break; } + if buf.len() < len { + break; + } let adts = buf.split_to(len).to_vec(); if let Some(aac_au) = strip_adts(&adts) { @@ -364,4 +381,4 @@ pub fn start_audio( } Ok(()) -} \ No newline at end of file +} diff --git a/camera_hub/src/test_camera.rs b/camera_hub/src/test_camera.rs index 68f9795..1fe994d 100644 --- a/camera_hub/src/test_camera.rs +++ b/camera_hub/src/test_camera.rs @@ -2,19 +2,19 @@ //! //! SPDX-License-Identifier: GPL-3.0-or-later -use std::fs::File; -use std::io::{self, Write}; -use rand::Rng; -use anyhow::{Error}; -use crate::motion::MotionResult; use crate::livestream::LivestreamWriter; -use crate::VideoInfo; +use crate::motion::MotionResult; use crate::Camera; -use tokio::time::{sleep, Duration}; +use crate::VideoInfo; +use anyhow::Error; +use image::RgbImage; +use rand::Rng; +use std::fs::File; +use std::io::{self, Write}; use std::thread; use tokio::io::AsyncWriteExt; use tokio::runtime::Runtime; -use image::RgbImage; +use tokio::time::{sleep, Duration}; pub struct TestCamera { pub name: String, @@ -67,7 +67,7 @@ impl Camera for TestCamera { motion: false, detections: vec![], thumbnail: None, - }) + }); } //create dummy thumbnail let width = 256; @@ -81,8 +81,7 @@ impl Camera for TestCamera { data[i] = (i % 256) as u8; } - let img = RgbImage::from_raw(width, height, data) - .expect("Buffer size mismatch"); + let img = RgbImage::from_raw(width, height, data).expect("Buffer size mismatch"); Ok(MotionResult { motion: true, diff --git a/camera_hub/src/traits.rs b/camera_hub/src/traits.rs index 189bfd3..dda0126 100644 --- a/camera_hub/src/traits.rs +++ b/camera_hub/src/traits.rs @@ -33,6 +33,9 @@ pub trait Mp4 { pub trait Camera { fn is_there_motion(&mut self) -> Result; + + // By default, this is a no-op. We override if we have a pipeline-like configuration. + fn set_motion_active(&self, _active: bool) {} fn record_motion_video(&self, info: &VideoInfo, duration: u64) -> io::Result<()>; fn launch_livestream(&self, livestream_writer: LivestreamWriter) -> io::Result<()>; fn get_name(&self) -> String; diff --git a/client_lib/Cargo.toml b/client_lib/Cargo.toml index d6c66b0..28c3fc6 100644 --- a/client_lib/Cargo.toml +++ b/client_lib/Cargo.toml @@ -5,9 +5,12 @@ edition = "2021" authors = ["Ardalan Amiri Sani "] [features] -default = ["logging"] +default = ["logging", "crypto-aws-lc"] logging = ["log"] http_client = ["dep:reqwest", "dep:base64"] +crypto-aws-lc = ["reqwest?/rustls"] +# crypto-ring on armv6 (Pi Zero W) +crypto-ring = ["reqwest?/rustls-no-provider", "dep:rustls", "rustls?/ring"] camera_secret_qrcode = ["dep:qrcode", "dep:image"] [dependencies] @@ -21,7 +24,9 @@ openmls_libcrux_crypto = "=0.3.1" openmls_memory_storage = { version = "=0.5.0", features = ["persistence"] } openmls_basic_credential = "=0.5.0" bincode = "1.3.3" -reqwest = { version = "0.13", default-features = false, features = ["blocking", "multipart", "rustls"], optional = true } +libc = "0.2" +reqwest = { version = "0.13", default-features = false, features = ["blocking", "multipart"], optional = true } +rustls = { version = "0.23", default-features = false, features = ["std", "tls12", "logging"], optional = true } base64 = { version = "0.22.1", optional = true } base64-url = {version = "3.0.3"} anyhow = "^1.0.64" # Locked to this version due to flutter_rust_bridge usage in app diff --git a/client_lib/src/crypto_selftest.rs b/client_lib/src/crypto_selftest.rs new file mode 100644 index 0000000..3058b9d --- /dev/null +++ b/client_lib/src/crypto_selftest.rs @@ -0,0 +1,131 @@ +//! SPDX-License-Identifier: GPL-3.0-or-later +//! +//! Startup diagnostic (microbench) for MLS on constrained hardware (right now, testing the Pi Zero W, ARMv6) +//! Runs the exact crypto the camera hub uses in a tight hot loop in an (otherwise) idle process. +//! Reports wall time, thread CPU time, throughput + +use crate::openmls_rust_persistent_crypto::OpenMlsRustPersistentCrypto; +use anyhow::{anyhow, Context}; +use openmls::prelude::*; +use openmls_basic_credential::SignatureKeyPair; +use openmls_traits::crypto::OpenMlsCrypto; +use openmls_traits::signatures::Signer; +use openmls_traits::types::AeadType; +use openmls_traits::OpenMlsProvider; +use std::time::{Duration, Instant}; + +// Matches mls_client::CIPHERSUITE +const CIPHERSUITE: Ciphersuite = Ciphersuite::MLS_256_XWING_CHACHA20POLY1305_SHA256_Ed25519; +const PAYLOAD_BYTES: usize = 64 * 1024; +const ITERS: usize = 20; + +/// Thread CPU time consumed so far (not wall time). +/// Get the on-core cost, excluding time the scheduler gave to other threads. +pub fn thread_cpu_time() -> anyhow::Result { + let mut ts = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + let rc = unsafe { libc::clock_gettime(libc::CLOCK_THREAD_CPUTIME_ID, &raw mut ts) }; + if rc != 0 { + return Ok(Duration::ZERO); + } + Ok(Duration::new(ts.tv_sec.cast_unsigned().into(), u32::try_from(ts.tv_nsec)?)) +} + +fn report(name: &str, wall: Duration, cpu: Duration, bytes: usize) -> anyhow::Result<()> { + let wall_ms = wall.as_secs_f64() * 1000.0f64 / f64::from(u32::try_from(ITERS)?); + let cpu_ms = cpu.as_secs_f64() * 1000.0f64 / f64::from(u32::try_from(ITERS)?); + let mb = f64::from(u32::try_from(bytes)?) / (1024.0f64 * 1024.0f64); + let mbps = mb / (wall_ms / 1000.0f64); + log::info!("{name} wall={wall_ms}ms cpu={cpu_ms}ms {mbps} MB/s"); + Ok(()) +} + +/// Run the diagnostic at process startup if the caller passes the env var. +pub fn run() -> anyhow::Result<()> { + log::info!("payload={} kb, iters={}", PAYLOAD_BYTES / 1024, ITERS); + + let provider = OpenMlsRustPersistentCrypto::default(); + let signer = SignatureKeyPair::new(CIPHERSUITE.signature_algorithm())?; + signer.store(provider.storage())?; + let credential = BasicCredential::new(b"secluso-selftest".to_vec()); + let credential_with_key = CredentialWithKey { + credential: credential.into(), + signature_key: signer.to_public_vec().into(), + }; + let group_config = MlsGroupCreateConfig::builder() + .ciphersuite(CIPHERSUITE) + .use_ratchet_tree_extension(true) + .build(); + let mut group = MlsGroup::new(&provider, &signer, &group_config, credential_with_key)?; + + let payload = vec![0xABu8; PAYLOAD_BYTES]; + + // Raw ChaCha20Poly1305 seal + { + let key = vec![0u8; 32]; + let nonce = vec![0u8; 12]; + let aad = b"secluso-selftest"; + let _ = + provider + .crypto() + .aead_encrypt(AeadType::ChaCha20Poly1305, &key, &payload, &nonce, aad); + let w = Instant::now(); + let c = thread_cpu_time()?; + for _ in 0..ITERS { + let _ct = provider.crypto().aead_encrypt( + AeadType::ChaCha20Poly1305, + &key, + &payload, + &nonce, + aad, + )?; + } + report( + "aead_seal", + w.elapsed(), + thread_cpu_time()?.checked_sub(c).context("failed to subtract s from thread cpu time")?, + PAYLOAD_BYTES, + )?; + } + + // Ed25519 signature over the full payload + { + let _ = signer.sign(&payload); + let w = Instant::now(); + let c = thread_cpu_time()?; + for _ in 0..ITERS { + // SignerError does not have StdError trait + if let Err(e) = signer.sign(&payload) { + println!("{e:?}"); + return Err(anyhow!("Signer error.")); + } + } + report( + "ed25519_sign", + w.elapsed(), + thread_cpu_time()?.checked_sub(c).context("failed to subtract s from thread cpu time")?, + PAYLOAD_BYTES, + )?; + } + + // MlsGroup::create_message + { + let _ = group.create_message(&provider, &signer, &payload); + let w = Instant::now(); + let c = thread_cpu_time()?; + for _ in 0..ITERS { + let _m = group.create_message(&provider, &signer, &payload)?; + } + report( + "create_message", + w.elapsed(), + thread_cpu_time()?.checked_sub(c).context("failed to subtract s from thread cpu time")?, + PAYLOAD_BYTES, + )?; + } + + log::info!("done"); + Ok(()) +} diff --git a/client_lib/src/http_client.rs b/client_lib/src/http_client.rs index 8717765..f87e980 100644 --- a/client_lib/src/http_client.rs +++ b/client_lib/src/http_client.rs @@ -15,6 +15,15 @@ use std::path::Path; use std::time::Duration; use std::env; +/// Installs the process-wide rustls crypto provider for the ring backend. +#[cfg(feature = "crypto-ring")] +pub fn install_crypto_provider() { + // install_default() errors only if a provider is already installed. + if rustls::crypto::ring::default_provider().install_default().is_err() { + log::warn!("[Pairing] rustls crypto provider was already installed. Expected ring"); + } +} + // Some of these constants are based on the ones in server/main.rs. const MAX_MOTION_FILE_SIZE: u64 = 50 * 1024 * 1024; // 50 mebibytes const MAX_LIVESTREAM_FILE_SIZE: u64 = 20 * 1024 * 1024; // 20 mebibytes diff --git a/client_lib/src/lib.rs b/client_lib/src/lib.rs index e1564bf..b235a0d 100644 --- a/client_lib/src/lib.rs +++ b/client_lib/src/lib.rs @@ -1,6 +1,7 @@ //! SPDX-License-Identifier: GPL-3.0-or-later pub mod config; +pub mod crypto_selftest; pub mod identity; pub mod mls_client; pub mod mls_clients; diff --git a/client_lib/src/mls_client.rs b/client_lib/src/mls_client.rs index 539be0a..7f05d68 100644 --- a/client_lib/src/mls_client.rs +++ b/client_lib/src/mls_client.rs @@ -7,6 +7,7 @@ use super::identity::Identity; use super::openmls_rust_persistent_crypto::OpenMlsRustPersistentCrypto; +use crate::crypto_selftest::thread_cpu_time; use openmls_traits::{storage::StorageProvider as StorageProviderTrait}; use crate::pairing; use openmls::prelude::*; @@ -15,7 +16,7 @@ use serde::{Deserialize, Serialize}; use std::fs::{self, File, OpenOptions}; use std::io; use std::io::{BufRead, BufReader, Write, Read}; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; use std::cmp; use std::path::{Path, PathBuf}; use tls_codec::{Deserialize as TlsDeserialize, Serialize as TlsSerialize}; @@ -852,16 +853,31 @@ impl MlsClient { let group_aad = group.group_name.clone() + " AAD"; group.mls_group.set_aad(group_aad.as_bytes().to_vec()); + let create_start = Instant::now(); + let create_cpu_start = thread_cpu_time().expect("thread cpu time failed"); let message_out = group .mls_group .create_message(&self.provider, &self.identity.signer, bytes) .map_err(|e| io::Error::other(format!("{e}")))?; + let create_cpu_end = thread_cpu_time().expect("thread cpu time failed"); + log::debug!( + "encrypt: create_message took {}ms wall / {}ms cpu for {} input bytes", + create_start.elapsed().as_millis(), + (create_cpu_end - create_cpu_start).as_millis(), + bytes.len() + ); let msg: MlsMessageOut = message_out; + let serialize_start = Instant::now(); let mut msg_vec = Vec::new(); msg.tls_serialize(&mut msg_vec) .map_err(|e| io::Error::other(format!("tls_serialize for msg failed ({e})")))?; + log::debug!( + "encrypt: tls_serialize took {}ms for {} output bytes", + serialize_start.elapsed().as_millis(), + msg_vec.len() + ); Ok(msg_vec) } diff --git a/motion_ai/pipeline/Cargo.toml b/motion_ai/pipeline/Cargo.toml index c4b5405..42b8dc6 100644 --- a/motion_ai/pipeline/Cargo.toml +++ b/motion_ai/pipeline/Cargo.toml @@ -7,6 +7,7 @@ edition = "2024" default = ["replay_backend"] mp4_player = ["dep:video-rs"] replay_backend = ["dep:tokio", "dep:rocket", "dep:walkdir"] +ai = ["dep:ort", "dep:include_dir"] [dependencies] ndarray = { version="=0.17.2", features = ["rayon"] } @@ -16,7 +17,8 @@ serde_json = "1.0.149" serde = { version = "1.0.228", features = ["derive"] } tap = "1.0.1" image = "0.25.10" # Locked version via imageproc -ort = { version = "=2.0.0-rc.12", features = ["ndarray", "load-dynamic", "api-24"], default-features = false } +include_dir = { version = "0.7.4", optional = true } +ort = { version = "=2.0.0-rc.12", features = ["ndarray", "load-dynamic", "api-24"], default-features = false, optional = true } imageproc = "0.25" rusttype = "0.9" # for text rendering thiserror = "2.0.18" @@ -32,8 +34,8 @@ fast_image_resize = { version = "5.5.0", features = ["rayon"]} uuid = { version = "1.22.0", features = ["v4"] } walkdir = { version = "2.5.0", optional = true } tokio = { version = "1.50.0", optional = true } -rocket = { version = "0.5.1", features = ["json"], optional = true } #todo: make this feature based -video-rs= { version = "0.10.5", features = ["ndarray"], optional = true } +rocket = { version = "0.5.1", features = ["json"], optional = true } +video-rs = { version = "0.10.5", features = ["ndarray"], optional = true } crossbeam-channel = "0.5.15" flume = "0.11.1" -include_dir="0.7.4" +cfg-if = "1.0" diff --git a/motion_ai/pipeline/src/backend.rs b/motion_ai/pipeline/src/backend.rs index 88231f6..4412478 100644 --- a/motion_ai/pipeline/src/backend.rs +++ b/motion_ai/pipeline/src/backend.rs @@ -36,7 +36,10 @@ struct FrontEvent { // Normalizes a UUID-like string into the standard lowercase dashed format. fn canon_uuid_like(s: &str) -> String { - let t = s.trim().trim_matches(|c| c == '{' || c == '}').to_lowercase(); + let t = s + .trim() + .trim_matches(|c| c == '{' || c == '}') + .to_lowercase(); // already dashed UUID? if t.len() == 36 && t.as_bytes().get(8) == Some(&b'-') @@ -451,7 +454,10 @@ async fn app_js_route( #[get("/sessions?")] async fn get_sessions(state: &State, q: Option) -> Json { let q = q.unwrap_or_default(); - let limit = q.limit.unwrap_or(DEFAULT_SESSION_LIMIT).clamp(1, MAX_SESSION_LIMIT); + let limit = q + .limit + .unwrap_or(DEFAULT_SESSION_LIMIT) + .clamp(1, MAX_SESSION_LIMIT); let offset = q.offset.unwrap_or(0); let ids = state.session_ids.read().unwrap(); @@ -719,31 +725,37 @@ fn build_series_from_telemetry(path: &Path, tail: Option) -> Result { let ts = as_u128_opt(&v, "ts"); let q = as_usize_any(&v, &["event_queue_len", "queue_len", "queue"]); - let mq = as_usize_any(&v, &["max_event_queue_len", "max_queue_len", "max_queue"]) - .or(q); + let mq = + as_usize_any(&v, &["max_event_queue_len", "max_queue_len", "max_queue"]).or(q); let shf = as_bool_any(&v, &["standby_has_frame", "standby"]).unwrap_or(false); let ahf = as_bool_any(&v, &["active_has_frame", "active"]).unwrap_or(false); if let (Some(ts), Some(q), Some(mq)) = (ts, q, mq) { - push_tail_tick(&mut ticks, SeriesTick { - ts, - queue: q, - max_queue: mq, - standby: shf, - active: ahf, - run: run_key_from_json(&v), - }); + push_tail_tick( + &mut ticks, + SeriesTick { + ts, + queue: q, + max_queue: mq, + standby: shf, + active: ahf, + run: run_key_from_json(&v), + }, + ); } } _ => { /* ignore */ } @@ -906,7 +918,10 @@ fn build_events_from_telemetry(path: &Path, tail: Option) -> (Vec { @@ -923,7 +938,10 @@ fn build_events_from_telemetry(path: &Path, tail: Option) -> (Vec = stats .iter() diff --git a/motion_ai/pipeline/src/frame.rs b/motion_ai/pipeline/src/frame.rs index 5232974..807b1ad 100644 --- a/motion_ai/pipeline/src/frame.rs +++ b/motion_ai/pipeline/src/frame.rs @@ -1,12 +1,17 @@ //! SPDX-License-Identifier: GPL-3.0-or-later use crate::logic::pipeline::RunId; -use crate::ml::models::DetectionType; -use crate::ml::models::{BoxInfo, DetectionResult}; + +cfg_if::cfg_if! { + if #[cfg(feature = "ai")] { + use crate::ml::models::{DetectionType, BoxInfo, DetectionResult}; + use imageproc::{drawing::draw_hollow_rect_mut, rect::Rect}; + use image::Rgb; + } +} + use flume::{Receiver, Sender}; -use image::{GrayImage, Rgb, RgbImage}; -use imageproc::drawing::draw_hollow_rect_mut; -use imageproc::rect::Rect; +use image::{GrayImage, RgbImage}; use log::{debug, warn}; use once_cell::sync::Lazy; use rayon::iter::IndexedParallelIterator; @@ -42,6 +47,7 @@ pub struct RawFrame { pub timestamp: SystemTime, pub width: usize, pub height: usize, + #[cfg(feature = "ai")] pub detection_result: Option, pub dma_aligned: bool, } @@ -184,7 +190,7 @@ fn is_run_rejected(run_id: &str) -> bool { REJECTED_RUNS .read() .ok() - .map_or(false, |set| set.contains(run_id)) + .is_some_and(|set| set.contains(run_id)) } fn is_rejected_path(path: &Path) -> bool { @@ -246,7 +252,7 @@ impl RawFrame { session_id: &str, run_id: &RunId, file_name: &str, - draw_bb: bool, + #[cfg(feature = "ai")] draw_bb: bool, ) -> image::ImageResult { if !SAVE_IMAGES.load(Ordering::Relaxed) { return Ok("".into()); @@ -291,6 +297,7 @@ impl RawFrame { use image::imageops::FilterType; img = image::imageops::resize(&img, 416, 416, FilterType::CatmullRom); + #[cfg(feature = "ai")] if draw_bb && let Some(det) = &self.detection_result { img = self.draw_boxes(img, &det.results) } @@ -331,6 +338,7 @@ impl RawFrame { } /// Maps a label ID to a color using a fixed color palette. Used for drawing bounding boxes. + #[cfg(feature = "ai")] fn get_color_for_label(label: i32) -> [u8; 3] { let color_palette: [[u8; 3]; 5] = [ [255, 0, 0], // Red @@ -344,6 +352,7 @@ impl RawFrame { } /// Draws bounding boxes onto an RGB image for detections that are not classified as 'Other'. + #[cfg(feature = "ai")] fn draw_boxes(&self, mut img: RgbImage, boxes: &Vec) -> RgbImage { let len = boxes.len(); debug!("Drawing {len} boxes"); @@ -419,6 +428,7 @@ impl RawFrame { timestamp: SystemTime::now(), width: actual_width, height: actual_height, + #[cfg(feature = "ai")] detection_result: None, dma_aligned: true, } diff --git a/motion_ai/pipeline/src/lib.rs b/motion_ai/pipeline/src/lib.rs index 3b2facb..93822de 100644 --- a/motion_ai/pipeline/src/lib.rs +++ b/motion_ai/pipeline/src/lib.rs @@ -5,5 +5,6 @@ pub mod backend; mod config; pub mod frame; pub mod logic; +#[cfg(feature = "ai")] pub mod ml; pub mod motion; diff --git a/motion_ai/pipeline/src/logic/context.rs b/motion_ai/pipeline/src/logic/context.rs index fa65ebb..ba48704 100644 --- a/motion_ai/pipeline/src/logic/context.rs +++ b/motion_ai/pipeline/src/logic/context.rs @@ -5,9 +5,11 @@ use crate::logic::activity_states::ActivityState; use crate::logic::health_states::HealthState; use crate::logic::pipeline::{PipelineResult, RunId}; -use crate::ml::models::ModelKind; use crate::motion::detector::MotionDetection; -use std::collections::HashMap; +use std::{collections::HashMap, env}; + +#[cfg(feature = "ai")] +use crate::ml::models::ModelKind; /// Per-run, in-memory state updated by the FSM and stages pub struct StateContext { @@ -15,6 +17,8 @@ pub struct StateContext { pub(crate) activity: ActivityState, /// Coarse health classification (Normal / High Temp / etc) pub(crate) health: HealthState, + + #[cfg(feature = "ai")] /// Currently selected ML model for inference pub(crate) active_model: ModelKind, /// Motion detection accumulator and thresholds. @@ -39,14 +43,17 @@ impl StateContext { Self { activity: ActivityState::Idle, health: HealthState::Normal, + + #[cfg(feature = "ai")] active_model: ModelKind::Accurate, + motion_detection: MotionDetection::new(), run_id: RunId::new(), // Will be replaced with the first frame, so it can be this instead of an Option for ease-of-use //last_motion_time: None, //// latency_ema: 0.0, // temp_history: Default::default(), // backoff_until: None, - use_inference: true, + use_inference: env::var("CARGO_FEATURE_AI").is_ok(), // metadata: Default::default(), stats: Default::default(), last_detection: None, diff --git a/motion_ai/pipeline/src/logic/health_states.rs b/motion_ai/pipeline/src/logic/health_states.rs index 53c1cfb..f1d0d5f 100644 --- a/motion_ai/pipeline/src/logic/health_states.rs +++ b/motion_ai/pipeline/src/logic/health_states.rs @@ -8,7 +8,6 @@ use crate::logic::fsm::{StateHandler, TransitionDecision}; use crate::logic::intent::Intent; use crate::logic::pipeline::PipelineEvent::TemperatureDrop; use crate::logic::pipeline::{Pipeline, PipelineEvent}; -use crate::ml::models::ModelKind; use std::fmt; use crate::logic::telemetry::{TelemetryPacket, TelemetryRun}; @@ -20,6 +19,9 @@ use std::time::{SystemTime, UNIX_EPOCH}; use sysinfo::{Components, CpuRefreshKind, MemoryRefreshKind, RefreshKind, System}; use thiserror::Error; +#[cfg(feature = "ai")] +use crate::ml::models::ModelKind; + /// Represents the health status of the system based on temperature and resource usage. #[derive(Hash, Eq, PartialEq, Clone, Debug, Copy, Serialize, Deserialize)] pub enum HealthState { @@ -132,7 +134,9 @@ impl StateHandler for NormalState { to: HealthState::HighTemp, reason: "temp high".into(), intents: vec![ + #[cfg(feature = "ai")] Intent::SwitchModel(ModelKind::Fast), + #[cfg(feature = "ai")] Intent::AllowInference(true), ], }, @@ -140,7 +144,9 @@ impl StateHandler for NormalState { to: HealthState::ResourceLow, reason: "cpu/ram high".into(), intents: vec![ + #[cfg(feature = "ai")] Intent::SwitchModel(ModelKind::Fast), + #[cfg(feature = "ai")] Intent::AllowInference(true), ], }, @@ -170,7 +176,9 @@ impl StateHandler for HighTempState { to: HealthState::Normal, reason: "cooled below high".into(), intents: vec![ + #[cfg(feature = "ai")] Intent::SwitchModel(ModelKind::Accurate), // <-- back to big model + #[cfg(feature = "ai")] Intent::AllowInference(true), ], }, @@ -195,7 +203,9 @@ impl StateHandler for ResourceLowState { to: HealthState::Normal, reason: "resources normal".into(), intents: vec![ + #[cfg(feature = "ai")] Intent::SwitchModel(ModelKind::Accurate), // <-- restore + #[cfg(feature = "ai")] Intent::AllowInference(true), ], }, diff --git a/motion_ai/pipeline/src/logic/intent.rs b/motion_ai/pipeline/src/logic/intent.rs index ef85f33..f707b12 100644 --- a/motion_ai/pipeline/src/logic/intent.rs +++ b/motion_ai/pipeline/src/logic/intent.rs @@ -5,17 +5,20 @@ use crate::logic::health_states::HealthState; use crate::logic::pipeline::{PipelineEvent, PipelineHostData, RunId}; use crate::logic::stages::{StageResult, StageType}; use crate::logic::telemetry::TelemetryPacket; -use crate::ml::models::ModelKind; use serde::{Deserialize, Serialize}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +#[cfg(feature = "ai")] +use crate::ml::models::ModelKind; + /// Represents actions or commands that can be issued within the pipeline to modify behavior, log transitions, or trigger processing stages. #[derive(Debug, Clone, Serialize, Deserialize)] pub enum Intent { - RunStage(StageType), // Triggers execution of a specific pipeline stage - StartTimer(Duration), // Starts a countdown timer - CancelTimer, // Cancels any running timer - AllowInference(bool), // Enables or disables ML inference + RunStage(StageType), // Triggers execution of a specific pipeline stage + StartTimer(Duration), // Starts a countdown timer + CancelTimer, // Cancels any running timer + AllowInference(bool), // Enables or disables ML inference + #[cfg(feature = "ai")] SwitchModel(ModelKind), // Switches active ML model LogActivity { // Logs an activity state transition @@ -72,6 +75,7 @@ pub(crate) fn execute_intent( host_data.telemetry.run_id.clone().as_str(), &host_data.ctx.run_id, "acceptance", + #[cfg(feature = "ai")] false, )?; } @@ -153,10 +157,16 @@ pub(crate) fn execute_intent( // Enables or disables the inference stage based on resource or health context. Intent::AllowInference(choice) => { + // If it's true while we have AI disabled, we should panic. + #[cfg(feature = "ai")] + if *choice { + panic!("AllowInference cannot be enabled when the AI feature is enabled"); + } host_data.ctx.use_inference = *choice; // todo: check this flag when running InferenceStage to either run the stage or just return a blanket Continue } // Switches the active model and logs the transition based on current health context. + #[cfg(feature = "ai")] Intent::SwitchModel(model) => { let prev_model = host_data.ctx.active_model; host_data.ctx.active_model = *model; diff --git a/motion_ai/pipeline/src/logic/pipeline.rs b/motion_ai/pipeline/src/logic/pipeline.rs index 4d85369..0defeca 100644 --- a/motion_ai/pipeline/src/logic/pipeline.rs +++ b/motion_ai/pipeline/src/logic/pipeline.rs @@ -14,7 +14,7 @@ use crate::logic::intent::{Intent, execute_intent}; use crate::logic::stages::{PipelineStage, StageResult, StageType}; use crate::logic::telemetry::{TelemetryPacket, TelemetryRun}; use crate::logic::timer::{Timer, TimerManager}; -use crate::ml::models::{DetectionType, init_model_paths}; + use anyhow::{Context, Error}; use log::debug; use serde::{Deserialize, Serialize}; @@ -23,6 +23,9 @@ use std::collections::{BTreeMap, BinaryHeap, HashMap, VecDeque}; use std::default::Default; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +#[cfg(feature = "ai")] +use crate::ml::models::{DetectionType, init_model_paths}; + /// The main sequential container for executing image processing stages. /// Each stage handles a specific task (e.g., motion, detection, inference). pub struct Pipeline { @@ -186,6 +189,7 @@ pub struct PipelineController { last_health_change: Option<(HealthState, Instant)>, last_activity_change: Option<(ActivityState, Instant)>, max_event_queue_len: usize, + run_detections: bool, } /// Holds the current active and standby frame references used by the pipeline. @@ -201,6 +205,7 @@ pub struct PipelineHostData { pub pipeline: Pipeline, pub(crate) timer: Box, pub frame_buffer: FrameBuffer, + #[cfg(feature = "ai")] pub latest_detections: Vec, pub telemetry: TelemetryRun, } @@ -210,6 +215,7 @@ pub struct PipelineHostData { pub struct PipelineResult { pub time: Instant, pub motion: bool, + #[cfg(feature = "ai")] pub detections: Vec, pub thumbnail: RawFrame, } @@ -218,7 +224,11 @@ pub struct PipelineResult { /// and reacting to state transitions. impl PipelineController { /// Constructs and initializes the pipeline controller and FSM registries. - pub fn new(pipeline: Pipeline, write_logs: bool, save_all: bool) -> Result { + pub fn new( + pipeline: Pipeline, + write_logs: bool, + save_all: bool, + ) -> Result { let mut activity_registry: FsmRegistry = FsmRegistry { handlers: HashMap::new(), }; @@ -248,6 +258,7 @@ impl PipelineController { health_registry.register(HealthState::CriticalTemp, Box::new(CriticalTempState)); + #[cfg(feature = "ai")] init_model_paths()?; // We should occasionally query this to hot-reload. But for this purpose, initializing and checking everything is OK is good enough Ok(Self { @@ -264,10 +275,17 @@ impl PipelineController { active: None, }, telemetry: TelemetryRun::new(write_logs, save_all)?, + #[cfg(feature = "ai")] latest_detections: Vec::new(), }, last_activity_change: None, max_event_queue_len: 0, + + // By default, it should NOT run detections. + // It should only do so when prompted from the main loop. + // We could be pairing. We might be recording a motion event. We might be livestreaming. + // Either way, we don't need to run unnecessary computations. + run_detections: false, }) } @@ -289,12 +307,24 @@ impl PipelineController { } } + pub fn set_pipeline_active(&mut self, run_detections: bool) { + self.run_detections = run_detections; + } + /// Loads a new frame into the standby buffer and queues a NewFrame event. pub fn push_frame(&mut self, frame: RawFrame) { self.host_data.frame_buffer.standby = Some(frame); // Replace the standby frame with a more recent one. - self.host_data + // Only enqueue a NewFrame if one isn't already pending. Above already replaces with latest, no need for duplicates. + if !self + .host_data .event_queue - .push_back(PipelineEvent::NewFrame); // Should this be an event? We'd only need this to run once per tick, maybe use a boolean field + .iter() + .any(|e| matches!(e, PipelineEvent::NewFrame)) + { + self.host_data + .event_queue + .push_back(PipelineEvent::NewFrame); + } } /// Begins processing by queuing a MotionStart event. @@ -317,6 +347,11 @@ impl PipelineController { /// Main loop to process events, update health/activity FSMs, /// emit telemetry, and dispatch intents. pub fn tick(&mut self, temp_label: &'static str) -> Result { + // We skip if we're not running detections right now. + if !self.run_detections { + return Ok(true) + } + let time = Instant::now(); // Is there a timer event? @@ -335,9 +370,8 @@ impl PipelineController { if let Ok(Some(he)) = health_response { self.host_data.event_queue.push_back(he); - } else if let Err(e) = health_response { - // We should exit. Something's wrong with sensors... - return Err(e); + } else { + health_response?; } self.host_data.event_queue.push_back(PipelineEvent::Tick); @@ -649,10 +683,11 @@ impl Default for PipelineController { /// Macro to concisely build a pipeline using a chained stage definition. #[macro_export] macro_rules! pipeline { - ( $($stage:expr), * $(,)? ) => {{ + ( $( $(#[$meta:meta])* $stage:path ),* $(,)? ) => {{ let mut builder = secluso_motion_ai::logic::pipeline::PipelineBuilder::new(); $( - builder = builder.then($stage); + $(#[$meta])* + { builder = builder.then($stage); } )* builder.build() }}; diff --git a/motion_ai/pipeline/src/logic/stages.rs b/motion_ai/pipeline/src/logic/stages.rs index af5d1f1..cf980ea 100644 --- a/motion_ai/pipeline/src/logic/stages.rs +++ b/motion_ai/pipeline/src/logic/stages.rs @@ -1,21 +1,28 @@ //! SPDX-License-Identifier: GPL-3.0-or-later +cfg_if::cfg_if! { + if #[cfg(feature = "ai")] { + use crate::ml::models::DetectionType; + use std::collections::HashSet; + } +} + use crate::frame::RawFrame; use crate::logic::context::StateContext; use crate::logic::pipeline::PipelineResult; use crate::logic::telemetry::{TelemetryPacket, TelemetryRun}; -use crate::ml::models::DetectionType; use log::debug; use serde::{Deserialize, Serialize}; -use std::collections::HashSet; use std::fmt; -use std::time::{Instant, SystemTime, UNIX_EPOCH}; +use std::time::Instant; +use std::time::{SystemTime, UNIX_EPOCH}; /// Describes the type of stage within the pipeline (e.g., motion, inference). #[derive(PartialEq, Debug, Clone, Serialize, Deserialize, Hash, Eq)] pub enum StageType { Motion, Inference, + EmitFrameIfDetected, Custom(String), } @@ -25,6 +32,7 @@ impl fmt::Display for StageType { match self { StageType::Motion => write!(f, "motion"), StageType::Inference => write!(f, "inference"), + StageType::EmitFrameIfDetected => write!(f, "emit_frame_if_detected"), StageType::Custom(s) => write!(f, "{s}"), } } @@ -109,136 +117,187 @@ impl PipelineStage for MotionStage { } } -/// Performs object detection using the currently active ML model. -pub struct InferenceStage; - -/// Pipeline stage that runs inference and filters based on required labels (e.g., human detection). -impl PipelineStage for InferenceStage { - fn name(&self) -> &'static str { - "inference" - } +cfg_if::cfg_if! { +if #[cfg(feature = "ai")] { + /// Performs object detection using the currently active ML model. + pub struct InferenceStage; - fn kind(&self) -> StageType { - StageType::Inference - } - - fn handle( - &self, - frame: &mut RawFrame, - ctx: &mut StateContext, - telemetry: &mut TelemetryRun, - ) -> Result { - if !ctx.use_inference { - let ts = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_millis(); - telemetry.write(&TelemetryPacket::InferenceSkipped { - run_id: ctx.run_id.clone(), - ts, - reason: "use_inference=false", - })?; - telemetry.reject_run(&ctx.run_id); - return Ok(StageResult::Continue); + /// Pipeline stage that runs inference and filters based on required labels (e.g., human detection). + impl PipelineStage for InferenceStage { + fn name(&self) -> &'static str { + "inference" } - debug!("Inference stage handle called!"); + fn kind(&self) -> StageType { + StageType::Inference + } - // Run the current model and handle inference errors. - let result = match ctx.active_model.run(frame, telemetry, &ctx.run_id) { - Ok(res) => res, - Err(e) => { - log::error!("Model run failed: {e:?}"); + fn handle( + &self, + frame: &mut RawFrame, + ctx: &mut StateContext, + telemetry: &mut TelemetryRun, + ) -> Result { + if !ctx.use_inference { let ts = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() .as_millis(); - telemetry.write(&TelemetryPacket::DroppedFrame { + telemetry.write(&TelemetryPacket::InferenceSkipped { run_id: ctx.run_id.clone(), ts, - reason: "infer_error", + reason: "use_inference=false", })?; telemetry.reject_run(&ctx.run_id); - return Ok(StageResult::Fault("Failed to run model".into())); + return Ok(StageResult::Continue); } - }; - - // Attach detection results to the frame. - frame.detection_result = Some(result.clone()); - - // Save annotated detection frame to disk. - let rel_path = match frame.save_png( - telemetry.run_id.clone().as_str(), - &ctx.run_id, - "det_box", - true, - ) { - Ok(p) => p, - Err(e) => { - log::error!("PNG write error: {e:?}"); - let ts = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_millis(); - telemetry.write(&TelemetryPacket::DroppedFrame { - run_id: ctx.run_id.clone(), - ts, - reason: "io_error:save_png", - })?; - telemetry.reject_run(&ctx.run_id); - return Ok(StageResult::Fault("Failed to write image".into())); + + debug!("Inference stage handle called!"); + + // Run the current model and handle inference errors. + let result = match ctx.active_model.run(frame, telemetry, &ctx.run_id) { + Ok(res) => res, + Err(e) => { + log::error!("Model run failed: {e:?}"); + let ts = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis(); + telemetry.write(&TelemetryPacket::DroppedFrame { + run_id: ctx.run_id.clone(), + ts, + reason: "infer_error", + })?; + telemetry.reject_run(&ctx.run_id); + return Ok(StageResult::Fault("Failed to run model".into())); + } + }; + + // Attach detection results to the frame. + frame.detection_result = Some(result.clone()); + + // Save annotated detection frame to disk. + let rel_path = match frame.save_png( + telemetry.run_id.clone().as_str(), + &ctx.run_id, + "det_box", + true, + ) { + Ok(p) => p, + Err(e) => { + log::error!("PNG write error: {e:?}"); + let ts = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis(); + telemetry.write(&TelemetryPacket::DroppedFrame { + run_id: ctx.run_id.clone(), + ts, + reason: "io_error:save_png", + })?; + telemetry.reject_run(&ctx.run_id); + return Ok(StageResult::Fault("Failed to write image".into())); + } + }; + + let pkt = TelemetryPacket::Detection { + run_id: ctx.run_id.clone(), + frame_rel: rel_path.as_str(), + detections: result.results.len(), + latency_ms: result.runtime.as_millis() as u32, + ts: SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("Time should go forward.") + .as_millis(), + }; + + if let Err(e) = telemetry.write(&pkt) { + log::error!("Telemetry write error: {e:?}"); + telemetry.reject_run(&ctx.run_id); + return Ok(StageResult::Fault("Failed to write telemetry".into())); + } + + // TODO: Make this adjustable. + const REQUIRED_LABEL: DetectionType = DetectionType::Human; + if result.results.iter().any(|b| b.det_type == REQUIRED_LABEL) { + let mut detection_results = HashSet::new(); + for box_data in result.results { + match box_data.det_type { + DetectionType::Human | DetectionType::Car | DetectionType::Animal => { + detection_results.insert(box_data.det_type); + } + _ => {} + } + } + + debug!("Updating detection results: {}", detection_results.len()); + telemetry.approve_run(&ctx.run_id); + Ok(StageResult::Continue) + } else { + let ts = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis(); + telemetry.write(&TelemetryPacket::DroppedFrame { + run_id: ctx.run_id.clone(), + ts, + reason: "no_human", + })?; + telemetry.reject_run(&ctx.run_id); + Ok(StageResult::Drop("no human detected".into())) + } } - }; - - let pkt = TelemetryPacket::Detection { - run_id: ctx.run_id.clone(), - frame_rel: rel_path.as_str(), - detections: result.results.len(), - latency_ms: result.runtime.as_millis() as u32, - ts: SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("Time should go forward.") - .as_millis(), - }; - - if let Err(e) = telemetry.write(&pkt) { - log::error!("Telemetry write error: {e:?}"); - telemetry.reject_run(&ctx.run_id); - return Ok(StageResult::Fault("Failed to write telemetry".into())); } + } +} + +// TODO: Make an end stage that saves the results. Detects if only motion or inference too +// Then sets the detections accordingly and the last_detected_frame - const REQUIRED_LABEL: DetectionType = DetectionType::Human; - if result.results.iter().any(|b| b.det_type == REQUIRED_LABEL) { - let mut detection_results = HashSet::new(); - for box_data in result.results { - match box_data.det_type { - DetectionType::Human | DetectionType::Car | DetectionType::Animal => { - detection_results.insert(box_data.det_type); +pub struct EmitFrameIfDetected; + +impl PipelineStage for EmitFrameIfDetected { + fn name(&self) -> &'static str { + "emit_frame_if_detected" + } + + fn kind(&self) -> StageType { + StageType::EmitFrameIfDetected + } + + fn handle( + &self, + frame: &mut RawFrame, + ctx: &mut StateContext, + _telemetry: &mut TelemetryRun, + ) -> Result { + // If this stage is reached, that means that it passed detection. + // Otherwise, it would be StageResult::Drop. + + cfg_if::cfg_if! { + if #[cfg(feature = "ai")] { + let mut detection_results = HashSet::new(); + if let Some(result) = &frame.detection_result { + for box_data in &result.results { + match box_data.det_type { + DetectionType::Human | DetectionType::Car | DetectionType::Animal => { + detection_results.insert(box_data.det_type.clone()); + } + _ => {} + } } - _ => {} } } - ctx.last_detection = Some(PipelineResult { - time: Instant::now(), - motion: true, - detections: detection_results.clone().into_iter().collect(), - thumbnail: frame.clone(), - }); - debug!("Updating detection results: {}", detection_results.len()); - telemetry.approve_run(&ctx.run_id); - Ok(StageResult::Continue) - } else { - let ts = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_millis(); - telemetry.write(&TelemetryPacket::DroppedFrame { - run_id: ctx.run_id.clone(), - ts, - reason: "no_human", - })?; - telemetry.reject_run(&ctx.run_id); - Ok(StageResult::Drop("no human detected".into())) } + + ctx.last_detection = Some(PipelineResult { + time: Instant::now(), + motion: true, + #[cfg(feature = "ai")] + detections: detection_results.into_iter().collect(), + thumbnail: frame.clone(), + }); + + Ok(StageResult::Continue) } }