katcp_codec package

Module contents

class katcp_codec.MessageType(*values)

Bases: Enum

Type of katcp message.

REQUEST = 1

REPLY = 2

INFORM = 3

class katcp_codec.Message(mtype: MessageType, name: bytes, mid: int | None, arguments: List[bytes], validate: dataclasses.InitVar[bool] = True)

Bases: object

A katcp message.

Raises:

• OverflowError – if the message ID is out of range

• ValueError – if the name does not conform to the specification

mtype: MessageType

Message type

name: bytes

Message name (must start with [A-Za-z] and contain only [A-Za-z0-9-])

mid: int | None

Message ID (if specified, must be positive and less than 2**31)

arguments: List[bytes]

Message arguments

validate: dataclasses.InitVar[bool] = True

If false, skip validating the name and message ID. This should only be done if they’ve already been checked, as the behaviour is undefined if they’re invalid.

class katcp_codec.Parser(max_line_length: int)

Bases: object

Message parser.

The parser accepts chunks of data from the wire (which need not be aligned to message boundaries) and returns whole messages as they are parsed.

Parameters:

max_line_length – The maximum number of bytes in a message. Longer messages will not break the parser but will be reported as errors.

append(data: Buffer) → List[Message | ValueError]

Append new data to the parser.

Returns:

Messages whose end was in the input data. Each message is either an instance of Message if it was valid or ValueError if not.

Return type:

messages

reset() → None

Reset the parser to its initial state.

This discards any incomplete message from the internal buffer.

property buffer_size: int

Get the current size of the internal buffer.

This contains the number of bytes received for the current partial message (if any). This does not indicate the memory usage, since those bytes may have already been parsed into internal structures.