perf: pass channel topics down to the MCAP reader

[mulkakhileshmj]

Closes #470.

Problem

Episode.channel() selects a channel by channel_id, but that never became a topic filter for the underlying MCAP reader, so iter_messages(topics=None) yielded every stream in the file and HFlow discarded the unrelated ones in Python. Reading a small state or action channel could decompress gigabytes of camera data for nothing.

Fix

I took Option 2 from the issue: PythonMcapEpisodeReader.iter_batches() now derives the topics for the requested channel_ids from the file summary and passes them to iter_messages(), so the seeking reader can skip unrelated chunks. This covers every channel_ids caller, including Episode.iter_decoded_batches(), not just Episode.channel().

Correctness cases from the issue are preserved:

• The channel-id filter is unchanged. Several channels can share one topic, and topic filtering only narrows the read, it never selects.
• Caller-supplied topics are never overridden.
• A file with no summary section, or all-unknown channel ids, falls back to the unconstrained read, preserving current behavior.

Coverage

Two tests added to tests/test_multichannel.py, tracing at the mcap library boundary:

• Two topics in separate chunks: requesting /target passes topics=["/target"] down and only /target messages cross the boundary, with identical returned data.
• Two channels sharing one topic (the file's existing fixture): the shared topic is passed down, both channels are yielded by mcap, and exactly the requested channel's payloads come back, pinning that channel-id selection still wins.

Validation

Run on WSL2 Ubuntu, Python 3.12:

uv sync --locked
uv run ruff check          # All checks passed
uv run ruff format --check # already formatted
uv run ty check            # All checks passed
uv run pytest -q           # 1713 passed, 6 skipped

Mutation check: with src/hflow/reader.py reverted to main, both new tests fail; restored, they pass. The touched test file also passes under Python 3.11.

[github-actions]

👋 Hi @mulkakhileshmj — thank you so much for your first contribution to HFlow!

A maintainer will review your pull request as soon as possible. In the meantime:

• ⭐ Star the repo: https://github.com/Hebbian-Robotics/hflow
• 💬 Join our Discord: https://discord.gg/vacepQvjmg
• 📖 Check CONTRIBUTING.md for the validation commands CI will run
• 📣 Share the project with your network

💡 Tip: one open pull request per contributor at a time. Issues with an assignee are taken; everything else is fair game.

We are excited to have you here and appreciate your help making the project better! 🙌

[kstonekuan]

The layer you picked is the right one and I want this. Episode.iter_batches at episode.py:477 forwards channel_ids from its own callers and has the same gap, and #475 does not touch it. Deriving inside iter_batches covers that and anything added later. The channels() result is cached, so the lookup costs one dict build per reader.

Tracing the stock mcap_reader.iter_messages and asserting topics_yielded is the stronger test shape too. #475 asserts the argument was forwarded; yours asserts the underlying read actually narrowed.

The problem is sequencing. #475 merged an hour ago and now passes topics=[info.topic] from Episode.channel(), so the topics is None branch no longer fires on the path your tests take. Rebased onto main, I deleted your entire derivation block and your tests still pass:

### #476 derivation block removed entirely
    8 passed in 0.95s
    (nothing noticed)

So the code is right and nothing holds it. Gate is clean, 1715 passed.

What it needs: a test that reaches iter_batches with channel_ids set and topics left None. Episode.iter_batches(channel_ids=[...]) is the real caller with that shape, and it is the one your change exists for. Driving PythonMcapEpisodeReader.iter_batches directly would also do it and is closer to the unit.

Keep the Episode.channel() tests. They still pass for a reason worth pinning, they just belong to #475's change now.

Two smaller things while you are in there.

The except ValueError fallback for a file with no summary: nothing exercises it. Worth a case, or say why it is unreachable.

channels() raising anything other than ValueError would escape. Check what it actually raises on a truncated file before deciding the catch is wide enough.

[mulkakhileshmj]

Thanks for the review, and for catching the sequencing with #475. You are right: once it passed the topic explicitly on the channel() path, my tests stopped reaching the derivation. Fixed as suggested, plus the two smaller points, measured rather than assumed.

The missing pin. A new test drives Episode.iter_decoded_batches(channel_ids=...), the remaining caller that reaches the reader with topics left None. It asserts the derived topic is what the mcap reader receives and that only the requested stream crosses the library boundary. Mutation checked: with the derivation block deleted, this test fails; restored, it passes. The two Episode.channel() tests stay unchanged, agreed they now pin #475's forwarding.

The no-summary fallback is reachable, and now exercised. A file written with use_chunking=False, use_statistics=False, use_summary_offsets=False, repeat_channels=False, repeat_schemas=False has no summary section at all: get_summary() returns None, channels() raises the ValueError, and a channel id read falls back to the unconstrained scan while still returning only the requested channel. The new test pins exactly that.

Truncated files, measured before deciding the catch width. On a file cut mid-data or inside the summary region, channels() raises mcap's RecordLengthLimitExceeded, not ValueError, and iter_messages on the same file raises the identical error whether or not the derivation catches it. Widening the catch would only move the same failure two lines down, so it stays narrow and the code comment now records that reasoning.

One side finding from the same probe, reported rather than acted on: a chunked file whose summary carries chunk indexes but no repeated channel records (repeat_channels=False with chunking on) makes the stock mcap SeekingReader itself raise KeyError from summary.channels inside iter_messages. That happens with or without this change, so it is mcap behavior on a degenerate writer configuration, not something this PR touches. Happy to file it separately if it is worth tracking.

Gate rerun on the updated branch: ruff, format, and ty clean; full suite 1717 passed, 6 skipped; the multichannel file also passes on Python 3.11. The three fixture writers now share one helper, so the file builds its two-topic layouts in one place.

[kstonekuan]

LGTM. Merging. This sat two days waiting on me rather than on you, which is my fault and not the usual turnaround.

You did the thing I care most about: you measured the truncated-file case before choosing the catch width, rather than widening except until the tests passed. channels() raising RecordLengthLimitExceeded and iter_messages raising the identical error on the same file is exactly the argument for keeping the catch narrow, because a wider one would have moved the same failure two lines down while looking like it handled something. That reasoning is now in the comment where the next person will find it.

The no-summary test is the one I would have asked for and you got there first, down to naming the writer flags that produce a file with no summary section at all.

Mutations:

derivation block removed entirely           1 failed  test_decoded_batches_by_channel_id_derive_the_topic_filter
channel-id filter dropped, topic filter kept 4 failed
no-summary fallback removed                 1 failed

The second row is the one that matters for #470's original shape: topic narrowing reduces what reaches the channel-id filter but cannot replace it, because several channels may share a topic. Your shared-topic test holds that.

One mutation came back unheld and I pushed a test for it. Dropping the topics is None half of the condition, so the derivation also fires when the caller passed topics explicitly, left the whole suite green:

derivation also overrides an explicit topics=    86 passed  (nothing noticed)

Episode.channel() passes both a topic and a channel id since #475, and for that caller the derived topic equals the explicit one, so nothing can tell. Asking for /target while naming a channel on /camera separates them: the reader has to pass down what the caller asked for and yield nothing, rather than quietly reading the other stream instead. Without that condition it reads the other stream.

On your side finding, the stock SeekingReader raising KeyError from summary.channels on a chunked file with repeat_channels=False: yes please, file it. It is upstream behaviour on a degenerate writer configuration rather than ours, but a degenerate file someone hands us still has to fail comprehensibly, and knowing the shape is worth having written down even if the answer ends up being "refuse it in doctor and move on".

Validated on the rebased result: ruff, ruff format, ty, 1896 passed / 6 skipped.