Add experimental send_recording python api (#9148)

### Related

* Resolves #2044

### What

Adds api to send a recording loaded from RRD to a new recording stream,
cloning the data in the process

Author: oxkitsune

.gitattributes

@@ -5,3 +5,4 @@ pixi.lock linguist-generated=true
 # git-lfs stuff
 **/snapshots/**/*.png filter=lfs diff=lfs merge=lfs -text
 tests/assets/video/*.mp4 filter=lfs diff=lfs merge=lfs -text
+examples/assets/example.rrd filter=lfs diff=lfs merge=lfs -text

crates/top/re_sdk/src/recording_stream.rs

@@ -1696,10 +1696,20 @@ impl RecordingStream {
         }
     }
 
-    /// Records a single [`Chunk`].
+    /// Logs multiple [`Chunk`]s.
     ///
     /// This will _not_ inject `log_tick` and `log_time` timeline columns into the chunk,
-    /// for that use [`Self::log_chunk`].
+    /// for that use [`Self::log_chunks`].
+    pub fn log_chunks(&self, chunks: impl IntoIterator<Item = Chunk>) {
+        for chunk in chunks {
+            self.log_chunk(chunk);
+        }
+    }
+
+    /// Records a single [`Chunk`].
+    ///
+    /// Will inject `log_tick` and `log_time` timeline columns into the chunk.
+    /// If you don't want to inject these, use [`Self::send_chunks`] instead.
     #[inline]
     pub fn send_chunk(&self, chunk: Chunk) {
         let f = move |inner: &RecordingStreamInner| {
@@ -1711,6 +1721,16 @@ impl RecordingStream {
         }
     }
 
+    /// Records multiple [`Chunk`]s.
+    ///
+    /// This will _not_ inject `log_tick` and `log_time` timeline columns into the chunk,
+    /// for that use [`Self::log_chunks`].
+    pub fn send_chunks(&self, chunks: impl IntoIterator<Item = Chunk>) {
+        for chunk in chunks {
+            self.send_chunk(chunk);
+        }
+    }
+
     /// Swaps the underlying sink for a new one.
     ///
     /// This guarantees that:

docs/snippets/all/concepts/send_recording.py

@@ -0,0 +1,16 @@
+"""Send a dataframe to a new recording stream."""
+
+import sys
+
+import rerun as rr
+
+path_to_rrd = sys.argv[1]
+
+# NOTE: This is specifically demonstrating how to send `rr.dataframe.Recording` into the viewer.
+# If you just want to view an RRD file, use the simpler `rr.log_file()` function instead:
+#   rr.log_file("path/to/file.rrd", spawn=True)
+
+recording = rr.dataframe.load_recording(path_to_rrd)
+
+rr.init(recording.application_id(), recording_id=recording.recording_id(), spawn=True)
+rr.send_recording(recording)

docs/snippets/all/concepts/send_recording.rs

@@ -0,0 +1,35 @@
+//! Send a `.rrd` to a new recording stream.
+
+use rerun::external::re_chunk_store::{ChunkStore, ChunkStoreConfig};
+use rerun::external::re_log_types::{LogMsg, SetStoreInfo};
+use rerun::external::re_tuid::Tuid;
+use rerun::VersionPolicy;
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Get the filename from the command-line args.
+    let filename = std::env::args().nth(2).ok_or("Missing filename argument")?;
+
+    // Load the chunk store from the file.
+    let (store_id, store) =
+        ChunkStore::from_rrd_filepath(&ChunkStoreConfig::DEFAULT, filename, VersionPolicy::Warn)?
+            .into_iter()
+            .next()
+            .ok_or("Expected exactly one recording in the archive")?;
+
+    // Use the same app and recording IDs as the original.
+    if let Some(info) = store.info().cloned() {
+        let new_recording = rerun::RecordingStreamBuilder::new(info.application_id.clone())
+            .recording_id(store_id.to_string())
+            .spawn()?;
+
+        new_recording.record_msg(LogMsg::SetStoreInfo(SetStoreInfo {
+            row_id: Tuid::new(),
+            info,
+        }));
+
+        // Forward all chunks to the new recording stream.
+        new_recording.send_chunks(store.iter_chunks().map(|chunk| (**chunk).clone()));
+    }
+
+    Ok(())
+}

docs/snippets/snippets.toml

@@ -99,6 +99,9 @@ features = [
   "cpp",
   "rust",
 ]
+"concepts/send_recording" = [ # Not implemented for C++
+  "cpp",
+]
 "concepts/static/log_static" = [ # pseudo-code
   "py",
   "cpp",
@@ -361,6 +364,7 @@ quick_start = [ # These examples don't have exactly the same implementation.
 
 # `$config_dir` will be replaced with the absolute path of `docs/snippets`.
 [extra_args]
+"concepts/send_recording" = ["$config_dir/../../tests/assets/rrd/dna.rrd"]
 "archetypes/asset3d_simple" = ["$config_dir/../../tests/assets/cube.glb"]
 "archetypes/asset3d_out_of_tree" = ["$config_dir/../../tests/assets/cube.glb"]
 "archetypes/video_auto_frames" = [

examples/assets/example.rrd

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:4467aa40910101da2f308ba2a526a5984ab9512b628257a8dae104818e3f3c03
-size 1739
+oid sha256:1868150a997a96d4a98dec7da92f5ba313918416c3b3d1b57a205b0890e72c88
+size 2250

rerun_py/rerun_bindings/rerun_bindings.pyi

@@ -881,6 +881,14 @@ def send_blueprint(
 ) -> None:
     """Send a blueprint to the given recording stream."""
 
+def send_recording(rrd: Recording, recording: Optional[PyRecordingStream] = None) -> None:
+    """
+    Send all chunks from a [`PyRecording`] to the given recording stream.
+
+    .. warning::
+        ⚠️ This API is experimental and may change or be removed in future versions! ⚠️
+    """
+
 #
 # misc
 #

rerun_py/rerun_sdk/rerun/__init__.py

@@ -181,6 +181,7 @@
     disconnect as disconnect,
     save as save,
     send_blueprint as send_blueprint,
+    send_recording as send_recording,
     serve_grpc as serve_grpc,
     serve_web as serve_web,
     spawn as spawn,

rerun_py/rerun_sdk/rerun/recording_stream.py

@@ -13,6 +13,7 @@
 import numpy as np
 from typing_extensions import deprecated
 
+import rerun as rr
 from rerun import bindings
 from rerun.memory import MemoryRecording
 from rerun.time import reset_time
@@ -721,6 +722,24 @@ def send_blueprint(
 
         send_blueprint(blueprint=blueprint, make_active=make_active, make_default=make_default, recording=self)
 
+    def send_recording(self, recording: rr.dataframe.Recording) -> None:
+        """
+        Send a `Recording` loaded from a `.rrd` to the `RecordingStream`.
+
+        .. warning::
+            ⚠️ This API is experimental and may change or be removed in future versions! ⚠️
+
+        Parameters
+        ----------
+        recording:
+            A `Recording` loaded from a `.rrd`.
+
+        """
+
+        from .sinks import send_recording
+
+        send_recording(rrd=recording, recording=self)
+
     def spawn(
         self,
         *,

rerun_py/rerun_sdk/rerun/sinks.py

@@ -7,6 +7,8 @@
 import rerun_bindings as bindings
 
 from rerun.blueprint.api import BlueprintLike, create_in_memory_blueprint
+from rerun.dataframe import Recording
+from rerun.recording_stream import RecordingStream, get_application_id
 
 from ._spawn import _spawn_viewer
 
@@ -398,6 +400,34 @@ def send_blueprint(
     )
 
 
+def send_recording(rrd: Recording, recording: RecordingStream | None = None) -> None:
+    """
+    Send a `Recording` loaded from a `.rrd` to the `RecordingStream`.
+
+    .. warning::
+        ⚠️ This API is experimental and may change or be removed in future versions! ⚠️
+
+    Parameters
+    ----------
+    rrd:
+        A recording loaded from a `.rrd` file.
+    recording:
+        Specifies the [`rerun.RecordingStream`][] to use.
+        If left unspecified, defaults to the current active data recording, if there is one.
+        See also: [`rerun.init`][], [`rerun.set_global_data_recording`][].
+
+    """
+    application_id = get_application_id(recording=recording)  # NOLINT
+
+    if application_id is None:
+        raise ValueError("No application id found. You must call rerun.init before sending a recording.")
+
+    bindings.send_recording(
+        rrd,
+        recording=recording.to_native() if recording is not None else None,
+    )
+
+
 def spawn(
     *,
     port: int = 9876,

rerun_py/src/python_bridge.rs

@@ -42,6 +42,8 @@ impl PyRuntimeErrorExt for PyRuntimeError {
 
 use once_cell::sync::{Lazy, OnceCell};
 
+use crate::dataframe::PyRecording;
+
 // The bridge needs to have complete control over the lifetimes of the individual recordings,
 // otherwise all the recording shutdown machinery (which includes deallocating C, Rust and Python
 // data and joining a bunch of threads) can end up running at any time depending on what the
@@ -172,6 +174,7 @@ fn rerun_bindings(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_function(wrap_pyfunction!(log_file_from_contents, m)?)?;
     m.add_function(wrap_pyfunction!(send_arrow_chunk, m)?)?;
     m.add_function(wrap_pyfunction!(send_blueprint, m)?)?;
+    m.add_function(wrap_pyfunction!(send_recording, m)?)?;
 
     // misc
     m.add_function(wrap_pyfunction!(version, m)?)?;
@@ -1341,6 +1344,23 @@ fn send_blueprint(
     }
 }
 
+/// Send all chunks from a [`PyRecording`] to the given recording stream.
+///
+/// .. warning::
+///     ⚠️ This API is experimental and may change or be removed in future versions! ⚠️
+#[pyfunction]
+#[pyo3(signature = (rrd, recording = None))]
+fn send_recording(rrd: &PyRecording, recording: Option<&PyRecordingStream>) {
+    let Some(recording) = get_data_recording(recording) else {
+        return;
+    };
+
+    let store = rrd.store.read();
+    for chunk in store.iter_chunks() {
+        recording.send_chunk((**chunk).clone());
+    }
+}
+
 // --- Misc ---
 
 /// Return a verbose version string.
@@ -1471,7 +1491,7 @@ fn send_recording_start_time_nanos(
 
 // --- Helpers ---
 
-fn python_version(py: Python<'_>) -> re_log_types::PythonVersion {
+pub fn python_version(py: Python<'_>) -> re_log_types::PythonVersion {
     let py_version = py.version_info();
     re_log_types::PythonVersion {
         major: py_version.major,