[wip] Copper example

examples/copper_pipeline/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "copper"
+authors.workspace = true
+categories.workspace = true
+description.workspace = true
+edition.workspace = true
+homepage.workspace = true
+include.workspace = true
+license.workspace = true
+readme.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+bincode = "2.0.0-rc.3"
+cu29 = { git = "https://github.com/copper-project/copper-rs.git", branch = "gbin/override_copperlist_stream_allocation_size" }
+cu29-helpers = { git = "https://github.com/copper-project/copper-rs.git", branch = "gbin/override_copperlist_stream_allocation_size" }
+kornia = { workspace = true, features = ["gstreamer"] }
+rerun = { workspace = true }
+tempfile = "3.14"

examples/copper_pipeline/build.rs

@@ -0,0 +1,6 @@
+fn main() {
+    println!(
+        "cargo:rustc-env=LOG_INDEX_DIR={}",
+        std::env::var("OUT_DIR").unwrap()
+    );
+}

examples/copper_pipeline/kornia_app.ron

@@ -0,0 +1,39 @@
+(
+    tasks: [
+        (
+            id: "cam",
+            type: "copper::tasks::webcam::Webcam",
+            config: {
+                "camera_id": 0,
+                "res_rows": 480,
+                "res_cols": 640,
+                "fps": 30,
+            },
+        ),
+        (
+            id: "cam_garden",
+            type: "copper::tasks::rtsp::RtspCamera",
+            config: {
+                "url": "rtsp://tapo_entrance:[email protected]:554/stream2",
+            },
+        ),
+        (
+            id: "rerun",
+            type: "copper::tasks::rerun_viz::RerunViz",
+        ),
+        (
+            id: "sobel",
+            type: "copper::tasks::sobel::Sobel",
+        ),
+    ],
+    cnx: [
+        (src: "cam_garden", dst: "rerun", msg: "copper::tasks::ImageRGBU8Msg"),
+        (src: "cam", dst: "rerun", msg: "copper::tasks::ImageRGBU8Msg"),
+        (src: "sobel", dst: "rerun", msg: "copper::tasks::ImageGrayU8Msg"),
+        (src: "cam", dst: "sobel", msg: "copper::tasks::ImageRGBU8Msg"),
+    ],
+    logging: (
+        slab_size_mib: 1024, // Preallocates 1GiB of memory map file at a time
+        section_size_mib: 100, // Preallocates 100MiB of memory map per section for the main logger.
+    ),
+)

examples/copper_pipeline/src/lib.rs

@@ -0,0 +1 @@
+pub mod tasks;

examples/copper_pipeline/src/main.rs

@@ -0,0 +1,35 @@
+use cu29::prelude::*;
+use cu29_helpers::basic_copper_setup;
+use tempfile::tempdir;
+
+const SLAB_SIZE: Option<usize> = Some(150 * 1024 * 1024);
+
+#[copper_runtime(config = "kornia_app.ron")]
+struct KorniaApplication {}
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let logger_path = "kornia_app.copper";
+    let dir = tempdir().expect("could not create a tmp dir");
+    let logger_path = dir.path().join(logger_path);
+    let copper_ctx =
+        basic_copper_setup(&logger_path, SLAB_SIZE, true, None).expect("Failed to setup copper.");
+
+    debug!("Logger path: {}", path = &logger_path);
+
+    let clock = copper_ctx.clock;
+
+    let mut application = KorniaApplication::new(clock.clone(), copper_ctx.unified_logger.clone())
+        .expect("Failed to create application.");
+
+    debug!("Running... starting clock: {}.", clock.now());
+
+    application.start_all_tasks()?;
+
+    application.run()?;
+
+    application.stop_all_tasks()?;
+
+    debug!("End of program: {}.", clock.now());
+
+    Ok(())
+}

examples/copper_pipeline/src/tasks/cu_image.rs

@@ -0,0 +1,90 @@
+type ImageRGBU8 = kornia::image::Image<u8, 3>;
+type ImageGrayU8 = kornia::image::Image<u8, 1>;
+
+#[derive(Clone)]
+pub struct ImageRGBU8Msg {
+    pub image: ImageRGBU8,
+}
+
+impl std::fmt::Debug for ImageRGBU8Msg {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "ImageRGBU8Msg(size: {:?})", self.image.size())
+    }
+}
+
+impl Default for ImageRGBU8Msg {
+    fn default() -> Self {
+        Self {
+            image: ImageRGBU8::new([0, 0].into(), vec![]).unwrap(),
+        }
+    }
+}
+
+impl bincode::enc::Encode for ImageRGBU8Msg {
+    fn encode<E: bincode::enc::Encoder>(
+        &self,
+        encoder: &mut E,
+    ) -> Result<(), bincode::error::EncodeError> {
+        bincode::Encode::encode(&self.image.rows(), encoder)?;
+        bincode::Encode::encode(&self.image.cols(), encoder)?;
+        bincode::Encode::encode(&self.image.as_slice(), encoder)?;
+        Ok(())
+    }
+}
+
+impl bincode::de::Decode for ImageRGBU8Msg {
+    fn decode<D: bincode::de::Decoder>(
+        decoder: &mut D,
+    ) -> Result<Self, bincode::error::DecodeError> {
+        let rows = bincode::Decode::decode(decoder)?;
+        let cols = bincode::Decode::decode(decoder)?;
+        let data = bincode::Decode::decode(decoder)?;
+        let image = ImageRGBU8::new([rows, cols].into(), data)
+            .map_err(|e| bincode::error::DecodeError::OtherString(e.to_string()))?;
+        Ok(Self { image })
+    }
+}
+
+#[derive(Clone)]
+pub struct ImageGrayU8Msg {
+    pub image: ImageGrayU8,
+}
+
+impl std::fmt::Debug for ImageGrayU8Msg {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "ImageGrayU8Msg(size: {:?})", self.image.size())
+    }
+}
+
+impl Default for ImageGrayU8Msg {
+    fn default() -> Self {
+        Self {
+            image: ImageGrayU8::new([0, 0].into(), vec![]).unwrap(),
+        }
+    }
+}
+
+impl bincode::enc::Encode for ImageGrayU8Msg {
+    fn encode<E: bincode::enc::Encoder>(
+        &self,
+        encoder: &mut E,
+    ) -> Result<(), bincode::error::EncodeError> {
+        bincode::Encode::encode(&self.image.rows(), encoder)?;
+        bincode::Encode::encode(&self.image.cols(), encoder)?;
+        bincode::Encode::encode(&self.image.as_slice(), encoder)?;
+        Ok(())
+    }
+}
+
+impl bincode::de::Decode for ImageGrayU8Msg {
+    fn decode<D: bincode::de::Decoder>(
+        decoder: &mut D,
+    ) -> Result<Self, bincode::error::DecodeError> {
+        let rows = bincode::Decode::decode(decoder)?;
+        let cols = bincode::Decode::decode(decoder)?;
+        let data = bincode::Decode::decode(decoder)?;
+        let image = ImageGrayU8::new([rows, cols].into(), data)
+            .map_err(|e| bincode::error::DecodeError::OtherString(e.to_string()))?;
+        Ok(Self { image })
+    }
+}

examples/copper_pipeline/src/tasks/mod.rs

@@ -0,0 +1,7 @@
+pub mod rerun_viz;
+pub mod rtsp;
+pub mod sobel;
+pub mod webcam;
+
+mod cu_image;
+pub use cu_image::{ImageGrayU8Msg, ImageRGBU8Msg};

examples/copper_pipeline/src/tasks/rerun_viz.rs

@@ -0,0 +1,76 @@
+use cu29::prelude::*;
+
+use super::cu_image::{ImageGrayU8Msg, ImageRGBU8Msg};
+
+pub struct RerunViz {
+    rec: rerun::RecordingStream,
+}
+
+impl Freezable for RerunViz {}
+
+impl<'cl> CuSinkTask<'cl> for RerunViz {
+    type Input = input_msg!('cl, ImageRGBU8Msg, ImageRGBU8Msg, ImageGrayU8Msg);
+
+    fn new(_config: Option<&ComponentConfig>) -> Result<Self, CuError>
+    where
+        Self: Sized,
+    {
+        Ok(Self {
+            rec: rerun::RecordingStreamBuilder::new("kornia_app")
+                .spawn()
+                .map_err(|e| CuError::new_with_cause("Failed to spawn rerun stream", e))?,
+        })
+    }
+
+    fn process(&mut self, _clock: &RobotClock, input: Self::Input) -> Result<(), CuError> {
+        let (img1, img2, img3) = input;
+
+        if let Some(img) = img1.payload() {
+            log_image_rgb(&self.rec, "webcam", img)?;
+        }
+
+        if let Some(img) = img2.payload() {
+            log_image_gray(&self.rec, "sobel", img)?;
+        }
+
+        if let Some(img) = img3.payload() {
+            log_image_rgb(&self.rec, "rtsp", img)?;
+        }
+
+        Ok(())
+    }
+}
+
+fn log_image_rgb(
+    rec: &rerun::RecordingStream,
+    name: &str,
+    img: &ImageRGBU8Msg,
+) -> Result<(), CuError> {
+    rec.log(
+        name,
+        &rerun::Image::from_elements(
+            img.image.as_slice(),
+            img.image.size().into(),
+            rerun::ColorModel::RGB,
+        ),
+    )
+    .map_err(|e| CuError::new_with_cause("Failed to log image", e))?;
+    Ok(())
+}
+
+fn log_image_gray(
+    rec: &rerun::RecordingStream,
+    name: &str,
+    img: &ImageGrayU8Msg,
+) -> Result<(), CuError> {
+    rec.log(
+        name,
+        &rerun::Image::from_elements(
+            img.image.as_slice(),
+            img.image.size().into(),
+            rerun::ColorModel::L,
+        ),
+    )
+    .map_err(|e| CuError::new_with_cause("Failed to log image", e))?;
+    Ok(())
+}

examples/copper_pipeline/src/tasks/rtsp.rs

@@ -0,0 +1,63 @@
+use cu29::prelude::*;
+use kornia::io::stream::{CameraCapture, RTSPCameraConfig};
+
+use super::cu_image::ImageRGBU8Msg;
+
+// default config for the rtsp camera
+const DEFAULT_URL: &str = "rtsp://admin:[email protected]:554/Streaming/Channels/1";
+
+pub struct RtspCamera {
+    cam: CameraCapture,
+}
+
+impl Freezable for RtspCamera {}
+
+impl<'cl> CuSrcTask<'cl> for RtspCamera {
+    type Output = output_msg!('cl, ImageRGBU8Msg);
+
+    fn new(config: Option<&ComponentConfig>) -> Result<Self, CuError>
+    where
+        Self: Sized,
+    {
+        let url = if let Some(config) = config {
+            config
+                .get::<String>("url")
+                .unwrap_or(DEFAULT_URL.to_string())
+        } else {
+            DEFAULT_URL.to_string()
+        };
+
+        let cam = RTSPCameraConfig::new()
+            .with_url(&url)
+            .build()
+            .map_err(|e| CuError::new_with_cause("Failed to build camera", e))?;
+
+        Ok(Self { cam })
+    }
+
+    fn start(&mut self, _clock: &RobotClock) -> Result<(), CuError> {
+        self.cam
+            .start()
+            .map_err(|e| CuError::new_with_cause("Failed to start camera", e))
+    }
+
+    fn stop(&mut self, _clock: &RobotClock) -> Result<(), CuError> {
+        self.cam
+            .close()
+            .map_err(|e| CuError::new_with_cause("Failed to stop camera", e))
+    }
+
+    fn process(&mut self, _clock: &RobotClock, output: Self::Output) -> Result<(), CuError> {
+        let Some(img) = self
+            .cam
+            .grab()
+            .map_err(|e| CuError::new_with_cause("Failed to grab image", e))?
+        else {
+            return Ok(());
+        };
+
+        output.set_payload(ImageRGBU8Msg { image: img });
+
+        Ok(())
+    }
+}