recording

McapRecorder ¶

McapRecorder(
    *,
    output_dir: Path | str = "~/.rosys/mcap",
    max_file_size_mb: float = 100,
    max_total_size_mb: float = 1000,
    chunk_size: int = 1048576,
    flush_interval: float = 1.0,
    profile: str = "rosys",
    library: str = "rosys-mcap-recorder",
    logger_name: str = "rosys.mcap_recorder",
    auto_start: bool = True,
    max_queued_bytes: float = MAX_QUEUED_BYTES
)

Records sensor data to MCAP files for replay and analysis in Foxglove Studio.

Supports automatic file rotation by size and disk budget enforcement. Peak disk usage is max_total_size_mb + max_file_size_mb: the budget is enforced only before a file is opened, so the currently growing file can exceed it by up to one file's worth.

Messages are enqueued from the event loop (cheap, non-blocking) and written to disk by a single background consumer via rosys.run.io_bound so that encoding, ZSTD compression and file I/O never block the loop. Encoding runs on the writer too: sources enqueue the raw payload plus an encode callable, so no JSON/JPEG work happens on the loop. The log time is captured at enqueue, but encoding is deferred, so payloads are expected to be immutable value snapshots (as RoSys sensor events emit); a payload mutated after being enqueued would encode its later state.

Two locks keep the loop responsive. _lock serializes all writer access so the drain in stop() cannot race the background consumer; it may be held for the duration of a multi-second write. _queue_lock guards only queue mutation (enqueue, cap-drop, swap) and is held for microseconds, so the event loop never blocks behind a write when appending a message. Lock order is _lock outer, _queue_lock inner; the loop takes _queue_lock alone, so there is no deadlock.

The recorder is encoding-agnostic: a topic carries a :class:TopicSchema (schema name/bytes plus schema- and message-encoding). log_message takes either already-serialized bytes or a payload plus an encode callback. Conversion from application data types lives entirely in converters.py / foxglove.py. Topics are fed by opaque :class:RecordingSource objects that are activated on start() and deactivated on stop().

Create an MCAP recorder.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path \| str`	directory recordings are written to (created if missing).	`'~/.rosys/mcap'`
`max_file_size_mb`	`float`	on-disk size at which the active file is rotated to a new one.	`100`
`max_total_size_mb`	`float`	disk budget for the directory; the oldest recordings are deleted before a new file is opened to stay under it. Peak disk usage is therefore `max_total_size_mb + max_file_size_mb` (the budget is enforced only before a file is opened, so the growing file can exceed it by up to one file's worth).	`1000`
`chunk_size`	`int`	MCAP chunk size in bytes (larger chunks compress better and flush less often).	`1048576`
`flush_interval`	`float`	seconds between background flushes of the queue to disk.	`1.0`
`profile`	`str`	MCAP profile written into each file's header.	`'rosys'`
`library`	`str`	MCAP library string written into each file's header.	`'rosys-mcap-recorder'`
`logger_name`	`str`	name of the logger this recorder logs to.	`'rosys.mcap_recorder'`
`auto_start`	`bool`	start recording automatically on `rosys` startup.	`True`
`max_queued_bytes`	`float`	approximate memory cap for unwritten messages; the oldest are dropped once the queue exceeds this (or :attr:`max_queued_messages`), so a stalled writer cannot exhaust memory with raw camera frames (see :data:`MAX_QUEUED_BYTES`).	`MAX_QUEUED_BYTES`

current_recording `property` ¶

current_recording: Path | None

The file currently being written (unindexed until stopped), else None.

disabled_topics `property` ¶

disabled_topics: set[str]

Declared topics that the current selection drops (empty when everything is recorded).

dropped_message_count `property` ¶

dropped_message_count: int

Messages dropped (queue overflow or an aborted recording) since the recording started.

message_count `property` ¶

message_count: int

Number of messages written to the current recording.

recordings `property` ¶

recordings: list[Path]

All recording files, newest first (by modification time, so renames keep the order).

topics `property` ¶

topics: list[str]

All declared topic names (including those whose schema is not registered yet).

_cleanup_orphaned_reindex_files ¶

_cleanup_orphaned_reindex_files() -> None

Remove reindex temp files left by a crash mid-rebuild. Run once at construction, never during operation.

A live reindex writes to the same *.mcap.reindex-* name; deleting it mid-run would destroy the recovery, so this must not run from the periodic disk-budget path. Orphans are otherwise invisible to the budget, scan and UI (they do not match the *.mcap glob).

_collect_disk_stats ¶

_collect_disk_stats() -> _DiskStats

Stat the output directory (blocking I/O; run off the loop).

_drain_and_close ¶

_drain_and_close() -> Path | None

Write everything still queued and finalize the file. Runs off the loop; takes _lock.

Used by stop(). The final drain may rotate, so the finalized path is read after writing. _close_file runs in a finally so even a raising write still finalizes (never leaks) the open file.

Returns:

Type	Description
`Path \| None`	the path of the finalized file, or `None` if no file was open.

_emit_on_loop ¶

_emit_on_loop(
    callback: Callable[..., Any], *args: Any
) -> None

Run a callback on the event loop from the writer thread.

Events and source lifecycle must run on the loop (subscribers touch UI; sources own loop-bound subscriptions), never on the background writer, so they are scheduled thread-safely. Outside a running loop (e.g. a synchronous unit test) the call is skipped.

_enforce_queue_cap ¶

_enforce_queue_cap() -> None

Drop the oldest queued messages when the disk cannot keep up. Caller holds _queue_lock.

Bounds memory by both message count (:attr:max_queued_messages) and approximate payload bytes (:attr:max_queued_bytes) — a queued camera frame is a full uncompressed image, so the count cap alone would not stop the queue from growing to many gigabytes. The oldest messages are dropped until the queue is under both limits; the drop is logged at most once per _DROP_WARNING_INTERVAL so a persistent stall does not flood the log.

_hard_stop ¶

_hard_stop(reason: str, dropped: int) -> None

Abandon a recording whose writer can no longer be reopened. Runs on the writer thread, holds _lock.

Without this, a failed rotation leaves _writer None while _is_recording stays True, so every later batch is silently discarded while the UI still shows a live recording. Instead this marks the recorder not-recording, finalizes whatever file is still open, counts the lost messages (so the loss surfaces in dropped_message_count and the log rather than silently), and stops the sources back on the event loop (they own loop-bound subscriptions and timers).

Parameters:

Name	Type	Description	Default
`reason`	`str`	human-readable cause, logged at error level.	required
`dropped`	`int`	number of messages lost with the dead writer.	required

_refresh_stats `async` ¶

_refresh_stats() -> None

Refresh the cached directory stats off the event loop, then rebuild the stats grid.

_reopen_file ¶

_reopen_file(*, dropped_on_failure: int) -> bool

Reopen a fresh file after the writer was lost (e.g. a failed rotation). Caller holds _lock.

Parameters:

Name	Type	Description	Default
`dropped_on_failure`	`int`	messages to count as lost if reopening fails and the recorder hard-stops.	required

Returns:

Type	Description
`bool`	`True` if a writable file is open, `False` after a hard stop.

_rotate ¶

_rotate() -> None

Close the current file and start a fresh one. Caller must hold _lock; runs on the writer thread.

Emits RECORDING_STOPPED for the finalized file and RECORDING_STARTED for the new one, loop-safely, so per-file consumers (upload/post-processing) see every file of a long session — not just the first and last. The events are therefore per-file.

_rotate_file ¶

_rotate_file(*, dropped_on_failure: int) -> bool

Rotate to a new file, hard-stopping if the new file cannot be opened. Caller holds _lock.

Parameters:

Name	Type	Description	Default
`dropped_on_failure`	`int`	messages to count as lost if rotation fails and the recorder hard-stops.	required

Returns:

Type	Description
`bool`	`True` if a new file is open, `False` after a hard stop.

_start_with_selection ¶

_start_with_selection() -> None

Start recording the topics selected in the developer panel (default: all).

_stop_sources ¶

_stop_sources() -> None

Deactivate every source (must run on the event loop; sources own loop-bound state).

_write_batch ¶

_write_batch() -> None

Drain the queue and write it. Runs on the background writer thread.

_write_messages ¶

_write_messages(batch: list[_QueuedMessage]) -> None

Encode and write a batch of messages. Caller must hold _lock; runs on the writer thread.

Resilient to two failures that would otherwise lose data silently:

a converter that raises for one message is logged (once per topic) and skipped, so the rest of the batch still lands (see :meth:warn_converter_failure);
a lost writer — None while still recording, e.g. after a failed rotation — is reopened once; if that fails the recorder hard-stops with an error and a drop count rather than silently swallowing every future message (see :meth:_hard_stop).

accepts ¶

accepts(topic: str) -> bool

Whether a message on topic would currently be recorded.

Cheap, loop-safe pre-check (recording is active and the topic is not deselected) so callers can skip expensive encoding for topics that would be dropped anyway.

Parameters:

Name	Type	Description	Default
`topic`	`str`	the topic name to test.	required

Returns:

Type	Description
`bool`	`True` if a message on this topic would be enqueued.

add_source ¶

add_source(source: RecordingSource) -> None

Register a data source whose lifetime is bound to the recording state.

Activated immediately if recording is already running, otherwise on the next start().

add_topic ¶

add_topic(topic: str, schema: TopicSchema) -> None

Register a topic with its schema and encoding.

Can be called before or after start(). The channel is registered with the writer lazily on first message (or eagerly when a new file opens), so this method never touches the writer and is safe to call from the event loop while the background consumer is writing.

declare_topic ¶

declare_topic(topic: str) -> None

Announce a topic whose schema will only be registered on its first message.

Auto-dispatched converters cannot know their schema before a payload arrives, but declaring the name up front makes the topic visible in topics — so selections computed before the first message (e.g. "everything but camera images") still cover it.

delete_all_recordings ¶

delete_all_recordings() -> None

Delete all recordings except the one currently being written.

delete_recording ¶

delete_recording(path: Path | str) -> None

Delete a recording file (only within this recorder's output directory).

The currently-recording file is never deleted (the writer holds it open).

developer_ui ¶

developer_ui() -> None

Developer panel: auto-refreshing stats, start/stop buttons, and a topic selection.

Only the stats grid is rebuilt on the timer; the buttons are built once and toggle their visibility reactively, so they never flicker. The timer refreshes cached directory stats collected off the event loop (globbing and statting the directory would otherwise block the loop every second) and is bound to the client, so it is cleaned up on disconnect (no global event subscriptions). The topic checkboxes live in a collapsed expansion to keep the panel compact; the selection applies to the next start (a running recording is unaffected).

log_message ¶

log_message(
    topic: str,
    data: Any,
    *,
    encode: (
        Callable[[Any, int], bytes | None] | None
    ) = None,
    timestamp_ns: int | None = None
) -> None

Enqueue a message for the background writer.

Parameters:

Name	Type	Description	Default
`topic`	`str`	the (registered) topic to write to; unknown topics are dropped with a one-time warning.	required
`data`	`Any`	already-serialized `bytes` when `encode` is `None`, otherwise the raw payload passed to `encode` on the writer thread.	required
`encode`	`Callable[[Any, int], bytes \| None] \| None`	optional `(payload, timestamp_ns) -> bytes \| None` run off the event loop by the background writer; returning `None` drops the message.	`None`
`timestamp_ns`	`int \| None`	log time in nanoseconds (default: current `rosys.time()`).	`None`

reindex_unindexed `async` ¶

reindex_unindexed() -> None

Rebuild the index of every unindexed recording on a background thread.

rename_recording ¶

rename_recording(
    path: Path | str, new_name: str
) -> Path | None

Rename a recording within the output directory; returns the new path or None.

The currently-recording file cannot be renamed (the writer holds it open). new_name is reduced to a bare filename and given a .mcap suffix; empty, whitespace-only or dots-only names are rejected (they would escape the output directory) by returning None.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	the recording to rename.	required
`new_name`	`str`	the desired name (reduced to a bare `.mcap` filename).	required

Returns:

Type	Description
`Path \| None`	the new path, or `None` if the rename was rejected or the target exists.

scan_recordings ¶

scan_recordings() -> list[RecordingInfo]

Stat every recording and check its summary index; newest first.

Globs and stats every file and probes its summary index, which is blocking I/O; call via rosys.run.io_bound. The live file is flagged and never index-probed (the writer holds it open and it has no summary index until stopped). Files that vanish during the scan (e.g. deleted from the recordings page) are skipped.

Returns:

Type	Description
`list[RecordingInfo]`	one :class:`RecordingInfo` per file, ordered newest first.

start ¶

start(topics: Collection[str] | None = None) -> None

Start a new recording.

Parameters:

Name	Type	Description	Default
`topics`	`Collection[str] \| None`	record only these topics; all others are dropped (default: record every registered topic). A selected topic that is registered only after the recording started is picked up as soon as it exists. The selection lasts for this recording; the next `start()` records everything again unless a new selection is passed.	`None`

stop `async` ¶

stop() -> None

Stop recording, drain the queue, and finalize the file off the event loop.

The drain encodes and writes every still-queued message (JPEG/JSON/ZSTD) and finishes the MCAP file, which can take seconds for a large backlog; it runs on a worker thread via :func:asyncio.to_thread — deliberately not rosys.run.io_bound, which refuses work once the app is stopping (and stop is wired to rosys.on_shutdown) — so it never blocks the event loop. An empty recording (e.g. from rapid toggling) is discarded; otherwise RECORDING_STOPPED is emitted on the loop with the finalized path.

unindexed_recordings ¶

unindexed_recordings() -> list[Path]

Finished recordings without a summary index (e.g. left by a crash).

Opens and probes every finished file, which is blocking I/O; call via rosys.run.io_bound.

warn_converter_failure ¶

warn_converter_failure(topic: str, stage: str) -> None

Log a converter failure once per topic and stage, so one bad message never floods the log.

Both halves of a converter can raise, in different places: sample on the event loop, at the topic's full rate, and encode on the writer thread. Both report here, so a broken converter costs one log line rather than one per message. Call from an exception handler.

Parameters:

Name	Type	Description	Default
`topic`	`str`	the topic whose converter raised.	required
`stage`	`str`	which half raised, `'sampling'` or `'encoding'`; keyed separately so a failing `sample` does not mute a later `encode` failure on the same topic.	required

Events¶

Name	Description
RECORDING_STARTED	a recording file has been opened (argument: path); emitted per file, including on size rotation
RECORDING_STOPPED	a recording file has been finalized (argument: path); emitted per file, including on size rotation

RecordingInfo ¶

Bases: NamedTuple

A snapshot of one recording's on-disk facts, gathered off the event loop.

Events¶

Name	Description
RECORDING_STARTED	a recording file has been opened (argument: path); emitted per file, including on size rotation
RECORDING_STOPPED	a recording file has been finalized (argument: path); emitted per file, including on size rotation

RecordingSource ¶

Bases: Protocol

A topic data source whose lifetime is bound to the recorder's recording state.

Sources are activated when recording starts and deactivated when it stops, so event subscriptions and timers only run while a recording is actually open.

Events¶

Name	Description
RECORDING_STARTED	a recording file has been opened (argument: path); emitted per file, including on size rotation
RECORDING_STOPPED	a recording file has been finalized (argument: path); emitted per file, including on size rotation

RecordingsPage ¶

RecordingsPage(
    recorder: McapRecorder,
    *,
    header: Callable[[], None] | None = None
)

Lists the MCAP recordings for download and deletion.

The list refreshes whenever the recorder starts a new recording or stops one, can be filtered by date, and offers rebuilding the index of crash-orphaned (unindexed) recordings. All filesystem access (glob, stat, index check) runs off the event loop via rosys.run.io_bound; the render reads only a cached snapshot, so opening the page never blocks the loop on disk I/O.

A download endpoint at DOWNLOAD_PATH/{name} serves finished recordings over HTTP (basename only, refusing the live file with 409 and missing files with 404), so recordings can be fetched without scp.

Register the recordings page and its download endpoint on the nicegui app.

Parameters:

Name	Type	Description	Default
`recorder`	`McapRecorder`	the recorder whose recordings this page lists, manages, and serves.	required
`header`	`Callable[[], None] \| None`	optional callback rendered once at the top of the page before the content (e.g. a shared application header or navigation); `None` renders no header.	`None`

TopicSchema ¶

Bases: NamedTuple

Everything the writer needs to register a topic's channel and schema.

Events¶

Name	Description
RECORDING_STARTED	a recording file has been opened (argument: path); emitted per file, including on size rotation
RECORDING_STOPPED	a recording file has been finalized (argument: path); emitted per file, including on size rotation

recording

McapRecorder ¶

current_recording property ¶

disabled_topics property ¶

dropped_message_count property ¶

message_count property ¶

recordings property ¶

topics property ¶

_cleanup_orphaned_reindex_files ¶

_collect_disk_stats ¶

_drain_and_close ¶

_emit_on_loop ¶

_enforce_queue_cap ¶

_hard_stop ¶

_refresh_stats async ¶

_reopen_file ¶

_rotate ¶

_rotate_file ¶

_start_with_selection ¶

_stop_sources ¶

_write_batch ¶

_write_messages ¶

accepts ¶

add_source ¶

add_topic ¶

declare_topic ¶

delete_all_recordings ¶

delete_recording ¶

developer_ui ¶

log_message ¶

reindex_unindexed async ¶

rename_recording ¶

scan_recordings ¶

start ¶

stop async ¶

unindexed_recordings ¶

warn_converter_failure ¶

Events¶

RecordingInfo ¶

Events¶

RecordingSource ¶

Events¶

RecordingsPage ¶

TopicSchema ¶

Events¶

current_recording `property` ¶

disabled_topics `property` ¶

dropped_message_count `property` ¶

message_count `property` ¶

recordings `property` ¶

topics `property` ¶

_refresh_stats `async` ¶

reindex_unindexed `async` ¶

stop `async` ¶