Device¶

Connect to the OSECHI detector via serial port and send commands. PortDevice handles the actual serial communication, while DeviceProtocol defines the interface that both real and mock devices follow. Use connect() to create a device connection.

Architecture¶

Class Diagram¶

classDiagram
    class DeviceProtocol {
        <<interface>>
        +send_command(command: str) bool
        +receive_response(field_type: str) DeviceResponse
        +query(command: str) DeviceResponse
        +close() None
        +__enter__() Self
        +__exit__() None
    }

    class PortDevice {
        -port: str
        -baudrate: int
        -timeout: float
        -device: serial.Serial
        +__init__(port, baudrate, timeout)
        +send_command(command: str) bool
        +receive_response(field_type: str) DeviceResponse
        +query(command: str) DeviceResponse
        +close() None
    }

    class Serial {
        <<external>>
        -port: str
        -baudrate: int
        -timeout: float
        -is_open: bool
        +write(data: bytes) int
        +readline() bytes
        +reset_input_buffer() None
        +flush() None
        +close() None
    }

    DeviceProtocol <|.. PortDevice: implements
    PortDevice --> Serial: uses

Device Selection Flow¶

flowchart TD
    A["device(port, mock, timeout)"] --> B{mock = True?}
    B -->|Yes| C["Return MockDevice<br/>from mock.py"]
    B -->|No| D{port = 'auto'?}
    D -->|Yes| E["detect_port()"]
    D -->|No| F["Use specified port"]
    E --> G["PortDevice(port, timeout)"]
    F --> G
    G --> H["Serial connection<br/>to detector"]

    C --> I["Testing without<br/>hardware"]
    H --> J["Real device<br/>communication"]

connect¶

`kazunoko.device.connect(port='auto', timeout=DEFAULT_TIMEOUT)` ¶

Factory function: connect to device

Convenience function for one-liner device connection. If port is "auto", automatically detects an available serial port.

Parameters:

Name	Type	Description	Default
`port`	`Literal['auto'] \| str`	Serial port or "auto" for auto-detection (default: "auto")	`'auto'`
`timeout`	`float`	Serial communication timeout in seconds (default: 0.1)	`DEFAULT_TIMEOUT`

Returns:

Type	Description
`PortDevice`	Connected PortDevice instance

Raises:

Type	Description
`DeviceError`	If connection fails or no port found

Example

device = connect()  # Auto-detect
device = connect("auto")  # Explicit auto-detection
device = connect("/dev/ttyUSB0")  # Specific port
device = connect("/dev/ttyUSB0", timeout=0.5)  # Custom timeout
response = device.query("STATUS")
device.close()

Source code in src/kazunoko/device.py

def connect(
    port: Literal["auto"] | str = "auto",
    timeout: float = DEFAULT_TIMEOUT,
) -> PortDevice:
    """
    Factory function: connect to device

    Convenience function for one-liner device connection.
    If port is "auto", automatically detects an available serial port.

    Args:
        port: Serial port or "auto" for auto-detection (default: "auto")
        timeout: Serial communication timeout in seconds (default: 0.1)

    Returns:
        Connected PortDevice instance

    Raises:
        DeviceError: If connection fails or no port found

    Example:
        ```python
        device = connect()  # Auto-detect
        device = connect("auto")  # Explicit auto-detection
        device = connect("/dev/ttyUSB0")  # Specific port
        device = connect("/dev/ttyUSB0", timeout=0.5)  # Custom timeout
        response = device.query("STATUS")
        device.close()
        ```
    """
    if port == "auto":
        logger.debug("Auto-detecting port for connection")
        port = detect_port()
        logger.debug("Port selected", extra={"port": port})

    return PortDevice(port=port, timeout=timeout)

detect_port¶

`kazunoko.device.detect_port()` ¶

Auto-detect available serial port for OSECHI detector

Returns the first available serial port in order of preference:

/dev/cu.usbserial* (macOS USB serial adapters)
/dev/ttyUSB* (Linux USB serial adapters)
/dev/ttyACM* (Arduino, etc.)
/dev/ttyS* (built-in serial)
Raises DeviceError listing available ports (no fallback)

Returns:

Type	Description
`str`	Serial port path

Raises:

Type	Description
`DeviceError`	If no serial port is found or no preferred port matches

Example

# Auto-detect the first available serial port
port = detect_port()
device = PortDevice(port=port)
response = device.query("STATUS")
device.close()

Source code in src/kazunoko/device.py

def detect_port() -> str:
    """
    Auto-detect available serial port for OSECHI detector

    Returns the first available serial port in order of preference:

    1. /dev/cu.usbserial* (macOS USB serial adapters)
    2. /dev/ttyUSB* (Linux USB serial adapters)
    3. /dev/ttyACM* (Arduino, etc.)
    4. /dev/ttyS* (built-in serial)
    5. Raises DeviceError listing available ports (no fallback)

    Returns:
        Serial port path

    Raises:
        DeviceError: If no serial port is found or no preferred port matches

    Example:
        ```python
        # Auto-detect the first available serial port
        port = detect_port()
        device = PortDevice(port=port)
        response = device.query("STATUS")
        device.close()
        ```
    """

    logger.debug("Auto-detecting serial port")
    ports = comports()
    logger.debug("Found ports", extra={"count": len(ports), "ports": [p.device for p in ports]})

    if not ports:
        logger.error("No serial ports found")
        raise DeviceError("No serial ports found")

    # Prefer macOS USB serial ports
    for port_info in ports:
        if "cu.usbserial" in port_info.device:
            logger.debug("Selected port", extra={"port": port_info.device, "preference": "macOS USB"})
            logger.info("Port auto-detected", extra={"port": port_info.device})
            return port_info.device

    # Then Linux USB ports
    for port_info in ports:
        if "ttyUSB" in port_info.device:
            logger.debug("Selected port", extra={"port": port_info.device, "preference": "Linux USB"})
            logger.info("Port auto-detected", extra={"port": port_info.device})
            return port_info.device

    # Then ACM ports
    for port_info in ports:
        if "ttyACM" in port_info.device:
            logger.debug("Selected port", extra={"port": port_info.device, "preference": "ACM"})
            logger.info("Port auto-detected", extra={"port": port_info.device})
            return port_info.device

    # Then serial ports
    for port_info in ports:
        if "ttyS" in port_info.device:
            logger.debug("Selected port", extra={"port": port_info.device, "preference": "Serial"})
            logger.info("Port auto-detected", extra={"port": port_info.device})
            return port_info.device

    # No preferred port found - show available ports and raise error
    port_list = "\n".join(
        f"  - {port_info.device} ({port_info.description})"
        for port_info in ports
    )
    error_msg = (
        f"No suitable serial port found for OSECHI detector.\n\n"
        f"Available ports:\n{port_list}"
    )
    logger.error("No suitable port found", extra={"available_ports": [p.device for p in ports]})
    raise DeviceError(error_msg)

DeviceProtocol¶

`kazunoko.device.DeviceProtocol` ¶

Bases: Protocol

Protocol for device interface

Defines the interface that all device implementations must follow.

Source code in src/kazunoko/device.py

class DeviceProtocol(Protocol):
    """
    Protocol for device interface

    Defines the interface that all device implementations must follow.
    """

    def send_command(self, command: str) -> bool:
        """Send text command to device"""
        ...

    def receive_response(self, field_type: str) -> DeviceResponse:
        """Receive and parse JSONL response from device"""
        ...

    def query(self, command: str) -> DeviceResponse:
        """Send command and receive response in one operation"""
        ...

    def close(self) -> None:
        """Close device connection"""
        ...

    def __enter__(self):
        """Context manager entry"""
        ...

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit"""
        ...

`enter()` ¶

Context manager entry

Source code in src/kazunoko/device.py

def __enter__(self):
    """Context manager entry"""
    ...

`exit(exc_type, exc_val, exc_tb)` ¶

Context manager exit

Source code in src/kazunoko/device.py

def __exit__(self, exc_type, exc_val, exc_tb):
    """Context manager exit"""
    ...

`close()` ¶

Close device connection

Source code in src/kazunoko/device.py

def close(self) -> None:
    """Close device connection"""
    ...

`query(command)` ¶

Send command and receive response in one operation

Source code in src/kazunoko/device.py

def query(self, command: str) -> DeviceResponse:
    """Send command and receive response in one operation"""
    ...

`receive_response(field_type)` ¶

Receive and parse JSONL response from device

Source code in src/kazunoko/device.py

def receive_response(self, field_type: str) -> DeviceResponse:
    """Receive and parse JSONL response from device"""
    ...

`send_command(command)` ¶

Send text command to device

Source code in src/kazunoko/device.py

def send_command(self, command: str) -> bool:
    """Send text command to device"""
    ...

PortDevice¶

`kazunoko.device.PortDevice` ¶

Interface to OSECHI detector serial device

Single device management
Command transmission and response reception
Pre-parsed JSONL response return

Parameters:

Name	Type	Description	Default
`port`	`str`	Serial port (default: /dev/ttyUSB0)	`DEFAULT_PORT`
`baudrate`	`int`	Baud rate (default: 115200)	`DEFAULT_BAUDRATE`
`timeout`	`float`	Serial read timeout in seconds (default: 0.1)	`DEFAULT_TIMEOUT`

Raises:

Type	Description
`DeviceError`	If connection fails

Example

# Connect to device and send a command
device = PortDevice(port="/dev/ttyUSB0")
response = device.query("STATUS")
device.close()

# Using context manager for automatic cleanup
with PortDevice() as device:
    response = device.query("GET_VERSION")

Source code in src/kazunoko/device.py

class PortDevice:
    """
    Interface to OSECHI detector serial device

    - Single device management
    - Command transmission and response reception
    - Pre-parsed JSONL response return

    Args:
        port: Serial port (default: /dev/ttyUSB0)
        baudrate: Baud rate (default: 115200)
        timeout: Serial read timeout in seconds (default: 0.1)

    Raises:
        DeviceError: If connection fails

    Example:
        ```python
        # Connect to device and send a command
        device = PortDevice(port="/dev/ttyUSB0")
        response = device.query("STATUS")
        device.close()

        # Using context manager for automatic cleanup
        with PortDevice() as device:
            response = device.query("GET_VERSION")
        ```
    """

    def __init__(
        self,
        port: str = DEFAULT_PORT,
        baudrate: int = DEFAULT_BAUDRATE,
        timeout: float = DEFAULT_TIMEOUT,
    ):
        """Initialize device connection"""
        self.port = port
        self.baudrate = baudrate
        self.timeout = timeout
        self.device: serial.Serial | None = None

        try:
            logger.debug(
                "Opening serial port",
                extra={
                    "port": port,
                    "baudrate": baudrate,
                    "timeout": timeout,
                }
            )
            self.device = serial.Serial(
                port=port,
                baudrate=baudrate,
                timeout=timeout,
            )
            logger.success("Serial port opened successfully", extra={"port": port})
            logger.info("Device connected", extra={"port": port})
        except serial.SerialException as e:
            logger.error(
                "Failed to open serial port",
                extra={"port": port, "baudrate": baudrate},
                exc_info=True,
            )
            raise DeviceError(f"Failed to open serial port {port}: {e}") from e

    def send_command(self, command: str) -> bool:
        """
        Send text command to device with echo confirmation

        Args:
            command: Text command to send (e.g., "STATUS", "GET_VERSION")

        Returns:
            True if command sent successfully

        Raises:
            CommandError: If device not connected or send fails
        """
        if not self.device or not self.device.is_open:
            raise CommandError("Device not connected")

        # Append newline if not present
        if not command.endswith("\n"):
            command = command + "\n"

        try:
            logger.debug("Sending command", extra={"command": command.strip(), "port": self.port})
            # Send command
            self.device.reset_input_buffer()
            time.sleep(0.1)
            self.device.write(command.encode("utf-8"))
            self.device.flush()
            logger.success("Command sent successfully", extra={"command": command.strip(), "port": self.port})
            logger.info("Command executed", extra={"command": command.strip(), "port": self.port})
        except serial.SerialException as e:
            logger.error(
                "Failed to send command",
                extra={"command": command.strip(), "port": self.port},
                exc_info=True,
            )
            raise CommandError(f"Failed to send command: {e}") from e

        return True

    def receive_response(
        self,
        field_type: str,
    ) -> DeviceResponse:
        """
        Receive JSONL response from device and parse

        Args:
            field_type: Expected response type ("response" or "event")

        Returns:
            Parsed DeviceResponse object with received_us timestamp (in microseconds)

        Raises:
            ResponseTimeout: If no response received within timeout
            ResponseError: If response is invalid or malformed
        """
        if not self.device or not self.device.is_open:
            raise ResponseError("Device not connected")

        marker = f'"type":"{field_type}"'
        start_time = time.time()

        try:
            while time.time() - start_time < self.timeout:
                # Read one line from device
                line = self.device.readline().decode("utf-8", errors="ignore").strip()

                if not line:
                    time.sleep(0.005)
                    continue

                # Check if this is the expected type
                if marker not in line:
                    logger.debug("Received non-matching line", extra={"field_type": field_type, "port": self.port})
                    continue

                # Parse and validate
                received_at_us = int(time.time() * 1_000_000)
                parsed = parse_jsonl(line)
                if field_type is not None and parsed.type != field_type:
                    raise ResponseError(f"Expected {field_type}, got {parsed.type}")

                # Set reception timestamp in microseconds
                parsed.received_us = received_at_us
                return parsed

            raise ResponseTimeout(f"No response received within {self.timeout}s timeout")

        except ResponseTimeout:
            raise
        except ResponseError:
            logger.error(
                "Invalid response received",
                extra={"field_type": field_type, "port": self.port},
                exc_info=True,
            )
            raise
        except Exception as e:
            logger.error(
                "Failed to receive response",
                extra={"field_type": field_type, "port": self.port},
                exc_info=True,
            )
            raise ResponseError(f"Failed to receive response: {e}") from e

    def query(self, command: str) -> DeviceResponse:
        """
        Send command and receive response in one operation

        Convenience method that combines send_command() and receive_response().
        Automatically skips any data events to find the response.

        Args:
            command: Text command to send

        Returns:
            Parsed DeviceResponse object with type="response"

        Raises:
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If response timeout
        """
        # Send command
        self.send_command(command)

        # Receive response
        response = self.receive_response(field_type=FIELD_TYPE_RESPONSE)
        return response

    def close(self) -> None:
        """Close serial connection"""
        if self.device and self.device.is_open:
            self.device.close()

    def __enter__(self):
        """Context manager entry"""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit"""
        self.close()

`enter()` ¶

Context manager entry

Source code in src/kazunoko/device.py

def __enter__(self):
    """Context manager entry"""
    return self

`exit(exc_type, exc_val, exc_tb)` ¶

Context manager exit

Source code in src/kazunoko/device.py

def __exit__(self, exc_type, exc_val, exc_tb):
    """Context manager exit"""
    self.close()

`init(port=DEFAULT_PORT, baudrate=DEFAULT_BAUDRATE, timeout=DEFAULT_TIMEOUT)` ¶

Initialize device connection

Source code in src/kazunoko/device.py

def __init__(
    self,
    port: str = DEFAULT_PORT,
    baudrate: int = DEFAULT_BAUDRATE,
    timeout: float = DEFAULT_TIMEOUT,
):
    """Initialize device connection"""
    self.port = port
    self.baudrate = baudrate
    self.timeout = timeout
    self.device: serial.Serial | None = None

    try:
        logger.debug(
            "Opening serial port",
            extra={
                "port": port,
                "baudrate": baudrate,
                "timeout": timeout,
            }
        )
        self.device = serial.Serial(
            port=port,
            baudrate=baudrate,
            timeout=timeout,
        )
        logger.success("Serial port opened successfully", extra={"port": port})
        logger.info("Device connected", extra={"port": port})
    except serial.SerialException as e:
        logger.error(
            "Failed to open serial port",
            extra={"port": port, "baudrate": baudrate},
            exc_info=True,
        )
        raise DeviceError(f"Failed to open serial port {port}: {e}") from e

`close()` ¶

Close serial connection

Source code in src/kazunoko/device.py

def close(self) -> None:
    """Close serial connection"""
    if self.device and self.device.is_open:
        self.device.close()

`query(command)` ¶

Send command and receive response in one operation

Convenience method that combines send_command() and receive_response(). Automatically skips any data events to find the response.

Parameters:

Name	Type	Description	Default
`command`	`str`	Text command to send	required

Returns:

Type	Description
`DeviceResponse`	Parsed DeviceResponse object with type="response"

Raises:

Type	Description
`CommandError`	If command transmission fails
`ResponseError`	If response is invalid
`ResponseTimeout`	If response timeout

Source code in src/kazunoko/device.py

def query(self, command: str) -> DeviceResponse:
    """
    Send command and receive response in one operation

    Convenience method that combines send_command() and receive_response().
    Automatically skips any data events to find the response.

    Args:
        command: Text command to send

    Returns:
        Parsed DeviceResponse object with type="response"

    Raises:
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If response timeout
    """
    # Send command
    self.send_command(command)

    # Receive response
    response = self.receive_response(field_type=FIELD_TYPE_RESPONSE)
    return response

`receive_response(field_type)` ¶

Receive JSONL response from device and parse

Parameters:

Name	Type	Description	Default
`field_type`	`str`	Expected response type ("response" or "event")	required

Returns:

Type	Description
`DeviceResponse`	Parsed DeviceResponse object with received_us timestamp (in microseconds)

Raises:

Type	Description
`ResponseTimeout`	If no response received within timeout
`ResponseError`	If response is invalid or malformed

Source code in src/kazunoko/device.py

def receive_response(
    self,
    field_type: str,
) -> DeviceResponse:
    """
    Receive JSONL response from device and parse

    Args:
        field_type: Expected response type ("response" or "event")

    Returns:
        Parsed DeviceResponse object with received_us timestamp (in microseconds)

    Raises:
        ResponseTimeout: If no response received within timeout
        ResponseError: If response is invalid or malformed
    """
    if not self.device or not self.device.is_open:
        raise ResponseError("Device not connected")

    marker = f'"type":"{field_type}"'
    start_time = time.time()

    try:
        while time.time() - start_time < self.timeout:
            # Read one line from device
            line = self.device.readline().decode("utf-8", errors="ignore").strip()

            if not line:
                time.sleep(0.005)
                continue

            # Check if this is the expected type
            if marker not in line:
                logger.debug("Received non-matching line", extra={"field_type": field_type, "port": self.port})
                continue

            # Parse and validate
            received_at_us = int(time.time() * 1_000_000)
            parsed = parse_jsonl(line)
            if field_type is not None and parsed.type != field_type:
                raise ResponseError(f"Expected {field_type}, got {parsed.type}")

            # Set reception timestamp in microseconds
            parsed.received_us = received_at_us
            return parsed

        raise ResponseTimeout(f"No response received within {self.timeout}s timeout")

    except ResponseTimeout:
        raise
    except ResponseError:
        logger.error(
            "Invalid response received",
            extra={"field_type": field_type, "port": self.port},
            exc_info=True,
        )
        raise
    except Exception as e:
        logger.error(
            "Failed to receive response",
            extra={"field_type": field_type, "port": self.port},
            exc_info=True,
        )
        raise ResponseError(f"Failed to receive response: {e}") from e

`send_command(command)` ¶

Send text command to device with echo confirmation

Parameters:

Name	Type	Description	Default
`command`	`str`	Text command to send (e.g., "STATUS", "GET_VERSION")	required

Returns:

Type	Description
`bool`	True if command sent successfully

Raises:

Type	Description
`CommandError`	If device not connected or send fails

Source code in src/kazunoko/device.py

def send_command(self, command: str) -> bool:
    """
    Send text command to device with echo confirmation

    Args:
        command: Text command to send (e.g., "STATUS", "GET_VERSION")

    Returns:
        True if command sent successfully

    Raises:
        CommandError: If device not connected or send fails
    """
    if not self.device or not self.device.is_open:
        raise CommandError("Device not connected")

    # Append newline if not present
    if not command.endswith("\n"):
        command = command + "\n"

    try:
        logger.debug("Sending command", extra={"command": command.strip(), "port": self.port})
        # Send command
        self.device.reset_input_buffer()
        time.sleep(0.1)
        self.device.write(command.encode("utf-8"))
        self.device.flush()
        logger.success("Command sent successfully", extra={"command": command.strip(), "port": self.port})
        logger.info("Command executed", extra={"command": command.strip(), "port": self.port})
    except serial.SerialException as e:
        logger.error(
            "Failed to send command",
            extra={"command": command.strip(), "port": self.port},
            exc_info=True,
        )
        raise CommandError(f"Failed to send command: {e}") from e

    return True

Device¶

Architecture¶

Class Diagram¶

Device Selection Flow¶

connect¶

kazunoko.device.connect(port='auto', timeout=DEFAULT_TIMEOUT) ¶

detect_port¶

kazunoko.device.detect_port() ¶

DeviceProtocol¶

kazunoko.device.DeviceProtocol ¶

__enter__() ¶

__exit__(exc_type, exc_val, exc_tb) ¶

close() ¶

query(command) ¶

receive_response(field_type) ¶

send_command(command) ¶

PortDevice¶

kazunoko.device.PortDevice ¶

__enter__() ¶

__exit__(exc_type, exc_val, exc_tb) ¶

__init__(port=DEFAULT_PORT, baudrate=DEFAULT_BAUDRATE, timeout=DEFAULT_TIMEOUT) ¶

close() ¶

query(command) ¶

receive_response(field_type) ¶

send_command(command) ¶

`kazunoko.device.connect(port='auto', timeout=DEFAULT_TIMEOUT)` ¶

`kazunoko.device.detect_port()` ¶

`kazunoko.device.DeviceProtocol` ¶

`enter()` ¶

`exit(exc_type, exc_val, exc_tb)` ¶

`close()` ¶

`query(command)` ¶

`receive_response(field_type)` ¶

`send_command(command)` ¶

`kazunoko.device.PortDevice` ¶

`enter()` ¶

`exit(exc_type, exc_val, exc_tb)` ¶

`init(port=DEFAULT_PORT, baudrate=DEFAULT_BAUDRATE, timeout=DEFAULT_TIMEOUT)` ¶

`close()` ¶

`query(command)` ¶

`receive_response(field_type)` ¶

`send_command(command)` ¶