Command

Convenient wrapper around device communication with easy-to-use methods for common queries. The Command class provides a cleaner API than sending raw command strings, with methods like status(), threshold(), read(), and more.

---

kazunoko.command.Command

Convenience wrapper for device queries

Provides alias methods for common OSECHI detector commands. Wraps a DeviceProtocol instance to enable clean, discoverable API.

Parameters:

Name	Type	Description	Default
device	DeviceProtocol	A DeviceProtocol instance (PortDevice or mock)	required

Example

from kazunoko import connect, Command

with connect() as device:
    cmd = Command(device)
    resp = cmd.status()
    print(resp.status)      # "ok"
    print(resp.version)     # "1.10.1"

Source code in src/kazunoko/command.py

class Command:
    """
    Convenience wrapper for device queries

    Provides alias methods for common OSECHI detector commands.
    Wraps a DeviceProtocol instance to enable clean, discoverable API.

    Args:
        device: A DeviceProtocol instance (PortDevice or mock)

    Example:
        ```python
        from kazunoko import connect, Command

        with connect() as device:
            cmd = Command(device)
            resp = cmd.status()
            print(resp.status)      # "ok"
            print(resp.version)     # "1.10.1"
        ```
    """

    def __init__(self, device: DeviceProtocol) -> None:
        """
        Initialize Command with device instance

        Args:
            device: A DeviceProtocol instance for sending commands

        Example:
            ```python
            device = PortDevice()
            cmd = Command(device)
            ```
        """
        self._device = device

    def query(self, command: str) -> DeviceResponse:
        """
        Send a raw command to the device

        Low-level method for sending arbitrary commands directly to the device.
        Use this for custom commands or device-specific queries not covered by
        other convenience methods.

        Args:
            command: Command string to send to the device (e.g., "GET_STATUS", "SET_THRESHOLD 1 300")

        Returns:
            DeviceResponse object with type="response" and command result

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)

            # Use built-in convenience method
            resp = cmd.status()

            # Use raw query for custom commands
            resp = cmd.query("GET_STATUS")
            resp = cmd.query("SET_THRESHOLD 1 300")
            resp = cmd.query("CUSTOM_CMD arg1 arg2")
            ```
        """
        logger.debug("Executing device query", extra={"command": command})
        return self._device.query(command)

    def help(self) -> DeviceResponse:
        """
        Get available commands from device

        Queries the device for the list of available commands and returns
        detailed command information.

        Returns:
            DeviceResponse object with command information

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)
            resp = cmd.help()
            print(resp.commands)  # List of available commands

            # Use ResponseFormatter for flexible output formatting
            fmt = ResponseFormatter(resp)
            print(fmt.to_jsonl())  # Flattened JSONL
            ```
        """
        return self._device.query("GET_HELP")

    def usage(self) -> DeviceResponse:
        """
        Get list of queryable commands from device

        Queries the device for the list of all available commands that can be queried.
        Returns detailed command information including arguments and descriptions.

        Returns:
            DeviceResponse object with command list and details

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)
            resp = cmd.usage()
            print(resp.commands)  # List of all available commands

            # Use ResponseFormatter for flexible output formatting
            fmt = ResponseFormatter(resp)
            print(fmt.to_table())  # Formatted table
            ```
        """
        return self._device.query("GET_USAGE")

    def status(self) -> DeviceResponse:
        """
        Get device status and metadata

        Queries the device for current status, version, and other metadata.
        The Response object contains dynamic fields that vary by device firmware.

        Returns:
            DeviceResponse object with type="response" and dynamic fields:
            - status: "ok" or "error"
            - version: Device firmware version string
            - poll_count: Number of polled events
            - (other device-specific fields)

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)
            resp = cmd.status()
            print(resp.status)      # "ok"
            print(resp.version)     # "1.10.1"
            print(resp.poll_count)  # 100
            ```
        """
        return self._device.query("GET_STATUS")

    def version(self) -> DeviceResponse:
        """
        Get device firmware version

        Queries the device for firmware version information.

        Returns:
            DeviceResponse object with type="response" containing version information

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)
            resp = cmd.version()
            print(resp.version)  # "1.10.1"
            ```
        """
        return self._device.query("GET_VERSION")

    def poll_count(self, count: int) -> DeviceResponse:
        """
        Set poll count for detection sampling

        Configures the number of events to poll from the detector.

        Args:
            count: Poll count value (device-specific range, typically 1-1000)

        Returns:
            DeviceResponse object with type="response" and command status

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)
            resp = cmd.poll_count(100)
            print(resp.status)  # "ok"
            ```
        """
        return self._device.query(f"SET_POLL_COUNT {count}")

    def deadtime(self, milliseconds: int) -> DeviceResponse:
        """
        Set detection deadtime

        Configures the deadtime (dead period) during which the detector is insensitive
        to new detection events after processing a detection.

        Args:
            milliseconds: Deadtime value in milliseconds (device-specific range, typically 0-60000)

        Returns:
            DeviceResponse object with type="response" and command status

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)
            resp = cmd.deadtime(100)
            print(resp.status)  # "ok"
            ```
        """
        return self._device.query(f"SET_DEADTIME {milliseconds}")

    def uptime(self) -> DeviceResponse:
        """
        Get device uptime since power-on

        Queries the device for the time elapsed since it was powered on.

        Returns:
            DeviceResponse object with type="response" containing uptime information

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)
            resp = cmd.uptime()
            print(resp.uptime)  # Uptime value
            ```
        """
        return self._device.query("GET_UPTIME")

    def reset(self) -> DeviceResponse:
        """
        Reset device configuration to defaults

        Resets all device settings to their default values.

        Returns:
            DeviceResponse object with type="response" and command status

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)
            resp = cmd.reset()
            print(resp.status)  # "ok"
            ```
        """
        return self._device.query("RESET")

    def mac_address(self) -> DeviceResponse:
        """
        Get device MAC address

        Queries the device for its MAC address.

        Returns:
            DeviceResponse object with type="response" containing MAC address information

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)
            resp = cmd.mac_address()
            print(resp.mac_address)  # Device MAC address
            ```
        """
        return self._device.query("GET_MAC_ADDRESS")

    def queue_stats(self) -> DeviceResponse:
        """
        Get queue statistics from device

        Queries the device for queue statistics information.

        Returns:
            DeviceResponse object with type="response" containing queue statistics

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)
            resp = cmd.queue_stats()
            print(resp.queue_count)  # Queue statistics
            ```
        """
        return self._device.query("GET_QUEUE_STATS")

    def led(self, channel: int | Literal["all"], state: Literal["on", "off"]) -> DeviceResponse:
        """
        Control LED on specific channel or all channels

        Controls the LED state (on/off) for a specific detector channel or all channels.

        Args:
            channel: Channel number (1-3) or "all" for all channels
            state: LED state, either "on" or "off"

        Returns:
            DeviceResponse object with type="response" and command status

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)

            # Turn on LED for channel 1
            resp = cmd.led(1, "on")
            print(resp.status)  # "ok"

            # Turn off LED for all channels
            resp = cmd.led("all", "off")
            print(resp.status)  # "ok"
            ```
        """
        return self._device.query(f"TEST_LED {channel} {state}")

    def threshold(self, channel: int, value: int | None = None) -> DeviceResponse:
        """
        Get or set threshold for a single channel

        If value is None, retrieves the current threshold for the channel (GET_THRESHOLD).
        If value is provided, sets the threshold for the channel (SET_THRESHOLD).

        Args:
            channel: Channel number (device-specific range)
            value: Threshold value (device-specific range), or None to get current threshold

        Returns:
            DeviceResponse object with type="response" and command status.
            For GET_THRESHOLD: contains current threshold value
            For SET_THRESHOLD: contains confirmation status

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)

            # Get current threshold for channel 1
            resp = cmd.threshold(1)
            print(resp.value)  # Current threshold value

            # Set threshold for channel 1
            resp = cmd.threshold(1, 300)
            print(resp.status)  # "ok"
            ```
        """
        if value is None:
            return self._device.query(f"GET_THRESHOLD {channel}")
        else:
            return self._device.query(f"SET_THRESHOLD {channel} {value}")

    def thresholds(self, thresholds: dict[int, int] | None = None) -> list[DeviceResponse]:
        """
        Get or set thresholds for multiple channels

        If thresholds is None, retrieves thresholds for all channels (1-3).
        If thresholds is provided, sets thresholds for each channel in the dictionary.

        Args:
            thresholds: Dictionary mapping channel number to threshold value, or None to get all thresholds

        Returns:
            List of DeviceResponse objects, one per channel

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)

            # Get thresholds for all channels
            responses = cmd.thresholds()
            for resp in responses:
                print(resp.value)  # Threshold value for each channel

            # Set thresholds for specific channels
            responses = cmd.thresholds({1: 300, 2: 400, 3: 500})
            for resp in responses:
                print(resp.status)  # "ok" for each
            ```
        """
        if thresholds is None:
            # Get thresholds for all channels (1-3)
            logger.debug("Getting thresholds for all channels")
            responses = [self.threshold(ch) for ch in range(1, 4)]
            logger.debug("Retrieved all channel thresholds", extra={"channel_count": len(responses)})
            return responses
        else:
            # Set thresholds for specified channels
            logger.debug(
                "Setting thresholds for multiple channels",
                extra={"channel_count": len(thresholds), "channels": sorted(thresholds.keys())}
            )
            responses = [self.threshold(ch, val) for ch, val in thresholds.items()]
            logger.debug("Set thresholds for all channels", extra={"channel_count": len(responses)})
            return responses

    def rtc_time(self, value: int | None = None) -> DeviceResponse:
        """
        Get or set device RTC (Real-Time Clock) time

        If value is None, retrieves the current device time (GET_RTC_TIME).
        If value is an integer, sets the device time (SET_RTC_TIME).

        Args:
            value: Unix timestamp to set, or None to get current time

        Returns:
            DeviceResponse object with type="response" and command status.
            For GET_RTC_TIME: contains current time fields
            For SET_RTC_TIME: contains confirmation status

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            import time
            cmd = Command(device)

            # Get current device time
            resp = cmd.rtc_time()
            print(resp.timestamp)  # Current device time

            # Set device time to current PC time
            unixtime = int(time.time())
            resp = cmd.rtc_time(unixtime)
            print(resp.status)  # "ok"
            ```
        """
        if value is None:
            return self._device.query("GET_RTC_TIME")
        else:
            return self._device.query(f"SET_RTC_TIME {value}")

    def read(self) -> DeviceResponse:
        """
        Read a detection event from device

        Queries the device for a single detection event. Returns DeviceResponse object
        with type="event" containing the event data.

        Returns:
            DeviceResponse object with type="event" containing detection event fields.

        Raises:
            DeviceError: If device connection fails
            CommandError: If command transmission fails
            ResponseError: If response is invalid
            ResponseTimeout: If no response received within timeout

        Example:
            ```python
            cmd = Command(device)
            event = cmd.read()
            print(event.type)       # "event"
            print(event.timestamp)  # Event timestamp
            print(event.channel)    # Detected channel

            # Use ResponseFormatter for flexible output formatting
            fmt = ResponseFormatter(event)
            print(fmt.to_csv())  # Flattened CSV
            ```
        """
        return self._device.receive_response(field_type=FIELD_TYPE_EVENT)

__init__(device)

Initialize Command with device instance

Parameters:

Name	Type	Description	Default
device	DeviceProtocol	A DeviceProtocol instance for sending commands	required

Example

device = PortDevice()
cmd = Command(device)

Source code in src/kazunoko/command.py

def __init__(self, device: DeviceProtocol) -> None:
    """
    Initialize Command with device instance

    Args:
        device: A DeviceProtocol instance for sending commands

    Example:
        ```python
        device = PortDevice()
        cmd = Command(device)
        ```
    """
    self._device = device

deadtime(milliseconds)

Set detection deadtime

Configures the deadtime (dead period) during which the detector is insensitive to new detection events after processing a detection.

Parameters:

Name	Type	Description	Default
milliseconds	int	Deadtime value in milliseconds (device-specific range, typically 0-60000)	required

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="response" and command status

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)
resp = cmd.deadtime(100)
print(resp.status)  # "ok"

Source code in src/kazunoko/command.py

def deadtime(self, milliseconds: int) -> DeviceResponse:
    """
    Set detection deadtime

    Configures the deadtime (dead period) during which the detector is insensitive
    to new detection events after processing a detection.

    Args:
        milliseconds: Deadtime value in milliseconds (device-specific range, typically 0-60000)

    Returns:
        DeviceResponse object with type="response" and command status

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)
        resp = cmd.deadtime(100)
        print(resp.status)  # "ok"
        ```
    """
    return self._device.query(f"SET_DEADTIME {milliseconds}")

help()

Get available commands from device

Queries the device for the list of available commands and returns detailed command information.

Returns:

Type	Description
DeviceResponse	DeviceResponse object with command information

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)
resp = cmd.help()
print(resp.commands)  # List of available commands

# Use ResponseFormatter for flexible output formatting
fmt = ResponseFormatter(resp)
print(fmt.to_jsonl())  # Flattened JSONL

Source code in src/kazunoko/command.py

def help(self) -> DeviceResponse:
    """
    Get available commands from device

    Queries the device for the list of available commands and returns
    detailed command information.

    Returns:
        DeviceResponse object with command information

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)
        resp = cmd.help()
        print(resp.commands)  # List of available commands

        # Use ResponseFormatter for flexible output formatting
        fmt = ResponseFormatter(resp)
        print(fmt.to_jsonl())  # Flattened JSONL
        ```
    """
    return self._device.query("GET_HELP")

led(channel, state)

Control LED on specific channel or all channels

Controls the LED state (on/off) for a specific detector channel or all channels.

Parameters:

Name	Type	Description	Default
channel	int | Literal['all']	Channel number (1-3) or "all" for all channels	required
state	Literal['on', 'off']	LED state, either "on" or "off"	required

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="response" and command status

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)

# Turn on LED for channel 1
resp = cmd.led(1, "on")
print(resp.status)  # "ok"

# Turn off LED for all channels
resp = cmd.led("all", "off")
print(resp.status)  # "ok"

Source code in src/kazunoko/command.py

def led(self, channel: int | Literal["all"], state: Literal["on", "off"]) -> DeviceResponse:
    """
    Control LED on specific channel or all channels

    Controls the LED state (on/off) for a specific detector channel or all channels.

    Args:
        channel: Channel number (1-3) or "all" for all channels
        state: LED state, either "on" or "off"

    Returns:
        DeviceResponse object with type="response" and command status

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)

        # Turn on LED for channel 1
        resp = cmd.led(1, "on")
        print(resp.status)  # "ok"

        # Turn off LED for all channels
        resp = cmd.led("all", "off")
        print(resp.status)  # "ok"
        ```
    """
    return self._device.query(f"TEST_LED {channel} {state}")

mac_address()

Get device MAC address

Queries the device for its MAC address.

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="response" containing MAC address information

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)
resp = cmd.mac_address()
print(resp.mac_address)  # Device MAC address

Source code in src/kazunoko/command.py

def mac_address(self) -> DeviceResponse:
    """
    Get device MAC address

    Queries the device for its MAC address.

    Returns:
        DeviceResponse object with type="response" containing MAC address information

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)
        resp = cmd.mac_address()
        print(resp.mac_address)  # Device MAC address
        ```
    """
    return self._device.query("GET_MAC_ADDRESS")

poll_count(count)

Set poll count for detection sampling

Configures the number of events to poll from the detector.

Parameters:

Name	Type	Description	Default
count	int	Poll count value (device-specific range, typically 1-1000)	required

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="response" and command status

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)
resp = cmd.poll_count(100)
print(resp.status)  # "ok"

Source code in src/kazunoko/command.py

def poll_count(self, count: int) -> DeviceResponse:
    """
    Set poll count for detection sampling

    Configures the number of events to poll from the detector.

    Args:
        count: Poll count value (device-specific range, typically 1-1000)

    Returns:
        DeviceResponse object with type="response" and command status

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)
        resp = cmd.poll_count(100)
        print(resp.status)  # "ok"
        ```
    """
    return self._device.query(f"SET_POLL_COUNT {count}")

query(command)

Send a raw command to the device

Low-level method for sending arbitrary commands directly to the device. Use this for custom commands or device-specific queries not covered by other convenience methods.

Parameters:

Name	Type	Description	Default
command	str	Command string to send to the device (e.g., "GET_STATUS", "SET_THRESHOLD 1 300")	required

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="response" and command result

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)

# Use built-in convenience method
resp = cmd.status()

# Use raw query for custom commands
resp = cmd.query("GET_STATUS")
resp = cmd.query("SET_THRESHOLD 1 300")
resp = cmd.query("CUSTOM_CMD arg1 arg2")

Source code in src/kazunoko/command.py

def query(self, command: str) -> DeviceResponse:
    """
    Send a raw command to the device

    Low-level method for sending arbitrary commands directly to the device.
    Use this for custom commands or device-specific queries not covered by
    other convenience methods.

    Args:
        command: Command string to send to the device (e.g., "GET_STATUS", "SET_THRESHOLD 1 300")

    Returns:
        DeviceResponse object with type="response" and command result

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)

        # Use built-in convenience method
        resp = cmd.status()

        # Use raw query for custom commands
        resp = cmd.query("GET_STATUS")
        resp = cmd.query("SET_THRESHOLD 1 300")
        resp = cmd.query("CUSTOM_CMD arg1 arg2")
        ```
    """
    logger.debug("Executing device query", extra={"command": command})
    return self._device.query(command)

queue_stats()

Get queue statistics from device

Queries the device for queue statistics information.

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="response" containing queue statistics

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)
resp = cmd.queue_stats()
print(resp.queue_count)  # Queue statistics

Source code in src/kazunoko/command.py

def queue_stats(self) -> DeviceResponse:
    """
    Get queue statistics from device

    Queries the device for queue statistics information.

    Returns:
        DeviceResponse object with type="response" containing queue statistics

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)
        resp = cmd.queue_stats()
        print(resp.queue_count)  # Queue statistics
        ```
    """
    return self._device.query("GET_QUEUE_STATS")

read()

Read a detection event from device

Queries the device for a single detection event. Returns DeviceResponse object with type="event" containing the event data.

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="event" containing detection event fields.

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)
event = cmd.read()
print(event.type)       # "event"
print(event.timestamp)  # Event timestamp
print(event.channel)    # Detected channel

# Use ResponseFormatter for flexible output formatting
fmt = ResponseFormatter(event)
print(fmt.to_csv())  # Flattened CSV

Source code in src/kazunoko/command.py

def read(self) -> DeviceResponse:
    """
    Read a detection event from device

    Queries the device for a single detection event. Returns DeviceResponse object
    with type="event" containing the event data.

    Returns:
        DeviceResponse object with type="event" containing detection event fields.

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)
        event = cmd.read()
        print(event.type)       # "event"
        print(event.timestamp)  # Event timestamp
        print(event.channel)    # Detected channel

        # Use ResponseFormatter for flexible output formatting
        fmt = ResponseFormatter(event)
        print(fmt.to_csv())  # Flattened CSV
        ```
    """
    return self._device.receive_response(field_type=FIELD_TYPE_EVENT)

reset()

Reset device configuration to defaults

Resets all device settings to their default values.

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="response" and command status

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)
resp = cmd.reset()
print(resp.status)  # "ok"

Source code in src/kazunoko/command.py

def reset(self) -> DeviceResponse:
    """
    Reset device configuration to defaults

    Resets all device settings to their default values.

    Returns:
        DeviceResponse object with type="response" and command status

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)
        resp = cmd.reset()
        print(resp.status)  # "ok"
        ```
    """
    return self._device.query("RESET")

rtc_time(value=None)

Get or set device RTC (Real-Time Clock) time

If value is None, retrieves the current device time (GET_RTC_TIME). If value is an integer, sets the device time (SET_RTC_TIME).

Parameters:

Name	Type	Description	Default
value	int | None	Unix timestamp to set, or None to get current time	None

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="response" and command status.
DeviceResponse	For GET_RTC_TIME: contains current time fields
DeviceResponse	For SET_RTC_TIME: contains confirmation status

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

import time
cmd = Command(device)

# Get current device time
resp = cmd.rtc_time()
print(resp.timestamp)  # Current device time

# Set device time to current PC time
unixtime = int(time.time())
resp = cmd.rtc_time(unixtime)
print(resp.status)  # "ok"

Source code in src/kazunoko/command.py

def rtc_time(self, value: int | None = None) -> DeviceResponse:
    """
    Get or set device RTC (Real-Time Clock) time

    If value is None, retrieves the current device time (GET_RTC_TIME).
    If value is an integer, sets the device time (SET_RTC_TIME).

    Args:
        value: Unix timestamp to set, or None to get current time

    Returns:
        DeviceResponse object with type="response" and command status.
        For GET_RTC_TIME: contains current time fields
        For SET_RTC_TIME: contains confirmation status

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        import time
        cmd = Command(device)

        # Get current device time
        resp = cmd.rtc_time()
        print(resp.timestamp)  # Current device time

        # Set device time to current PC time
        unixtime = int(time.time())
        resp = cmd.rtc_time(unixtime)
        print(resp.status)  # "ok"
        ```
    """
    if value is None:
        return self._device.query("GET_RTC_TIME")
    else:
        return self._device.query(f"SET_RTC_TIME {value}")

status()

Get device status and metadata

Queries the device for current status, version, and other metadata. The Response object contains dynamic fields that vary by device firmware.

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="response" and dynamic fields:
DeviceResponse	status: "ok" or "error"
DeviceResponse	version: Device firmware version string
DeviceResponse	poll_count: Number of polled events
DeviceResponse	(other device-specific fields)

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)
resp = cmd.status()
print(resp.status)      # "ok"
print(resp.version)     # "1.10.1"
print(resp.poll_count)  # 100

Source code in src/kazunoko/command.py

def status(self) -> DeviceResponse:
    """
    Get device status and metadata

    Queries the device for current status, version, and other metadata.
    The Response object contains dynamic fields that vary by device firmware.

    Returns:
        DeviceResponse object with type="response" and dynamic fields:
        - status: "ok" or "error"
        - version: Device firmware version string
        - poll_count: Number of polled events
        - (other device-specific fields)

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)
        resp = cmd.status()
        print(resp.status)      # "ok"
        print(resp.version)     # "1.10.1"
        print(resp.poll_count)  # 100
        ```
    """
    return self._device.query("GET_STATUS")

threshold(channel, value=None)

Get or set threshold for a single channel

If value is None, retrieves the current threshold for the channel (GET_THRESHOLD). If value is provided, sets the threshold for the channel (SET_THRESHOLD).

Parameters:

Name	Type	Description	Default
channel	int	Channel number (device-specific range)	required
value	int | None	Threshold value (device-specific range), or None to get current threshold	None

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="response" and command status.
DeviceResponse	For GET_THRESHOLD: contains current threshold value
DeviceResponse	For SET_THRESHOLD: contains confirmation status

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)

# Get current threshold for channel 1
resp = cmd.threshold(1)
print(resp.value)  # Current threshold value

# Set threshold for channel 1
resp = cmd.threshold(1, 300)
print(resp.status)  # "ok"

Source code in src/kazunoko/command.py

def threshold(self, channel: int, value: int | None = None) -> DeviceResponse:
    """
    Get or set threshold for a single channel

    If value is None, retrieves the current threshold for the channel (GET_THRESHOLD).
    If value is provided, sets the threshold for the channel (SET_THRESHOLD).

    Args:
        channel: Channel number (device-specific range)
        value: Threshold value (device-specific range), or None to get current threshold

    Returns:
        DeviceResponse object with type="response" and command status.
        For GET_THRESHOLD: contains current threshold value
        For SET_THRESHOLD: contains confirmation status

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)

        # Get current threshold for channel 1
        resp = cmd.threshold(1)
        print(resp.value)  # Current threshold value

        # Set threshold for channel 1
        resp = cmd.threshold(1, 300)
        print(resp.status)  # "ok"
        ```
    """
    if value is None:
        return self._device.query(f"GET_THRESHOLD {channel}")
    else:
        return self._device.query(f"SET_THRESHOLD {channel} {value}")

thresholds(thresholds=None)

Get or set thresholds for multiple channels

If thresholds is None, retrieves thresholds for all channels (1-3). If thresholds is provided, sets thresholds for each channel in the dictionary.

Parameters:

Name	Type	Description	Default
thresholds	dict[int, int] | None	Dictionary mapping channel number to threshold value, or None to get all thresholds	None

Returns:

Type	Description
list[DeviceResponse]	List of DeviceResponse objects, one per channel

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)

# Get thresholds for all channels
responses = cmd.thresholds()
for resp in responses:
    print(resp.value)  # Threshold value for each channel

# Set thresholds for specific channels
responses = cmd.thresholds({1: 300, 2: 400, 3: 500})
for resp in responses:
    print(resp.status)  # "ok" for each

Source code in src/kazunoko/command.py

def thresholds(self, thresholds: dict[int, int] | None = None) -> list[DeviceResponse]:
    """
    Get or set thresholds for multiple channels

    If thresholds is None, retrieves thresholds for all channels (1-3).
    If thresholds is provided, sets thresholds for each channel in the dictionary.

    Args:
        thresholds: Dictionary mapping channel number to threshold value, or None to get all thresholds

    Returns:
        List of DeviceResponse objects, one per channel

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)

        # Get thresholds for all channels
        responses = cmd.thresholds()
        for resp in responses:
            print(resp.value)  # Threshold value for each channel

        # Set thresholds for specific channels
        responses = cmd.thresholds({1: 300, 2: 400, 3: 500})
        for resp in responses:
            print(resp.status)  # "ok" for each
        ```
    """
    if thresholds is None:
        # Get thresholds for all channels (1-3)
        logger.debug("Getting thresholds for all channels")
        responses = [self.threshold(ch) for ch in range(1, 4)]
        logger.debug("Retrieved all channel thresholds", extra={"channel_count": len(responses)})
        return responses
    else:
        # Set thresholds for specified channels
        logger.debug(
            "Setting thresholds for multiple channels",
            extra={"channel_count": len(thresholds), "channels": sorted(thresholds.keys())}
        )
        responses = [self.threshold(ch, val) for ch, val in thresholds.items()]
        logger.debug("Set thresholds for all channels", extra={"channel_count": len(responses)})
        return responses

uptime()

Get device uptime since power-on

Queries the device for the time elapsed since it was powered on.

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="response" containing uptime information

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)
resp = cmd.uptime()
print(resp.uptime)  # Uptime value

Source code in src/kazunoko/command.py

def uptime(self) -> DeviceResponse:
    """
    Get device uptime since power-on

    Queries the device for the time elapsed since it was powered on.

    Returns:
        DeviceResponse object with type="response" containing uptime information

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)
        resp = cmd.uptime()
        print(resp.uptime)  # Uptime value
        ```
    """
    return self._device.query("GET_UPTIME")

usage()

Get list of queryable commands from device

Queries the device for the list of all available commands that can be queried. Returns detailed command information including arguments and descriptions.

Returns:

Type	Description
DeviceResponse	DeviceResponse object with command list and details

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)
resp = cmd.usage()
print(resp.commands)  # List of all available commands

# Use ResponseFormatter for flexible output formatting
fmt = ResponseFormatter(resp)
print(fmt.to_table())  # Formatted table

Source code in src/kazunoko/command.py

def usage(self) -> DeviceResponse:
    """
    Get list of queryable commands from device

    Queries the device for the list of all available commands that can be queried.
    Returns detailed command information including arguments and descriptions.

    Returns:
        DeviceResponse object with command list and details

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)
        resp = cmd.usage()
        print(resp.commands)  # List of all available commands

        # Use ResponseFormatter for flexible output formatting
        fmt = ResponseFormatter(resp)
        print(fmt.to_table())  # Formatted table
        ```
    """
    return self._device.query("GET_USAGE")

version()

Get device firmware version

Queries the device for firmware version information.

Returns:

Type	Description
DeviceResponse	DeviceResponse object with type="response" containing version information

Raises:

Type	Description
DeviceError	If device connection fails
CommandError	If command transmission fails
ResponseError	If response is invalid
ResponseTimeout	If no response received within timeout

Example

cmd = Command(device)
resp = cmd.version()
print(resp.version)  # "1.10.1"

Source code in src/kazunoko/command.py

def version(self) -> DeviceResponse:
    """
    Get device firmware version

    Queries the device for firmware version information.

    Returns:
        DeviceResponse object with type="response" containing version information

    Raises:
        DeviceError: If device connection fails
        CommandError: If command transmission fails
        ResponseError: If response is invalid
        ResponseTimeout: If no response received within timeout

    Example:
        ```python
        cmd = Command(device)
        resp = cmd.version()
        print(resp.version)  # "1.10.1"
        ```
    """
    return self._device.query("GET_VERSION")