Measurement Layer¶

Event streaming and measurement session management.

Design Intent¶

Event reading has two distinct use cases:

Simple streaming — read events as fast as possible, no setup required.
Measurement sessions — configure thresholds, sync the clock, collect metadata, merge it into every event.

Both share the same streaming interface (stream_by_count, stream_by_time). Streamer defines the interface; Reader and Measure are the two concrete implementations.

Class hierarchy¶

Streamer (abstract base)
├── Reader   — lightweight, no initialization
└── Measure  — full session with setup and metadata

Streamer¶

Defines the streaming interface used by all consumers (CLI, example scripts, custom code):

Method	Returns	Description
`stream_by_count(n)`	Generator	Yield N events one at a time
`stream_by_time(duration)`	Generator	Yield events for `duration` seconds
`read_by_count(n)`	list	Collect N events into memory
`read_by_time(duration)`	list	Collect events for `duration` seconds into memory
`read_event()`	DeviceResponse	Abstract — implemented by subclasses

Prefer generators (stream_by_*) for large collections to avoid accumulating events in memory.

Reader¶

Lightweight reader — no device configuration, no metadata, minimal overhead.

Implements read_event() with timeout retry logic: if the device does not return an event within event_timeout, it retries until the count or duration is satisfied.
Provides set_rtc_time() for standalone clock sync.
Used by the kazunoko read CLI command.

from kazunoko import connect, Reader

with connect() as device:
    reader = Reader(device, event_timeout=5.0)

    # Generator (memory-efficient for large counts)
    for event in reader.stream_by_count(10_000):
        process(event)

    # List (convenient for small counts)
    events = reader.read_by_count(100)

Measure¶

Full measurement session manager.

setup() must be called before streaming. It configures thresholds, syncs the RTC, and collects metadata (MeasureMetadata: MAC address, firmware version, threshold values).
Every event yielded by stream_by_count / stream_by_time has the metadata fields merged in automatically.
stream_by_count and stream_by_time raise RuntimeError if setup() has not been called, preventing silent misconfigured sessions.
Used by the kazunoko measure CLI command and the get_events.py example script.

from kazunoko import connect, Measure, MeasureConfig, parse_thresholds

config = MeasureConfig(
    thresholds=parse_thresholds("1:300;2:320;3:310"),
    poll_count=10,
    event_timeout=5.0,
)

with connect() as device:
    measure = Measure(device, config)
    measure.setup()  # configure device + collect metadata

    for event in measure.stream_by_count(100):
        print(event.mac)        # from metadata
        print(event.threshold1) # from metadata
        print(event.signal1)    # from event

StreamingStats¶

Both Reader and Measure track skipped events internally via StreamingStats:

Field	Description
`total_yielded`	Successfully yielded events
`skipped_timeout`	Events skipped due to `EventTimeout`
`skipped_protocol`	Events skipped due to `ProtocolError`
`skipped_response`	Events skipped due to `ResponseError`

Properties: total_attempts, total_skipped, success_rate.

Stats are logged and displayed when --verbose is enabled.

MeasureConfig and MeasureMetadata¶

MeasureConfig holds the session parameters passed into Measure:

thresholds: dict[int, int] — channel → value mapping
poll_count: int — polling cycle depth
event_timeout: float — seconds to wait per event

MeasureMetadata is collected by setup() from the device and merged into events:

threshold1, threshold2, threshold3
mac_address
Firmware version fields