Design

kazunoko is built with a 3-layer architecture, where each layer has a distinct responsibility. Higher layers are closer to the user (easier to use), while lower layers are closer to the hardware (more flexible).

---

Layer 1: Device Layer

Responsibility: Manage communication with the detector

This layer handles all serial communication with the detector. It sends commands and receives responses.

Key Classes

• PortDevice: Communicates with real hardware
  • Connects via USB/Serial port
  • Sends commands and receives responses
• MockDevice: Simulated device for testing
  • Works without physical hardware
  • Useful for development and testing

Device Layer Example

from kazunoko import connect, MockDevice

# Connect to real detector
device = connect(port="/dev/ttyUSB0")

# Or use mock device for testing
device = MockDevice()

# Low-level operation: send raw command
response = device.query("STATUS")

---

Layer 2: Measurement Layer

Responsibility: Manage event reading and measurement sessions with flexible streaming options

Provides base class and specialized classes for different measurement scenarios: simple streaming or integrated measurements with metadata.

Key Classes

• Streamer: Base class providing unified streaming interface (v0.1.50+)

  • stream_by_count(n): Generator yielding N events (memory-efficient)
  • stream_by_time(duration): Generator yielding events for duration (memory-efficient)
  • read_by_count(n): Convenience method returning list of N events
  • read_by_time(duration): Convenience method returning list of events for duration
  • Flexible pattern: generators for streaming, lists for batch collection
  • Abstract read_event() method for subclass customization

• Reader: Lightweight event reader for simple streaming

  • Inherits from Streamer (v0.1.50+)
  • No initialization overhead
  • Timeout retry logic for reliable event reading
  • Single events: read_event()
  • Streaming: stream_by_count(), stream_by_time(), read_by_count(), read_by_time()
  • Device time sync: set_rtc_time()
  • Suitable for cli.read command and simple streaming use cases

• Measure: Full-featured measurement session manager

  • Inherits from Streamer (v0.1.50+)
  • Configures device thresholds via initialize()
  • Collects device metadata (MAC, firmware version, thresholds)
  • Automatically merges metadata into all events
  • Single events: read_event() with metadata merged
  • Streaming: inherited methods with metadata merged into each event
  • Overrides streaming methods to enforce initialize() call before streaming
  • Device time sync: set_rtc_time()
  • Suitable for cli.measure command and systematic measurements with threshold configuration

Measurement Layer Example

from kazunoko import connect, Reader, Measure, MeasureConfig

# Simple streaming with Reader - generator (memory-efficient)
device = connect()
reader = Reader(device, event_timeout=5.0)
for event in reader.stream_by_count(1000000):
    process(event)  # Process one at a time

# Simple streaming with Reader - list (convenient)
events = reader.read_by_count(100)
events = reader.read_by_time(60.0)  # 60 seconds

# Integrated measurement with Measure - generator
config = MeasureConfig(thresholds={1: 300, 2: 300, 3: 300})
measure = Measure(device, config)
measure.setup()
for event in measure.stream_by_count(1000):
    process(event)  # Each event has metadata merged

# Integrated measurement with Measure - list
events = measure.read_by_count(100)
events = measure.read_by_time(60.0)

---

Layer 4: Command Layer

Responsibility: Provide user-friendly methods for device operations

Wraps the Device Layer with convenient, high-level methods for common tasks.

Key Class

• Command: High-level interface for detector operations
  • Works with any DeviceProtocol implementation
  • Provides discoverable, IDE-friendly methods

Common Methods

Method	Purpose	Example
status()	Get current device status	cmd.status()
version()	Get firmware version	cmd.version()
poll_count(n)	Set polling count	cmd.poll_count(100)
threshold(ch, value)	Set detection threshold	cmd.threshold(0, 500)
read()	Read one detection event	cmd.read()
reset()	Reset the device	cmd.reset()

Command Layer Example

from kazunoko import connect, Command

device = connect(port="/dev/ttyUSB0")
cmd = Command(device)

# Get status
status = cmd.status()
print(f"Version: {status.version}")

# Configure device
cmd.poll_count(100)
cmd.threshold(0, 500)

# Read event
event = cmd.read()
print(f"Channel 0: {event.ch0}")

---

Layer 5: Parser Layer

Responsibility: Convert JSON responses to Python objects

The device returns responses in JSON format. This layer validates and converts them to usable Python objects.

Key Classes

• Response: Pydantic model representing device responses
• ResponseParser: Validates and parses JSON responses
• parse_jsonl(): Convenience function for one-liner parsing

Parser Layer Example

from kazunoko import parse_jsonl

# Parse JSON response
json_text = '{"type": "response", "version": "1.0.0", "status": "ok"}'
response = parse_jsonl(json_text)

print(response.version)  # "1.0.0"
print(response.type)     # "response"

---

Data Flow

Here's what happens when you call cmd.status():

sequenceDiagram
    participant User
    participant Command
    participant Device
    participant Hardware
    participant Parser

    User->>Command: cmd.status()
    Command->>Device: device.query("GET_STATUS")
    Device->>Hardware: Send "GET_STATUS\n"
    Hardware-->>Device: {"type": "response", ...}
    Device->>Parser: Parse JSON
    Parser-->>Device: Response object
    Device-->>Command: Response object
    Command-->>User: Response object

---

Why This Architecture?

Benefit 1: Separation of Concerns

• Device Layer: Handles all hardware communication
• Command Layer: Provides user-friendly API
• Parser Layer: Validates and transforms data

Each layer is independent, so changes to one don't affect others.

Benefit 2: Easy Testing

Use MockDevice instead of real hardware:

# Test without hardware
device = MockDevice()
cmd = Command(device)
status = cmd.status()  # Same code works!

Benefit 3: Extensibility

Need a different communication method (Bluetooth, Network, etc.)? Just extend the Device Layer. Command and Parser remain unchanged.

---

Which Layer to Use

Use Device Layer when

• You need low-level control
• You want to send custom commands
• You need to understand communication details

device = connect()
response = device.query("CUSTOM_CMD")

Use Measurement Layer when

• You need to read detection events
• Simple streaming: Use Reader for lightweight event reading
• Integrated measurements: Use Measure for systematic measurements with metadata

# Simple streaming
reader = Reader(device, event_timeout=5.0)
event = reader.read_event()

# Integrated measurement
config = MeasureConfig(thresholds={1: 300, 2: 300, 3: 300})
measure = Measure(device, config)
measure.setup()
event_with_metadata = measure.read_event()

Use Command Layer (Recommended)

• You want standard detector operations
• You want simple, readable code
• You like IDE autocompletion for commands

cmd = Command(device)
cmd.status()
cmd.poll_count(100)

---

Quick Reference

Layer	Role	When to Use
Parser	JSON ↔ Python	Automatic (used internally)
Device	Serial communication	Custom operations needed
Measurement	Event reading	Reading detection events
Command	High-level API	Standard operations (recommended)

Best Practice

Start with the Command Layer. Use lower layers only when you need their specific capabilities.

---

Class Diagrams

Measurement Layer

classDiagram
    class Streamer {
        #_cmd: Command
        #_event_timeout: float
        +read_event() Response*
        +stream_by_count(count) Generator
        +stream_by_time(duration) Generator
        +read_by_count(count) list
        +read_by_time(duration) list
    }

    class Reader {
        +read_event() Response
        +set_rtc_time()
    }

    class Measure {
        -_config: MeasureConfig
        -_metadata: MeasureMetadata
        +initialize() MeasureMetadata
        +set_rtc_time()
        +read_event() Response
        +stream_by_count(count) Generator
        +stream_by_time(duration) Generator
        +metadata property
        +config property
    }

    class MeasureConfig {
        +thresholds: dict
        +poll_count: int
        +event_timeout: float
    }

    class MeasureMetadata {
        +threshold1: int
        +threshold2: int
        +threshold3: int
        +mac_address: str
        +kurikintons: str
        +kazunoko: str
    }

    Streamer <|-- Reader : inherits
    Streamer <|-- Measure : inherits
    Reader --> DeviceProtocol : uses
    Measure --> DeviceProtocol : uses
    Measure --> MeasureConfig : uses
    Measure --> MeasureMetadata : creates

Command & Device Layers

classDiagram
    class Command {
        -_device: DeviceProtocol
        +status() Response
        +version() Response
        +poll_count(n) Response
        +threshold(ch, value) Response
        +read() Response
        +reset() Response
    }

    class DeviceProtocol {
        <<interface>>
        +query(cmd: str) Response
        +receive_response(field_type: str) Response
    }

    class PortDevice {
        -port: str
        -baudrate: int
        -timeout: float
        +__enter__()
        +__exit__()
        +query(cmd: str) Response
        +receive_response(field_type: str) Response
    }

    class MockDevice {
        -generator: MockGenerator
        +__enter__()
        +__exit__()
        +query(cmd: str) Response
        +receive_response(field_type: str) Response
    }

    class MockGenerator {
        -events: list
        -speed: float
        -jitter: float
        +from_file(path) MockGenerator
        +from_random(count, seed) MockGenerator
        +set_speed(multiplier) MockGenerator
        +set_jitter(seconds) MockGenerator
    }

    Command --> DeviceProtocol : uses
    DeviceProtocol <|.. PortDevice : implements
    DeviceProtocol <|.. MockDevice : implements
    MockDevice --> MockGenerator : uses

Parser Layer

classDiagram
    class Response {
        +type: str
        +status: str*
        +model_dump() dict
        +model_dump_json() str
    }

    class ResponseParser {
        <<static>>
        +validate_response(data) Response
        +parse_jsonl(json_text) Response
    }

    class ResponseView {
        +raw: Response
        +flat: Response
    }

    class ResponseFormatter {
        +display(response: Response, format: str)
    }

    ResponseParser --> Response : creates
    ResponseView --> Response : wraps
    ResponseFormatter --> ResponseView : uses
    ResponseFormatter --> Response : displays

Full Architecture

classDiagram
    class Streamer {
        #_cmd: Command
        #_event_timeout: float
        +read_event() Response*
        +stream_by_count(count) Generator
        +stream_by_time(duration) Generator
        +read_by_count(count) list
        +read_by_time(duration) list
    }

    class Reader {
        +read_event() Response
        +set_rtc_time()
    }

    class Measure {
        -_config: MeasureConfig
        -_metadata: MeasureMetadata
        +initialize() MeasureMetadata
        +read_event() Response
        +stream_by_count(count) Generator
        +stream_by_time(duration) Generator
        +set_rtc_time()
        +metadata property
        +config property
    }

    class MeasureConfig {
        +thresholds: dict
        +poll_count: int
        +event_timeout: float
    }

    class MeasureMetadata {
        +threshold1: int
        +threshold2: int
        +threshold3: int
        +mac_address: str
        +kurikintons: str
        +kazunoko: str
    }

    class Command {
        -_device: DeviceProtocol
        +status() Response
        +read() Response
        +threshold(ch, value) Response
        +thresholds() Response
    }

    class DeviceProtocol {
        <<interface>>
        +query(cmd: str) Response
    }

    class PortDevice {
        +query(cmd: str) Response
    }

    class MockDevice {
        +query(cmd: str) Response
    }

    class Response {
        +type: str
        +status: str
    }

    class ResponseParser {
        <<static>>
        +validate_response()
    }

    class ResponseView {
        +raw: Response
        +flat: Response
    }

    Streamer <|-- Reader : inherits
    Streamer <|-- Measure : inherits
    Reader --> Command : uses
    Measure --> Command : uses
    Measure --> MeasureConfig : uses
    Measure --> MeasureMetadata : creates
    Command --> DeviceProtocol : uses
    DeviceProtocol <|.. PortDevice : implements
    DeviceProtocol <|.. MockDevice : implements
    PortDevice --> Response : returns
    MockDevice --> Response : returns
    ResponseParser --> Response : creates
    ResponseView --> Response : wraps