analysis

McapRecorder ¶

McapRecorder(
    *,
    output_dir: Path | str = "~/.rosys/mcap",
    max_part_size_mb: float = 100,
    max_part_duration: float | None = None,
    max_total_size_mb: float = 1000,
    max_kept_size_mb: float = 500,
    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.

A run is written as numbered parts, <run>_<part>.mcap, into the parts/ folder of the output directory, rotated by size and duration; every .mcap at the top level of the output directory is a kept recording (renamed, merged or placed there deliberately). The folder decides the kind, not the name. Peak disk usage is max_total_size_mb + max_part_size_mb: the budget is enforced only before a part is opened, so the growing part can exceed it by up to one part's worth. The budget deletes parts (oldest first) before kept recordings, which have a bound of their own so they cannot eat the rolling window. A merge holds its sources open, so the space the budget frees during one only returns when the merge finishes, and the peak then also carries the merge's size.

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); parts go into its `parts/` folder, every `.mcap` at its top level counts as kept.	`'~/.rosys/mcap'`
`max_part_size_mb`	`float`	on-disk size at which the active part is rotated to a new one.	`100`
`max_part_duration`	`float \| None`	seconds after which the active part is rotated to a new one (default: no duration-based rotation). Checked as messages are written, so an idle recording only rotates once data flows again.	`None`
`max_total_size_mb`	`float`	disk budget for the output directory including `parts/`; the oldest recordings are deleted before a part is opened to stay under it, parts first. Peak disk usage is therefore `max_total_size_mb + max_part_size_mb`, as the growing part is not counted.	`1000`
`max_kept_size_mb`	`float`	bound for kept recordings within the budget; beyond it the oldest are deleted, sparing the newest (the one just filed away) however large it is.	`500`
`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.

merging `property` ¶

merging: frozenset[str]

The names of the recordings being merged right now (see :meth:merge).

message_count `property` ¶

message_count: int

Number of messages written to the current recording.

recordings `property` ¶

recordings: list[Path]

All parts and kept recordings, newest first (by modification time, so renames keep the order).

run_name `property` ¶

run_name: str

The running recording's name, <timestamp> or <timestamp>_<name>; empty before the first start.

topics `property` ¶

topics: list[str]

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

_collect_disk_stats ¶

_collect_disk_stats() -> _DiskStats

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

_drain_and_close ¶

_drain_and_close() -> tuple[Path | None, int]

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

The final drain may rotate, so the path is read after writing; _close_file runs in a finally so even a raising write still finalizes the open file.

Returns:

Type	Description
`tuple[Path \| None, int]`	the finalized file (`None` if none was open) and the number of messages it holds.

_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_disk_budget ¶

_enforce_disk_budget() -> None

Delete the oldest recordings until the kept ones fit max_kept_size and all fit max_total_size.

The kept bound spares the newest kept recording; the total is then met by parts first, oldest first.

_enforce_queue_cap ¶

_enforce_queue_cap() -> bool

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

The oldest messages are dropped until the queue is under both :attr:max_queued_messages and :attr:max_queued_bytes (see :data:MAX_QUEUED_BYTES).

Returns:

Type	Description
`bool`	whether the drop is due to be logged, at most once per `_DROP_WARNING_INTERVAL`.

_file_is_expired ¶

_file_is_expired() -> bool

Whether the current part has been open longer than max_part_duration.

_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

_recording_files ¶

_recording_files() -> list[Path]

Every part in parts/ and every kept recording at the top level of the output directory.

_refresh_stats `async` ¶

_refresh_stats() -> None

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

_remove_orphaned_temporary_files ¶

_remove_orphaned_temporary_files() -> None

Remove reindex and merge temp files left by a crash. Run once at construction, never during operation.

Deleting one mid-run would destroy the result, so this never runs from the disk-budget path. Orphans are otherwise invisible to the budget, scan and UI (they do not match *.mcap). A reindex stages beside the file it rebuilds, in either folder; a merge beside its top-level target.

_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 run — 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 part or a kept recording (only within this recorder's folders).

The currently-recording part 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`

merge `async` ¶

merge(
    sources: list[Path],
    name: str,
    *,
    start_time_ns: int = 0
) -> Path | None

Merge finished recordings into the kept recording <name>.mcap, deleting them once it is in place.

Runs off the loop; a failed merge leaves the sources untouched (see :func:~.merging.merge_into_place), a source that cannot be deleted afterwards is only logged.

Parameters:

Name	Type	Description	Default
`sources`	`list[Path]`	finished parts or kept recordings of this recorder, oldest first.	required
`name`	`str`	the name of the merged recording, without `.mcap`.	required
`start_time_ns`	`int`	messages logged before this time are dropped (default: keep all).	`0`

Returns:

Type	Description
`Path \| None`	the merged recording, or `None` if no message was left to merge (the sources are kept).

Raises:

Type	Description
`ValueError`	if `name` is no plain file name, or a source is no finished recording of this recorder.
`FileExistsError`	if `<name>.mcap` exists or is being merged already.
`RuntimeError`	if the app shut down before the merge ran; the sources are kept.

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 part or a kept recording, filing it away as kept at the top level of the output directory.

The currently-recording part cannot be renamed (the writer holds it open). Empty, whitespace-only or dots-only names would escape the output directory and are rejected.

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,
    *,
    name: str | None = None,
    metadata: dict | None = None
) -> str | 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`
`name`	`str \| None`	appended to the start time to name the run, `<timestamp>_<name>` (default: the start time alone). The run's parts are numbered `<run>_01.mcap`, `<run>_02.mcap`, ... as they rotate, so they sort and read as a unit.	`None`
`metadata`	`dict \| None`	written into every part of this run as a JSON metadata record, so a part carries the context it was recorded in.	`None`

Returns:

Type	Description
`str \| None`	the name of the run, or `None` if a recording is already running or the previous one is still being finalized.

Raises:

Type	Description
`ValueError`	if `name` is not a plain file name.
`TypeError`	if `metadata` cannot be serialized to JSON.

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; start() is refused meanwhile. A final part without messages is discarded, otherwise RECORDING_STOPPED is emitted with its 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 every rotation
RECORDING_STOPPED	a recording file has been finalized (argument: path); emitted per file, including on every rotation

logging_page ¶

LoggingPage ¶

LoggingPage(group_names: list[str] | None = None)

Logging Page

This module creates a page to change the log levels of different loggers. A list of logger names like ["field_friend", "rosys"] can be passed to group them together. It is mounted at /logging.

logs_page ¶

logs_page(*, logs_dir: Path | None = None)

Logs Page

This module creates a page to list and download log files. It is mounted at /logs.

profile_button ¶

profile_button()

Bases: button

The profile button allows starting and stopping a profiling session.

Use the profiling.profile decorator for including functions or methods in the analysis. The results are shown on the console.

recordings_page ¶

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

Lists the MCAP recordings for download and deletion.

A long recording rotates into many files, so the parts of one run are listed as a single entry that expands into its parts and can be merged into one recording; any other file stays a row of its own. 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 kept recordings and parts over HTTP by basename from the top level or the parts folder (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`

analysis

McapRecorder ¶

current_recording property ¶

disabled_topics property ¶

dropped_message_count property ¶

merging property ¶

message_count property ¶

recordings property ¶

run_name property ¶

topics property ¶

_collect_disk_stats ¶

_drain_and_close ¶

_emit_on_loop ¶

_enforce_disk_budget ¶

_enforce_queue_cap ¶

_file_is_expired ¶

_hard_stop ¶

_recording_files ¶

_refresh_stats async ¶

_remove_orphaned_temporary_files ¶

_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 ¶

merge async ¶

reindex_unindexed async ¶

rename_recording ¶

scan_recordings ¶

start ¶

stop async ¶

unindexed_recordings ¶

warn_converter_failure ¶

Events¶

logging_page ¶

LoggingPage ¶

logs_page ¶

profile_button ¶

recordings_page ¶

current_recording `property` ¶

disabled_topics `property` ¶

dropped_message_count `property` ¶

merging `property` ¶

message_count `property` ¶

recordings `property` ¶

run_name `property` ¶

topics `property` ¶

_refresh_stats `async` ¶

merge `async` ¶

reindex_unindexed `async` ¶

stop `async` ¶