mcap_file_write_read_with_schema_and_channel.py

1import time
2from mcap.writer import Writer
3from mcap.reader import make_reader
4
5# 1. Write to an MCAP file
6with open("example.mcap", "wb") as f:
7    writer = Writer(f)
8    writer.start()
9
10    # Register a schema (optional, but recommended for structured data)
11    schema_id = writer.register_schema(
12        name="sample_data",
13        encoding="jsonschema",
14        data='{"type": "object", "properties": {"value": {"type": "number"}}}'.encode(),
15    )
16
17    # Register a channel
18    channel_id = writer.register_channel(
19        topic="/data",
20        message_encoding="json",
21        schema_id=schema_id,
22    )
23
24    # Write a message
25    writer.add_message(
26        channel_id=channel_id,
27        log_time=time.time_ns(),
28        data='{"value": 42}'.encode(),
29        publish_time=time.time_ns(),
30    )
31
32    writer.finish()
33
34# 2. Read from the MCAP file
35with open("example.mcap", "rb") as f:
36    reader = make_reader(f)
37    for schema, channel, message in reader.iter_messages():
38        print(f"Topic: {channel.topic}")
39        print(f"Timestamp: {message.log_time}")
40        print(f"Data: {message.data.decode()}")