mcap_json_message_write_and_read_quickstart.py

1import time
2from mcap.writer import Writer
3from mcap.reader import make_reader
4
5# 1. Write data to an MCAP file
6with open("example.mcap", "wb") as f:
7    writer = Writer(f)
8    writer.start()
9
10    # Register a schema (optional for some encodings, but good practice)
11    schema_id = writer.register_schema(
12        name="sample_data",
13        encoding="jsonschema",
14        data=b'{"type": "object", "properties": {"content": {"type": "string"}}}',
15    )
16
17    # Register a channel
18    channel_id = writer.register_channel(
19        topic="/chatter",
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=b'{"content": "hello world"}',
29        publish_time=time.time_ns(),
30    )
31
32    writer.finish()
33
34# 2. Read data back 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"[{message.log_time}] {channel.topic}: {message.data.decode('utf-8')}")