Refactor mcap recording

Author: nvddr

examples/oxr/python/modular_example_with_mcap.py

@@ -8,14 +8,13 @@
 - Create independent trackers
 - Add only the trackers you need
 - Record all tracker data to an MCAP file for playback/analysis
-- McapRecorder.create() is similar to DeviceIOSession.run()
+- Pass mcap_filename and mcap_channels to DeviceIOSession.run() to enable recording
 """
 
 import sys
 import time
 from datetime import datetime
 import isaacteleop.deviceio as deviceio
-import isaacteleop.mcap as mcap
 import isaacteleop.oxr as oxr
 
 
@@ -50,55 +49,42 @@ def main():
         handles = oxr_session.get_handles()
         print("✓ OpenXR session created")
 
-        # Run deviceio session with trackers
-        print("\nRunning deviceio session...")
-        with deviceio.DeviceIOSession.run(trackers, handles) as session:
+        # Run deviceio session with MCAP recording enabled.
+        print("\nRunning deviceio session with MCAP recording...")
+        with deviceio.DeviceIOSession.run(
+            trackers,
+            handles,
+            mcap_filename=mcap_filename,
+            mcap_channels=[(hand_tracker, "hands"), (head_tracker, "head")],
+        ) as session:
             print("✓ DeviceIO session initialized with all trackers!")
+            print(f"✓ MCAP recording active → {mcap_filename}")
+            print()
 
-            with mcap.McapRecorder.create(
-                mcap_filename,
-                [
-                    (hand_tracker, "hands"),
-                    (head_tracker, "head"),
-                ],
-            ) as recorder:
-                print("✓ MCAP recording started!")
-                print()
-
-                # Main tracking loop
-                print("=" * 60)
-                print("Tracking (60 seconds)...")
-                print("=" * 60)
-                print()
-
-                frame_count = 0
-                start_time = time.time()
-
-                while time.time() - start_time < 30.0:
-                    # Update session and all trackers
-                    if not session.update():
-                        print("Update failed")
-                        break
-
-                    # Record all registered trackers
-                    recorder.record(session)
-
-                    # Print every 60 frames (~1 second)
-                    if frame_count % 60 == 0:
-                        elapsed = time.time() - start_time
-                        print(f"[{elapsed:4.1f}s] Frame {frame_count} (recording...)")
-                        print()
-
-                    frame_count += 1
-                    time.sleep(0.016)  # ~60 FPS
-
-                # Cleanup
-                print(f"\nProcessed {frame_count} frames")
-                print("Cleaning up (RAII)...")
-
-            print("✓ Recording stopped")
-
-        print("✓ DeviceIO session cleaned up")
+            # Main tracking loop
+            print("=" * 60)
+            print("Tracking (30 seconds)...")
+            print("=" * 60)
+            print()
+
+            frame_count = 0
+            start_time = time.time()
+
+            while time.time() - start_time < 30.0:
+                session.update()
+
+                # Print every 60 frames (~1 second)
+                if frame_count % 60 == 0:
+                    elapsed = time.time() - start_time
+                    print(f"[{elapsed:4.1f}s] Frame {frame_count} (recording...)")
+                    print()
+
+                frame_count += 1
+                time.sleep(0.016)  # ~60 FPS
+
+            print(f"\nProcessed {frame_count} frames")
+
+        print("✓ Recording stopped (MCAP file closed by session destructor)")
 
     print()
     print("=" * 60)

examples/oxr/python/test_oak_camera.py

@@ -24,7 +24,6 @@
 
 import isaacteleop.plugin_manager as pm
 import isaacteleop.deviceio as deviceio
-import isaacteleop.mcap as mcap
 import isaacteleop.oxr as oxr
 
 PLUGIN_ROOT_DIR = Path(__file__).resolve().parent.parent.parent.parent / "plugins"
@@ -66,62 +65,53 @@ def _run_schema_pusher(
         handles = oxr_session.get_handles()
         print("  ✓ OpenXR session created")
 
-        # Create DeviceIOSession with all trackers
-        with deviceio.DeviceIOSession.run([tracker], handles) as session:
-            print("  ✓ DeviceIO session initialized")
-
-            # Create MCAP recorder with per-stream FrameMetadataOak channels
-            mcap_entries = [(tracker, "oak_metadata")]
-            with mcap.McapRecorder.create(
-                mcap_filename,
-                mcap_entries,
-            ) as recorder:
-                print("  ✓ MCAP recording started")
-                print()
-
-                # 6. Main tracking loop
-                print(f"[Step 6] Recording video and metadata ({duration} seconds)...")
-                print("-" * 80)
-                start_time = time.time()
-                frame_count = 0
-                last_print_time = 0
-                last_seq = dict.fromkeys(stream_names, -1)
-                metadata_samples = dict.fromkeys(stream_names, 0)
-
-                while time.time() - start_time < duration:
-                    plugin.check_health()
-                    if not session.update():
-                        print("  Warning: Session update failed")
-                        continue
-                    recorder.record(session)
-                    frame_count += 1
-
-                    elapsed = time.time() - start_time
-                    for idx, name in enumerate(stream_names):
-                        tracked = tracker.get_stream_data(session, idx)
-                        if (
-                            tracked.data is not None
-                            and tracked.data.sequence_number != last_seq.get(name, -1)
-                        ):
-                            metadata_samples[name] = metadata_samples.get(name, 0) + 1
-                            last_seq[name] = tracked.data.sequence_number
-
-                    if int(elapsed) > last_print_time:
-                        last_print_time = int(elapsed)
-                        parts = []
-                        for name in stream_names:
-                            parts.append(f"{name}={metadata_samples.get(name, 0)}")
-                        print(f"  [{last_print_time:3d}s] samples: {', '.join(parts)}")
-                    time.sleep(0.016)
-
-                print("-" * 80)
-                print()
-                print(f"  ✓ Recording completed ({duration:.1f} seconds)")
-                print(f"  ✓ Processed {frame_count} update cycles")
-                for name in stream_names:
-                    print(
-                        f"  ✓ {name}: {metadata_samples.get(name, 0)} metadata samples"
-                    )
+        # Create DeviceIOSession with MCAP recording enabled.
+        with deviceio.DeviceIOSession.run(
+            [tracker],
+            handles,
+            mcap_filename=mcap_filename,
+            mcap_channels=[(tracker, "oak_metadata")],
+        ) as session:
+            print("  ✓ DeviceIO session initialized (recording active during update())")
+            print()
+
+            print(f"[Step 6] Recording video and metadata ({duration} seconds)...")
+            print("-" * 80)
+            start_time = time.time()
+            frame_count = 0
+            last_print_time = 0
+            last_seq = dict.fromkeys(stream_names, -1)
+            metadata_samples = dict.fromkeys(stream_names, 0)
+
+            while time.time() - start_time < duration:
+                plugin.check_health()
+                session.update()
+                frame_count += 1
+
+                elapsed = time.time() - start_time
+                for idx, name in enumerate(stream_names):
+                    tracked = tracker.get_stream_data(session, idx)
+                    if (
+                        tracked.data is not None
+                        and tracked.data.sequence_number != last_seq.get(name, -1)
+                    ):
+                        metadata_samples[name] = metadata_samples.get(name, 0) + 1
+                        last_seq[name] = tracked.data.sequence_number
+
+                if int(elapsed) > last_print_time:
+                    last_print_time = int(elapsed)
+                    parts = []
+                    for name in stream_names:
+                        parts.append(f"{name}={metadata_samples.get(name, 0)}")
+                    print(f"  [{last_print_time:3d}s] samples: {', '.join(parts)}")
+                time.sleep(0.016)
+
+            print("-" * 80)
+            print()
+            print(f"  ✓ Recording completed ({duration:.1f} seconds)")
+            print(f"  ✓ Processed {frame_count} update cycles")
+            for name in stream_names:
+                print(f"  ✓ {name}: {metadata_samples.get(name, 0)} metadata samples")
 
 
 def run_test(duration: float = 10.0, mode: str = MODE_NO_METADATA):

src/core/deviceio_base/cpp/inc/deviceio_base/tracker.hpp

@@ -4,9 +4,7 @@
 #pragma once
 
 #include <openxr/openxr.h>
-#include <schema/timestamp_generated.h>
 
-#include <functional>
 #include <memory>
 #include <string>
 #include <string_view>
@@ -19,62 +17,14 @@ namespace core
 struct OpenXRSessionHandles;
 class ITrackerFactory;
 
-// Base interface for tracker implementations
-// These are the actual worker objects that get updated by the session
+// Base interface for tracker implementations.
+// The actual worker objects updated each frame by DeviceIOSession.
 class ITrackerImpl
 {
 public:
     virtual ~ITrackerImpl() = default;
 
-    // Update the tracker with the current time
     virtual bool update(XrTime time) = 0;
-
-    /**
-     * @brief Callback type for serialize_all.
-     *
-     * Receives (log_time_ns, data_ptr, data_size) for each serialized record.
-     *
-     * @param log_time_ns  Monotonic nanoseconds used as the MCAP logTime/publishTime
-     *                     for this record. This is the time at which the recording
-     *                     system processed the record (update-tick time), not the
-     *                     sample capture time. The full per-sample DeviceDataTimestamp
-     *                     (including sample_time and raw_device_time) is embedded
-     *                     inside the serialized FlatBuffer payload.
-     *
-     * @warning The data_ptr and data_size are only valid for the duration of the
-     *          callback invocation. The buffer is owned by a FlatBufferBuilder
-     *          local to the tracker's serialize_all implementation and will be
-     *          destroyed on return. If you need the bytes after the callback
-     *          returns, copy them into your own storage before returning.
-     */
-    using RecordCallback = std::function<void(int64_t log_time_ns, const uint8_t*, size_t)>;
-
-    /**
-     * @brief Serialize all records accumulated since the last update() call.
-     *
-     * Each call to update() clears the previous batch and accumulates a fresh
-     * set of records (one for OpenXR-direct trackers; potentially many for
-     * SchemaTracker-based tensor-device trackers). serialize_all emits every
-     * record in that batch via the callback.
-     *
-     * @note For multi-channel trackers the recorder calls serialize_all once per
-     *       channel index (channel_index = 0, 1, … N-1) after each update().
-     *       All serialize_all calls for a given update() are guaranteed to
-     *       complete before the next update() is issued. Implementations may
-     *       therefore maintain a single shared pending batch and clear it at the
-     *       start of the next update(); there is no need to track per-channel
-     *       drain state.
-     *
-     * For read access without MCAP recording, use the tracker's typed get_*()
-     * accessors, which always reflect the last record in the current batch.
-     *
-     * @note The buffer pointer passed to the callback is only valid for the
-     *       duration of that callback call. Copy if you need it beyond return.
-     *
-     * @param channel_index Which record channel to serialize (0-based).
-     * @param callback Invoked once per record with (timestamp, data_ptr, data_size).
-     */
-    virtual void serialize_all(size_t channel_index, const RecordCallback& callback) const = 0;
 };
 
 /**
@@ -100,8 +50,8 @@ class ITrackerSession
     virtual const ITrackerImpl& get_tracker_impl(const class ITracker& tracker) const = 0;
 };
 
-// Base interface for all trackers
-// PUBLIC API: Only exposes methods that external users should call
+// Base interface for all trackers.
+// Public API: configuration, extension requirements, and typed data accessors.
 class ITracker
 {
 public:
@@ -110,48 +60,14 @@ class ITracker
     virtual std::vector<std::string> get_required_extensions() const = 0;
     virtual std::string_view get_name() const = 0;
 
-    /**
-     * @brief Get the FlatBuffer schema name (root type) for MCAP recording.
-     *
-     * This should return the fully qualified FlatBuffer type name (e.g., "core.HandPose")
-     * which matches the root_type defined in the .fbs schema file.
-     */
-    virtual std::string_view get_schema_name() const = 0;
-
-    /**
-     * @brief Get the binary FlatBuffer schema text for MCAP recording.
-     */
-    virtual std::string_view get_schema_text() const = 0;
-
-    /**
-     * @brief Get the channel names for MCAP recording.
-     *
-     * Every tracker must return at least one non-empty channel name. The returned
-     * vector size determines how many times serialize_all() is called per update,
-     * with the vector index used as the channel_index argument.
-     *
-     * Single-channel trackers return one name (e.g. {"head"}).
-     * Multi-channel trackers return multiple (e.g. {"left_hand", "right_hand"}).
-     *
-     * The MCAP recorder combines each channel name with the base channel name
-     * provided at registration as "base_name/channel_name". For example, a
-     * single-channel head tracker registered with base name "tracking" produces
-     * the MCAP channel "tracking/head". A multi-channel hand tracker registered
-     * with base name "hands" produces "hands/left_hand" and "hands/right_hand".
-     *
-     * @return Non-empty vector of non-empty channel name strings.
-     */
-    virtual std::vector<std::string> get_record_channels() const = 0;
-
     /**
      * @brief Create the tracker's implementation via the provided factory.
      *
      * Uses double dispatch: the tracker calls the factory method specific to its
-     * type (e.g., factory.create_head_tracker_impl()), so the factory controls
-     * which concrete impl is constructed without the tracker needing to know the
-     * session type.
+     * concrete type, passing `this` so the factory can query a recorder for a
+     * channel writer and attach it to the impl at construction time.
      *
-     * @param factory Session-provided factory (e.g., LiveDeviceIOFactory).
+     * @param factory Session-provided factory (e.g. LiveDeviceIOFactory).
      * @return Owning pointer to the newly created impl.
      */
     virtual std::unique_ptr<ITrackerImpl> create_tracker_impl(ITrackerFactory& factory) const = 0;

src/core/deviceio_session/cpp/CMakeLists.txt

@@ -24,6 +24,7 @@ target_link_libraries(deviceio_session
     PRIVATE
         deviceio::deviceio_trackers
         deviceio::live_trackers
+        mcap::mcap
 )
 
 add_library(deviceio::deviceio_session ALIAS deviceio_session)