> ## Documentation Index
> Fetch the complete documentation index at: https://docs.roboticks.io/llms.txt
> Use this file to discover all available pages before exploring further.

# MCAP Capture Reference

> Reference for the mcap_capture context manager - topics, allowlist, compression, upload policy, storage paths.

# MCAP capture reference

`roboticks.mcap_capture` is a context manager that records selected ROS2 topics to an MCAP file for the duration of a `with` block. For a tutorial introduction, see [Testing → MCAP capture](/testing/mcap-capture); this page is the API reference.

```python theme={null}
from roboticks import mcap_capture
```

<Warning>
  **Requires the `mcap` extra.** Install with `pip install 'roboticks[mcap]'`. The bare `roboticks` package does not pull in the `mcap` or `mcap-ros2-support` libraries.
</Warning>

## Signature

```python theme={null}
def mcap_capture(
    *,
    topics: list[str] | None = None,
    path: str | Path | None = None,
    upload_on: Literal["failure", "always", "never"] = "failure",
    compression: Literal["zstd", "lz4", "none"] = "zstd",
    chunk_size_bytes: int = 1 << 20,
    qos: rclpy.qos.QoSProfile | None = None,
    node: rclpy.node.Node | None = None,
) -> ContextManager[McapHandle]: ...
```

## Parameters

| Parameter          | Type                                    | Default                       | Behaviour                                                                              |
| ------------------ | --------------------------------------- | ----------------------------- | -------------------------------------------------------------------------------------- |
| `topics`           | `list[str] \| None`                     | `["*"]`                       | Allowlist. Wildcards `*` and `**` per rosbag2 semantics. `None` is treated as `["*"]`. |
| `path`             | `str \| Path \| None`                   | `test_results/<test_id>.mcap` | Local write path. Parents are `mkdir -p`'d.                                            |
| `upload_on`        | `Literal["failure", "always", "never"]` | `"failure"`                   | When the runner uploads the bag to the platform.                                       |
| `compression`      | `Literal["zstd", "lz4", "none"]`        | `"zstd"`                      | Per-chunk compression.                                                                 |
| `chunk_size_bytes` | `int`                                   | 1 MiB                         | MCAP chunk size before flush.                                                          |
| `qos`              | `QoSProfile`                            | rclpy default                 | QoS for subscribed topics.                                                             |
| `node`             | `Node`                                  | ephemeral                     | Reuse an existing node.                                                                |

## Returns (as context value)

An `McapHandle` with:

| Attribute       | Description                                           |
| --------------- | ----------------------------------------------------- |
| `path`          | Final local path of the MCAP file.                    |
| `bytes_written` | Bytes written at context exit.                        |
| `topic_counts`  | `dict[str, int]` — messages captured per topic.       |
| `uploaded`      | `bool` — set in the `__exit__` after upload decision. |
| `upload_url`    | `str \| None` — platform URL if uploaded.             |

## Behaviour

### Allowlist semantics

| Pattern      | Matches                            |
| ------------ | ---------------------------------- |
| `/topic`     | exact                              |
| `/parent/*`  | immediate children of `/parent`    |
| `/parent/**` | recursive descendants of `/parent` |
| `*`          | all topics, immediate root level   |
| `**`         | all topics, any depth              |

Subscriptions are created on first-seen — the recorder subscribes when a publisher appears on a matching topic, not at context entry. This avoids hangs waiting for topics that won't appear.

### Upload behaviour

| `upload_on` | After context exit                                                                                                  |
| ----------- | ------------------------------------------------------------------------------------------------------------------- |
| `"failure"` | Uploads iff the surrounding test failed. Default.                                                                   |
| `"always"`  | Uploads unconditionally.                                                                                            |
| `"never"`   | Local file only. The runner does not upload; the file is part of the test artifacts collected per workspace policy. |

The "test failed" signal is read from the pytest test status at session-end. For non-pytest contexts (e.g. raw scripts), `"failure"` is treated as `"always"`.

### Threading

The recorder runs in a background thread spun up at context entry. It uses a dedicated `rclpy` executor; subscriptions don't interfere with the test's main executor.

### Compression

| Option | Wall-clock cost        | File size    | Use when                                           |
| ------ | ---------------------- | ------------ | -------------------------------------------------- |
| `zstd` | \~5% over uncompressed | \~30% of raw | default; nearly free                               |
| `lz4`  | \~2%                   | \~50% of raw | very high-rate topics where zstd CPU is noticeable |
| `none` | 0%                     | 100%         | only when re-compressing downstream                |

## Examples

### Default (allow-all, upload-on-failure)

```python theme={null}
with mcap_capture() as bag:
    run_test()
# bag.path, bag.bytes_written, bag.topic_counts, bag.uploaded available here
```

### Explicit topic allowlist

```python theme={null}
with mcap_capture(
    topics=["/cmd_vel", "/safety/*", "/tf"],
    path="captures/estop.mcap",
) as bag:
    run_test()
print(bag.topic_counts)
# {'/cmd_vel': 142, '/safety/estop_engaged': 1, '/tf': 880}
```

### Always upload

```python theme={null}
with mcap_capture(upload_on="always") as bag:
    run_test()
assert bag.uploaded
assert bag.upload_url.startswith("https://app.roboticks.io/runs/")
```

### Pytest fixture for session-scoped recording

```python theme={null}
import pytest
from roboticks import mcap_capture

@pytest.fixture(scope="session")
def session_bag():
    with mcap_capture(
        topics=["**"],
        path="captures/session.mcap",
        upload_on="always",
    ) as bag:
        yield bag
```

## Limits

| Limit                               | Value   | Behaviour                                                                     |
| ----------------------------------- | ------- | ----------------------------------------------------------------------------- |
| Max in-flight bytes                 | 256 MiB | Excess buffered to disk-spill chunk.                                          |
| Max bag file size                   | 4 GiB   | Recorder rolls into `<path>.1.mcap`, `.2.mcap`, …; upload includes all parts. |
| Max recorded topics                 | 1024    | More than 1024 distinct topics raises `LimitExceeded`.                        |
| Max concurrent captures per process | 4       | More than 4 nested `mcap_capture` blocks raises.                              |

## File layout

```
<test_results_dir>/
  <test_id>.mcap            # default per-test path
  captures/
    estop.mcap              # explicit path
  uploads.manifest.json     # what the runner uploaded, where
```

The `uploads.manifest.json` is read by `rbtk` when correlating local artifacts to a remote run.

## Next

<CardGroup cols={2}>
  <Card title="MCAP capture tutorial" icon="floppy-disk" href="/testing/mcap-capture">
    Worked examples and storage-cost guidance.
  </Card>

  <Card title="LLM triage" icon="wand-magic-sparkles" href="/llm/triage">
    How the platform reads your MCAP to explain failures.
  </Card>

  <Card title="Fault injection" icon="bug" href="/sdk/fault-injection">
    Pair with `mcap_capture` for failure forensics.
  </Card>

  <Card title="Pricing" icon="dollar-sign" href="/pricing">
    Hot and archive storage costs.
  </Card>
</CardGroup>
