get_runs.py

Purpose: Collect detector events split across multiple files (runs)

Use case: Long-term data acquisition, splitting data into manageable file sizes

Features:

• Automatically generates and saves multiple files in sequence
• Each run (session) is written to a separate file
• Sequential file IDs assigned automatically
• Device setup (RTC sync, metadata) performed only once for efficiency
• Progress bar output
• Verbose logging support

Output filename format:

YYYYMMDD/events_{MAC}_{timestamp}_0000000.jsonl
YYYYMMDD/events_{MAC}_{timestamp}_0000001.jsonl
YYYYMMDD/events_{MAC}_{timestamp}_0000002.jsonl
...

Usage:

# 1000 events × 10 files
uv run examples/get_runs.py "1:300;2:320;3:310" 1000 10

# Time-based (60 seconds × 10 files)
uv run examples/get_runs.py "1:300;2:320;3:310" 60 10 --use-sec

# With verbose logging redirected to a log file
uv run examples/get_runs.py "1:300;2:320;3:310" 100 10 --verbose 2> measure.log

# Test without hardware
uv run examples/get_runs.py "1:300;2:320;3:310" 100 10 --mock

CLI Options:

Option	Default	Description
THRESHOLDS	(required)	Threshold config as ch1:value;ch2:value;ch3:value
EVENTS_OR_SECS	(required)	Number of events (--use-event) or duration in seconds (--use-sec)
FILES	(required)	Number of files to collect
--use-event / --use-sec	--use-event	Collect by event count or by duration
--port / -p	auto	Serial port (auto for auto-detection)
--timeout / -t	0.1	Serial communication timeout in seconds
--poll-count / -pc	100	Events to poll per cycle (1–65535)
--event-timeout / -et	5.0	Timeout in seconds to wait for an event
--verbose / --quiet	--quiet	Show or suppress status messages
--mock	off	Use mock device for testing without hardware
--log-level	error	Log level (debug/info/error)

Implementation pattern:

# Device setup is performed only once
measure = Measure(detector, config)
measure.setup()

for file_id in range(files):
    output = generate_filename(mac_address=measure.metadata.mac_address, file_id=file_id)
    output.parent.mkdir(parents=True, exist_ok=True)
    with output.open("w", encoding="utf-8") as fd:
        file_console = Console(file=fd)
        for event in measure.stream_by_count(events_or_secs):
            file_console.print(event.model_dump_json(), soft_wrap=True)