feat: Image decoder in the frontend

.github/workflows/container-validation-dynamo.yml

@@ -26,6 +26,8 @@ jobs:
     steps:
       - name: Checkout repository
         uses: actions/checkout@v4
+        with:
+          lfs: true
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
       - name: Login to NGC

.github/workflows/pre-merge-rust.yml

@@ -44,6 +44,8 @@ jobs:
       contents: read
     steps:
     - uses: actions/checkout@v4
+      with:
+        lfs: true
     - name: Set up system dependencies
       run: |
         # Install protoc for Rust build dependencies (NOTE: much faster than apt install)
@@ -94,6 +96,8 @@ jobs:
       contents: read
     steps:
     - uses: actions/checkout@v4
+      with:
+        lfs: true
     - name: Set up system dependencies
       run: |
         # Install protoc for Rust build dependencies (NOTE: much faster than apt install)

lib/llm/Cargo.toml

@@ -21,7 +21,7 @@ testing-full = ["testing-cuda", "testing-nixl"]
 testing-cuda = ["dep:cudarc"]
 testing-nixl = ["dep:nixl-sys"]
 testing-etcd = []
-block-manager = ["dep:nixl-sys", "dep:cudarc", "dep:ndarray", "dep:nix", "dep:aligned-vec"]
+block-manager = ["dep:nixl-sys", "dep:cudarc", "dep:nix", "dep:aligned-vec"]
 cuda = ["dep:cudarc"]
 integration = ["dynamo-runtime/integration"]
 
@@ -97,7 +97,6 @@ dialoguer = { version = "0.11", default-features = false, features = [
 aligned-vec = { version = "0.6.4", optional = true }
 nixl-sys = { version = "=0.7.0", optional = true }
 cudarc = { workspace = true, optional = true }
-ndarray = { version = "0.16", optional = true }
 nix = { version = "0.26", optional = true }
 
 
@@ -143,6 +142,9 @@ json-five = { version = "0.3" }
 # media loading in the preprocessor
 reqwest = { workspace = true }
 base64 = { version = "0.22" }
+image = { version = "0.25" }
+tokio-rayon = {version = "2" }
+ndarray = { version = "0.16" }
 
 # Publishers
 zeromq = "0.4.1"

lib/llm/src/preprocessor.rs

@@ -330,7 +330,7 @@ impl OpenAIPreprocessor {
             let _results = futures::future::join_all(
                 fetch_tasks
                     .iter()
-                    .map(|(_, content_part)| loader.fetch_media_part(content_part)),
+                    .map(|(_, content_part)| loader.fetch_and_decode_media_part(content_part)),
             )
             .await;
 

lib/llm/src/preprocessor/media.rs

@@ -2,7 +2,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 mod common;
+mod decoders;
 mod loader;
 
 pub use common::EncodedMediaData;
+pub use decoders::{Decoder, ImageDecoder, MediaDecoder};
 pub use loader::MediaLoader;

lib/llm/src/preprocessor/media/decoders.rs

@@ -0,0 +1,69 @@
+// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use anyhow::Result;
+
+use super::common::EncodedMediaData;
+use ndarray::{ArrayBase, Dimension, OwnedRepr};
+mod image;
+
+pub use image::{ImageDecoder, ImageMetadata};
+
+#[derive(Debug)]
+pub enum DecodedMediaMetadata {
+    #[allow(dead_code)] // used in followup MR
+    Image(ImageMetadata),
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub enum DataType {
+    UINT8,
+}
+
+// Decoded media data (image RGB, video frames pixels, ...)
+#[derive(Debug)]
+pub struct DecodedMediaData {
+    #[allow(dead_code)] // used in followup MR
+    pub(crate) data: Vec<u8>,
+    #[allow(dead_code)] // used in followup MR
+    pub(crate) shape: Vec<usize>,
+    #[allow(dead_code)] // used in followup MR
+    pub(crate) dtype: DataType,
+    #[allow(dead_code)] // used in followup MR
+    pub(crate) metadata: Option<DecodedMediaMetadata>,
+}
+
+// convert Array{N}<u8> to DecodedMediaData
+// TODO: Array1<f32> for audio
+impl<D: Dimension> From<ArrayBase<OwnedRepr<u8>, D>> for DecodedMediaData {
+    fn from(array: ArrayBase<OwnedRepr<u8>, D>) -> Self {
+        let shape = array.shape().to_vec();
+        let (data, _) = array.into_raw_vec_and_offset();
+        Self {
+            data,
+            shape,
+            dtype: DataType::UINT8,
+            metadata: None,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+pub trait Decoder: Clone + Send + 'static {
+    fn decode(&self, data: EncodedMediaData) -> Result<DecodedMediaData>;
+
+    async fn decode_async(&self, data: EncodedMediaData) -> Result<DecodedMediaData> {
+        // light clone (only config params)
+        let decoder = self.clone();
+        // compute heavy -> rayon
+        let result = tokio_rayon::spawn(move || decoder.decode(data)).await?;
+        Ok(result)
+    }
+}
+
+#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
+pub struct MediaDecoder {
+    #[serde(default)]
+    pub image_decoder: ImageDecoder,
+    // TODO: video, audio decoders
+}