Enabling mono16 depth frame

binaries/daemon/src/lib.rs

@@ -251,10 +251,16 @@ impl Daemon {
             None => None,
         };
 
-        let mut zenoh_config = match std::env::var(zenoh::Config::DEFAULT_CONFIG_PATH_ENV) {
-            Ok(path) => zenoh::Config::from_file(&path)
-                .map_err(|e| eyre!(e))
-                .wrap_err_with(|| format!("failed to read zenoh config from {path}"))?,
+        let zenoh_session = match std::env::var(zenoh::Config::DEFAULT_CONFIG_PATH_ENV) {
+            Ok(path) => {
+                let zenoh_config = zenoh::Config::from_file(&path)
+                    .map_err(|e| eyre!(e))
+                    .wrap_err_with(|| format!("failed to read zenoh config from {path}"))?;
+                zenoh::open(zenoh_config)
+                    .await
+                    .map_err(|e| eyre!(e))
+                    .context("failed to open zenoh session")?
+            }
             Err(std::env::VarError::NotPresent) => {
                 let mut zenoh_config = zenoh::Config::default();
 
@@ -266,24 +272,36 @@ impl Daemon {
                         )
                         .unwrap();
                     zenoh_config
-                .insert_json5(
-                    "listen/endpoints",
-                    r#"{ router: ["tcp/[::]:7447"], peer: ["tcp/[::]:0", "tcp/[::]:5456"] }"#,
-                )
-                .unwrap();
+                        .insert_json5(
+                            "listen/endpoints",
+                            r#"{ router: ["tcp/[::]:7447"], peer: ["tcp/[::]:5456"] }"#,
+                        )
+                        .unwrap();
+                    zenoh_config
+                        .insert_json5("routing/peer", r#"{ mode: "linkstate" }"#)
+                        .unwrap();
+                    if cfg!(not(target_os = "linux")) {
+                        warn!("disabling multicast on non-linux systems. Enable it with the ZENOH_CONFIG env variable or file");
+                        zenoh_config
+                            .insert_json5("scouting/multicast", r#"{ enabled: false }"#)
+                            .unwrap();
+                    }
                 };
-                zenoh_config
+                if let Ok(zenoh_session) = zenoh::open(zenoh_config).await {
+                    zenoh_session
+                } else {
+                    warn!("failed to open zenoh session, retrying with default config");
+                    zenoh::open(zenoh::Config::default())
+                        .await
+                        .map_err(|e| eyre!(e))
+                        .wrap_err("failed to open zenoh session")?
+                }
             }
             Err(std::env::VarError::NotUnicode(_)) => eyre::bail!(
                 "{} env variable is not valid unicode",
                 zenoh::Config::DEFAULT_CONFIG_PATH_ENV
             ),
         };
-        let zenoh_session = zenoh::open(zenoh_config)
-            .await
-            .map_err(|e| eyre!(e))
-            .context("failed to open zenoh session")?;
-
         let (dora_events_tx, dora_events_rx) = mpsc::channel(5);
         let daemon = Self {
             logger: Logger {

node-hub/dora-dav1d/Cargo.toml

@@ -11,3 +11,4 @@ log = "0.4"
 structopt = "0.3"
 dora-node-api = { workspace = true, features = ["tracing"] }
 eyre = "0.6.8"
+bytemuck = "1.7.0"

node-hub/dora-dav1d/dataflow.yml

@@ -9,42 +9,29 @@ nodes:
     outputs:
       - image
     env:
-      CAPTURE_PATH: 0
+      CAPTURE_PATH: 1
       IMAGE_WIDTH: 640
       IMAGE_HEIGHT: 480
-      ENCODING: yuv420
-  - id: camera2
-    path: dora-pyrealsense
-    _unstable_deploy:
-      machine: encoder
-    inputs:
-      tick: dora/timer/millis/10
-    outputs:
-      - image
-    env:
-      IMAGE_WIDTH: 640
-      IMAGE_HEIGHT: 480
-      ENCODING: jpeg
 
   - id: echo
     build: pip install -e ../../node-hub/dora-echo
     path: dora-echo
     _unstable_deploy:
       machine: decoder
-    inputs:
-      image: camera2/image
+    #inputs:
+    #  image: camera/image
     outputs:
       - image
 
   - id: rav1e-local
-    path: dora-av1-encoder
-    build: cargo build -p dora-av1-encoder --release
+    path: dora-rav1e
+    build: cargo build -p dora-rav1e --release
     _unstable_deploy:
       machine: encoder
-    #inputs:
-    #  image: camera/image
+    inputs:
+      image: camera/image
     outputs:
-      - frame
+      - image
     env:
       IMAGE_WIDTH: 640
       IMAGE_HEIGHT: 480
@@ -55,22 +42,22 @@ nodes:
     _unstable_deploy:
       machine: decoder
     inputs:
-      image: av-local/frame
+      image: rav1e-local/image
     outputs:
-      - frame
+      - image
     env:
       IMAGE_WIDTH: 640
       IMAGE_HEIGHT: 480
 
   - id: rav1e-remote
-    path: dora-av1-encoder
-    build: cargo build -p dora-av1-encoder --release
+    path: dora-rav1e
+    build: cargo build -p dora-rav1e --release
     _unstable_deploy:
       machine: decoder
     inputs:
       image: dav1d-remote/image
     outputs:
-      - frame
+      - image
     env:
       IMAGE_WIDTH: 640
       IMAGE_HEIGHT: 480
@@ -79,11 +66,11 @@ nodes:
     path: dora-dav1d
     build: cargo build -p dora-dav1d --release
     _unstable_deploy:
-      machine: decoder
+      machine: encoder
     inputs:
-      image: rav1e-local/frame
+      image: rav1e-remote/image
     outputs:
-      - frame
+      - image
     env:
       IMAGE_WIDTH: 640
       IMAGE_HEIGHT: 480
@@ -94,14 +81,14 @@ nodes:
       machine: encoder
     path: dora-rerun
     inputs:
-      image: dav1d-local/frame
-    # image: echo/image
+      image_decode: dav1d-local/image
+      image_echo: echo/image
     #image_2: camera6/image
 
-  - id: plot2
-    build: pip install -e ../../node-hub/opencv-plot
-    _unstable_deploy:
-      machine: encoder
-    path: opencv-plot
-    inputs:
-      image: echo/image
+  # - id: plot2
+  # build: pip install -e ../../node-hub/opencv-plot
+  # _unstable_deploy:
+  # machine: encoder
+  # path: opencv-plot
+  # inputs:
+  # image: echo/image

node-hub/dora-dav1d/src/main.rs

@@ -41,7 +41,6 @@ fn main() -> Result<()> {
     let mut settings = Settings::new();
     // settings.set_n_threads(16);
     settings.set_max_frame_delay(1);
-
     let mut dec =
         dav1d::Decoder::with_settings(&settings).expect("failed to create decoder instance");
 
@@ -78,17 +77,34 @@ fn main() -> Result<()> {
                             } else {
                                 480
                             };
-                            let y = p.plane(dav1d::PlanarImageComponent::Y);
-                            let u = p.plane(dav1d::PlanarImageComponent::U);
-                            let v = p.plane(dav1d::PlanarImageComponent::V);
-                            let y = yuv420_to_bgr(&y, &u, &v, width, height);
-
-                            let arrow = y.into_arrow();
-                            metadata.parameters.insert(
-                                "encoding".to_string(),
-                                dora_node_api::Parameter::String("bgr8".to_string()),
-                            );
-                            node.send_output(id, metadata.parameters, arrow).unwrap();
+                            match p.pixel_layout() {
+                                dav1d::PixelLayout::I420 => {
+                                    let y = p.plane(dav1d::PlanarImageComponent::Y);
+                                    let u = p.plane(dav1d::PlanarImageComponent::U);
+                                    let v = p.plane(dav1d::PlanarImageComponent::V);
+                                    let y = yuv420_to_bgr(&y, &u, &v, width, height);
+                                    let arrow = y.into_arrow();
+                                    metadata.parameters.insert(
+                                        "encoding".to_string(),
+                                        dora_node_api::Parameter::String("bgr8".to_string()),
+                                    );
+                                    node.send_output(id, metadata.parameters, arrow).unwrap();
+                                }
+                                dav1d::PixelLayout::I400 => {
+                                    let y = p.plane(dav1d::PlanarImageComponent::Y);
+                                    let vec16: Vec<u16> = bytemuck::cast_slice(&y).to_vec();
+                                    let arrow = vec16.into_arrow();
+                                    metadata.parameters.insert(
+                                        "encoding".to_string(),
+                                        dora_node_api::Parameter::String("mono16".to_string()),
+                                    );
+                                    node.send_output(id, metadata.parameters, arrow).unwrap();
+                                }
+                                _ => {
+                                    warn!("Unsupported pixel layout");
+                                    continue;
+                                }
+                            };
                         }
                     }
                 }

node-hub/dora-rav1e/src/main.rs

@@ -9,7 +9,7 @@
 
 use std::env::var;
 
-use dora_node_api::arrow::array::UInt8Array;
+use dora_node_api::arrow::array::{UInt16Array, UInt8Array};
 use dora_node_api::{DoraNode, Event, IntoArrow, Parameter};
 use eyre::{Context as EyreContext, Result};
 use log::warn;
@@ -119,6 +119,14 @@ fn main() -> Result<()> {
                     low_latency: true,
                     ..Default::default()
                 };
+                match encoding {
+                    "mono16" => {
+                        enc.bit_depth = 12;
+                        enc.chroma_sampling = color::ChromaSampling::Cs400;
+                    }
+                    _ => {}
+                }
+
                 let cfg = Config::new().with_encoder_config(enc.clone());
                 if encoding == "bgr8" {
                     let buffer: &UInt8Array = data.as_any().downcast_ref().unwrap();
@@ -192,6 +200,48 @@ fn main() -> Result<()> {
                     let stride = (width + xdec) >> xdec;
                     f.planes[2].copy_from_raw_u8(&v, stride, 1);
 
+                    match ctx.send_frame(f) {
+                        Ok(_) => {}
+                        Err(e) => match e {
+                            EncoderStatus::EnoughData => {
+                                warn!("Unable to send frame ");
+                            }
+                            _ => {
+                                warn!("Unable to send frame ");
+                            }
+                        },
+                    }
+                    metadata
+                        .parameters
+                        .insert("encoding".to_string(), Parameter::String("av1".to_string()));
+                    ctx.flush();
+                    match ctx.receive_packet() {
+                        Ok(pkt) => {
+                            let data = pkt.data;
+                            let arrow = data.into_arrow();
+                            node.send_output(id, metadata.parameters, arrow)
+                                .context("could not send output")
+                                .unwrap();
+                        }
+                        Err(e) => match e {
+                            EncoderStatus::LimitReached => {}
+                            EncoderStatus::Encoded => {}
+                            EncoderStatus::NeedMoreData => {}
+                            _ => {
+                                panic!("Unable to receive packet",);
+                            }
+                        },
+                    }
+                    vec![]
+                } else if encoding == "mono16" {
+                    let buffer: &UInt16Array = data.as_any().downcast_ref().unwrap();
+                    let buffer: &[u16] = buffer.values();
+                    let mut ctx: Context<u16> = cfg.new_context().unwrap();
+                    let mut f = ctx.new_frame();
+
+                    let origin = f.planes[0].data_origin_mut();
+                    origin.copy_from_slice(buffer);
+
                     match ctx.send_frame(f) {
                         Ok(_) => {}
                         Err(e) => match e {