Skip to content

Parser

Parse JSONL (JSON Lines) responses from the device into Python objects. The DeviceResponse model provides flexible field support with Pydantic validation, while ResponseParser handles parsing and validation logic.

kazunoko.parser

JSONL parsing and validation for kazunoko responses

DeviceResponse

Bases: BaseModel

Device response model for JSONL parsing

Flexible model that accepts any field from device. Used for parsing responses where field structure varies by command.

Example

  • Command response 

    {"type":"response","status":"ok","version":"1.10.1",...}

  • Detection event 

    {"type":"event","signal1":95,"signal2":87,...}

Source code in src/kazunoko/parser.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class DeviceResponse(BaseModel):
    """
    Device response model for JSONL parsing

    Flexible model that accepts any field from device.
    Used for parsing responses where field structure varies by command.

    Example:
        - Command response
        ```json
        {"type":"response","status":"ok","version":"1.10.1",...}
        ```

        - Detection event
        ```json
        {"type":"event","signal1":95,"signal2":87,...}
        ```
    """

    type: str  # "response" or "event"
    status: str | None = None  # "ok" or "error" (optional for event type)
    received_us: int | None = None  # Reception timestamp in microseconds since epoch

    model_config = ConfigDict(extra="allow")  # Allow additional fields

ResponseParser

Parse and validate JSONL responses from device

Source code in src/kazunoko/parser.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class ResponseParser:
    """
    Parse and validate JSONL responses from device
    """

    @staticmethod
    def parse_line(line: str) -> dict[str, Any]:
        """
        Parse single JSONL line to dictionary

        Args:
            line: JSONL line (single JSON object as string)

        Returns:
            Parsed dictionary

        Raises:
            ProtocolError: If JSON is invalid or malformed
        """
        line = line.strip()

        if not line:
            logger.debug("Empty line received during parsing")
            raise ProtocolError("Empty line received")

        try:
            data = json.loads(line)
        except json.JSONDecodeError as e:
            logger.warning(
                "JSON parsing failed",
                extra={
                    "error": str(e),
                    "line_preview": line[:100],
                    "line_length": len(line),
                }
            )
            raise ProtocolError(f"Invalid JSON: {e}") from e

        if not isinstance(data, dict):
            logger.warning(
                "Unexpected JSON structure",
                extra={
                    "expected_type": "dict",
                    "received_type": type(data).__name__,
                }
            )
            raise ProtocolError(f"Expected JSON object, got {type(data).__name__}")

        return data

    @staticmethod
    def validate_response(data: dict[str, Any]) -> DeviceResponse:
        """
        Validate parsed response against Pydantic model

        Args:
            data: Parsed JSONL dictionary

        Returns:
            Validated DeviceResponse object

        Raises:
            ResponseError: If validation fails
        """
        try:
            response = DeviceResponse(**data)
            return response
        except ValidationError as e:
            logger.warning(
                "Response validation failed",
                extra={
                    "response_type": data.get("type"),
                    "error_count": len(e.errors()),
                    "first_error": str(e.errors()[0]) if e.errors() else None,
                }
            )
            raise ResponseError(f"Response validation failed: {e}") from e

    @staticmethod
    def parse_jsonl(line: str) -> DeviceResponse:
        """
        Parse and validate JSONL line in one call

        Args:
            line: JSONL line from device

        Returns:
            Validated DeviceResponse object

        Raises:
            ProtocolError: If JSON parsing fails
            ResponseError: If validation fails
        """
        data = ResponseParser.parse_line(line)
        return ResponseParser.validate_response(data)

parse_jsonl(line) staticmethod

Parse and validate JSONL line in one call

Parameters:

Name Type Description Default
line str

JSONL line from device

 required

Returns:

Type Description
DeviceResponse

Validated DeviceResponse object

Raises:

Type Description
ProtocolError

If JSON parsing fails
ResponseError

If validation fails
Source code in src/kazunoko/parser.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@staticmethod
def parse_jsonl(line: str) -> DeviceResponse:
    """
    Parse and validate JSONL line in one call

    Args:
        line: JSONL line from device

    Returns:
        Validated DeviceResponse object

    Raises:
        ProtocolError: If JSON parsing fails
        ResponseError: If validation fails
    """
    data = ResponseParser.parse_line(line)
    return ResponseParser.validate_response(data)

parse_line(line) staticmethod

Parse single JSONL line to dictionary

Parameters:

Name Type Description Default
line str

JSONL line (single JSON object as string)

 required

Returns:

Type Description
dict[str, Any]

Parsed dictionary

Raises:

Type Description
ProtocolError

If JSON is invalid or malformed
Source code in src/kazunoko/parser.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@staticmethod
def parse_line(line: str) -> dict[str, Any]:
    """
    Parse single JSONL line to dictionary

    Args:
        line: JSONL line (single JSON object as string)

    Returns:
        Parsed dictionary

    Raises:
        ProtocolError: If JSON is invalid or malformed
    """
    line = line.strip()

    if not line:
        logger.debug("Empty line received during parsing")
        raise ProtocolError("Empty line received")

    try:
        data = json.loads(line)
    except json.JSONDecodeError as e:
        logger.warning(
            "JSON parsing failed",
            extra={
                "error": str(e),
                "line_preview": line[:100],
                "line_length": len(line),
            }
        )
        raise ProtocolError(f"Invalid JSON: {e}") from e

    if not isinstance(data, dict):
        logger.warning(
            "Unexpected JSON structure",
            extra={
                "expected_type": "dict",
                "received_type": type(data).__name__,
            }
        )
        raise ProtocolError(f"Expected JSON object, got {type(data).__name__}")

    return data

validate_response(data) staticmethod

Validate parsed response against Pydantic model

Parameters:

Name Type Description Default
data dict[str, Any]

Parsed JSONL dictionary

 required

Returns:

Type Description
DeviceResponse

Validated DeviceResponse object

Raises:

Type Description
ResponseError

If validation fails
Source code in src/kazunoko/parser.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@staticmethod
def validate_response(data: dict[str, Any]) -> DeviceResponse:
    """
    Validate parsed response against Pydantic model

    Args:
        data: Parsed JSONL dictionary

    Returns:
        Validated DeviceResponse object

    Raises:
        ResponseError: If validation fails
    """
    try:
        response = DeviceResponse(**data)
        return response
    except ValidationError as e:
        logger.warning(
            "Response validation failed",
            extra={
                "response_type": data.get("type"),
                "error_count": len(e.errors()),
                "first_error": str(e.errors()[0]) if e.errors() else None,
            }
        )
        raise ResponseError(f"Response validation failed: {e}") from e

parse_jsonl(line)

Convenience function: parse JSONL line

Parameters:

Name Type Description Default
line str

JSONL line from device

 required

Returns:

Type Description
DeviceResponse

Validated DeviceResponse object

Raises:

Type Description
ProtocolError

If JSON parsing fails
ResponseError

If validation fails
Source code in src/kazunoko/parser.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def parse_jsonl(line: str) -> DeviceResponse:
    """
    Convenience function: parse JSONL line

    Args:
        line: JSONL line from device

    Returns:
        Validated DeviceResponse object

    Raises:
        ProtocolError: If JSON parsing fails
        ResponseError: If validation fails
    """
    return ResponseParser.parse_jsonl(line)

parse_thresholds(value)

Parse threshold string to dictionary format

Converts a semicolon-separated threshold specification into a dictionary mapping channel numbers to threshold values. This is useful for setting device thresholds via the Command API or in standalone scripts.

Parameters:

Name Type Description Default
value str

Threshold string (e.g., "1:300;2:300;3:300")

 required

Returns:

Type Description
dict[int, int]

Dictionary mapping channel numbers (int) to threshold values (int)

Raises:

Type Description
ValueError

If format is invalid
Example 
from kazunoko import parse_thresholds, Command, connect

# Parse threshold string
thresholds = parse_thresholds("1:300;2:300;3:300")
# Returns: {1: 300, 2: 300, 3: 300}

# Use with Command API
with connect() as device:
    cmd = Command(device)
    cmd.thresholds(thresholds)
Source code in src/kazunoko/parser.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def parse_thresholds(value: str) -> dict[int, int]:
    """
    Parse threshold string to dictionary format

    Converts a semicolon-separated threshold specification into a dictionary
    mapping channel numbers to threshold values. This is useful for setting
    device thresholds via the Command API or in standalone scripts.

    Args:
        value: Threshold string (e.g., "1:300;2:300;3:300")

    Returns:
        Dictionary mapping channel numbers (int) to threshold values (int)

    Raises:
        ValueError: If format is invalid

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

        # Parse threshold string
        thresholds = parse_thresholds("1:300;2:300;3:300")
        # Returns: {1: 300, 2: 300, 3: 300}

        # Use with Command API
        with connect() as device:
            cmd = Command(device)
            cmd.thresholds(thresholds)
        ```
    """
    logger.debug("Parsing threshold string", extra={"input": value})
    thresholds = {}
    for i, pair in enumerate(value.split(";")):
        try:
            key, val = pair.split(":")
            thresholds[int(key)] = int(val)
        except ValueError as e:
            logger.warning(
                "Threshold parse error",
                extra={
                    "pair": pair,
                    "pair_index": i,
                    "error": str(e),
                }
            )
            raise ValueError(f"Invalid threshold format: '{pair}'") from e

    logger.debug(
        "Threshold string parsed successfully",
        extra={
            "channel_count": len(thresholds),
            "channels": sorted(thresholds.keys()),
        }
    )
    return thresholds