Add file-backed ring buffer with per-topic configurability

src/rosbag_blackbox/config/example_config.yaml

@@ -12,6 +12,11 @@
 #   pattern: <glob>   – match by topic NAME using the syntax above
 #   type:    <string> – match by message TYPE (exact, e.g. geometry_msgs/msg/Twist)
 # Both keys may be combined; in that case BOTH conditions must hold.
+#
+# Buffer type (optional, default: memory):
+#   buffer: memory  – keep messages in RAM (default, lowest latency)
+#   buffer: file    – keep messages in a temporary file (reduces RAM usage;
+#                     recommended for high-bandwidth topics like images)
 
 # ── Ring-buffer settings ────────────────────────────────────────────────────
 buffer:
@@ -50,6 +55,7 @@ topics:
 
   # ── All Camera topics at max 1 Hz ─────────────────────────────────────────
   - type: "sensor_msgs/msg/Image"
+    buffer: file              # use a file-backed buffer to reduce RAM usage
     max_frequency: 1.0        # Hz  (omit or set to 0 for unlimited)
     qos:
       reliability: best_effort

src/rosbag_blackbox/rosbag_blackbox/ring_buffer.py

@@ -12,12 +12,23 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Thread-safe, time-bounded ring buffer for serialised ROS2 messages.
+"""Thread-safe, time-bounded ring buffers for serialised ROS2 messages.
+
+Two implementations are provided:
+
+* :class:`RingBuffer` – in-memory (default), uses a :class:`collections.deque`.
+* :class:`FileRingBuffer` – file-backed; only a small index is kept in RAM
+  while message payloads are stored in a temporary file.  Suitable for
+  high-bandwidth topics such as camera images where RAM usage is a concern.
+
+Both classes expose the same public API so they are interchangeable.
 
 Kept in a separate module so it can be unit-tested without a ROS2
 installation.
 """
 
+import struct
+import tempfile
 import threading
 from collections import deque
 from typing import List, Tuple
@@ -71,3 +82,140 @@ def memory_bytes(self) -> int:
         """Estimated total byte size of all buffered message payloads."""
         with self._lock:
             return sum(len(d) for _, _, _, d in self._buf)
+
+
+class FileRingBuffer:
+    """File-backed ring buffer that retains the last *duration_seconds* of data.
+
+    Message payloads are written to a temporary file so that RAM usage stays
+    low regardless of payload size.  Only a compact in-memory index
+    ``(timestamp_ns, file_offset, record_size)`` is kept per message.
+
+    The temporary file is automatically removed when the buffer is garbage-
+    collected or when :meth:`close` is called explicitly.
+
+    The public API is identical to :class:`RingBuffer`, making the two
+    classes interchangeable.
+    """
+
+    # Binary record header: timestamp_ns (int64 BE), topic_len (uint32 BE),
+    # type_name_len (uint32 BE), data_len (uint32 BE)
+    _HDR: struct.Struct = struct.Struct('>qIII')
+
+    def __init__(self, duration_seconds: float) -> None:
+        self._duration_ns: int = int(duration_seconds * 1_000_000_000)
+        self._lock: threading.Lock = threading.Lock()
+        # TemporaryFile is deleted automatically on close/GC
+        self._tmpfile = tempfile.TemporaryFile()
+        self._write_pos: int = 0
+        # Each entry: (timestamp_ns, file_offset, record_size_bytes)
+        self._index: deque = deque()
+        self._live_bytes: int = 0  # bytes occupied by non-purged records
+
+    # ------------------------------------------------------------------
+    # Serialisation helpers
+    # ------------------------------------------------------------------
+
+    def _pack(
+        self,
+        timestamp_ns: int,
+        topic: str,
+        type_name: str,
+        data: bytes,
+    ) -> bytes:
+        tb = topic.encode()
+        tnb = type_name.encode()
+        hdr = self._HDR.pack(timestamp_ns, len(tb), len(tnb), len(data))
+        return hdr + tb + tnb + data
+
+    def _unpack_at(self, offset: int) -> Tuple[int, str, str, bytes]:
+        self._tmpfile.seek(offset)
+        hdr_bytes = self._tmpfile.read(self._HDR.size)
+        ts_ns, tl, tnl, dl = self._HDR.unpack(hdr_bytes)
+        topic = self._tmpfile.read(tl).decode()
+        type_name = self._tmpfile.read(tnl).decode()
+        data = self._tmpfile.read(dl)
+        return (ts_ns, topic, type_name, data)
+
+    # ------------------------------------------------------------------
+    # Core operations
+    # ------------------------------------------------------------------
+
+    def add(
+        self,
+        timestamp_ns: int,
+        topic: str,
+        type_name: str,
+        data: bytes,
+    ) -> None:
+        """Append a message and evict entries outside the retention window."""
+        record = self._pack(timestamp_ns, topic, type_name, data)
+        with self._lock:
+            self._tmpfile.seek(self._write_pos)
+            self._tmpfile.write(record)
+            self._index.append((timestamp_ns, self._write_pos, len(record)))
+            self._write_pos += len(record)
+            self._live_bytes += len(record)
+            self._purge(timestamp_ns)
+
+    def _purge(self, now_ns: int) -> None:
+        """Evict records older than the retention window (called under lock)."""
+        if self._duration_ns <= 0:
+            return
+        cutoff = now_ns - self._duration_ns
+        while self._index and self._index[0][0] < cutoff:
+            _, _, size = self._index.popleft()
+            self._live_bytes -= size
+        # Reclaim disk space when dead bytes exceed live bytes
+        dead = self._write_pos - self._live_bytes
+        if self._index and dead > self._live_bytes:
+            self._compact()
+
+    def _compact(self) -> None:
+        """Rewrite the temp file keeping only live records (called under lock)."""
+        # Read all live raw records before modifying the file
+        live_chunks: List[bytes] = []
+        for _ts, offset, size in self._index:
+            self._tmpfile.seek(offset)
+            live_chunks.append(self._tmpfile.read(size))
+
+        self._tmpfile.seek(0)
+        self._tmpfile.truncate(0)
+        new_pos: int = 0
+        new_index: deque = deque()
+        for (ts_ns, _old_off, size), chunk in zip(self._index, live_chunks):
+            self._tmpfile.write(chunk)
+            new_index.append((ts_ns, new_pos, size))
+            new_pos += size
+        self._tmpfile.flush()
+        self._write_pos = new_pos
+        self._live_bytes = new_pos
+        self._index = new_index
+
+    # ------------------------------------------------------------------
+    def snapshot(self) -> List[Tuple[int, str, str, bytes]]:
+        """Return a point-in-time copy of all buffered messages."""
+        with self._lock:
+            return [self._unpack_at(offset) for _, offset, _ in self._index]
+
+    def size(self) -> int:
+        """Number of messages currently in the buffer."""
+        with self._lock:
+            return len(self._index)
+
+    def memory_bytes(self) -> int:
+        """Bytes occupied by live records in the backing file."""
+        with self._lock:
+            return self._live_bytes
+
+    # ------------------------------------------------------------------
+    def close(self) -> None:
+        """Close and delete the backing temporary file."""
+        with self._lock:
+            self._tmpfile.close()
+
+    def __del__(self) -> None:
+        try:
+            self._tmpfile.close()
+        except Exception:  # noqa: BLE001
+            pass

src/rosbag_blackbox/rosbag_blackbox/ringbuffer_node.py

@@ -64,7 +64,7 @@
 
 import rosbag2_py
 
-from rosbag_blackbox.ring_buffer import RingBuffer
+from rosbag_blackbox.ring_buffer import FileRingBuffer, RingBuffer
 from rosbag_blackbox.snapshot_helpers import format_snapshot_name, param_value_to_python
 from rosbag_blackbox.topic_filter import find_matching_config, matches_pattern
 
@@ -201,6 +201,9 @@ def __init__(self) -> None:
         self._topic_types: Dict[str, str] = {}  # topic -> type_name
         self._last_msg_ns: Dict[str, int] = {}  # topic -> last timestamp (ns)
         self._topic_cfgs: list = self._cfg.get('topics', [])
+        # Per-topic file-backed buffers (keyed by topic name).
+        # Created for topics configured with ``buffer: file``.
+        self._file_bufs: Dict[str, FileRingBuffer] = {}
         # Latest message for transient_local topics – never evicted so that
         # latched topics always appear in every snapshot.
         self._latched_last: Dict[str, tuple] = {}  # topic -> (ts_ns, topic, type, data)
@@ -366,11 +369,28 @@ def _subscribe(
             qos_cfg.get('durability', 'volatile').lower() == 'transient_local'
         )
 
+        # Determine which buffer to use for this topic.
+        # Topics configured with ``buffer: file`` get their own FileRingBuffer;
+        # all other topics share the main in-memory RingBuffer (default).
+        use_file: bool = (
+            topic_cfg.get('buffer', 'memory').lower() == 'file'
+        )
+        if use_file:
+            duration = float(self._cfg['buffer']['duration_seconds'])
+            topic_buf = FileRingBuffer(duration)
+            self._file_bufs[topic] = topic_buf
+            self.get_logger().info(
+                f'Using file-backed buffer for {topic} [{type_name}]'
+            )
+        else:
+            topic_buf = self._buf
+
         def _make_cb(
             t: str,
             tn: str,
             min_interval: int,
             latched: bool,
+            buf,
         ):
             def _cb(msg: Any) -> None:
                 now_ns: int = self.get_clock().now().nanoseconds
@@ -384,7 +404,7 @@ def _cb(msg: Any) -> None:
 
                 try:
                     data = serialize_message(msg)
-                    self._buf.add(now_ns, t, tn, data)
+                    buf.add(now_ns, t, tn, data)
                     if latched:
                         self._latched_last[t] = (now_ns, t, tn, data)
                 except Exception as exc2:  # noqa: BLE001
@@ -398,7 +418,7 @@ def _cb(msg: Any) -> None:
             sub = self.create_subscription(
                 msg_class,
                 topic,
-                _make_cb(topic, type_name, min_interval_ns, is_latched),
+                _make_cb(topic, type_name, min_interval_ns, is_latched, topic_buf),
                 qos,
             )
             self._subs[topic] = sub
@@ -554,6 +574,13 @@ def _write_snapshot(self, trigger: str = 'manual') -> str:
         """
         messages = self._buf.snapshot()
 
+        # Collect messages from per-topic file-backed buffers and merge
+        for file_buf in self._file_bufs.values():
+            messages.extend(file_buf.snapshot())
+
+        # Sort merged messages by timestamp so the bag is time-ordered
+        messages.sort(key=lambda entry: entry[0])
+
         # Always include the latest value for transient_local (latched) topics
         # even if they were evicted from the ring buffer.
         buffered_topics = {topic for _, topic, _, _ in messages}