Skip to content

harp-protocol#

PyPI version

The Harp Protocol is a binary communication protocol created to facilitate and unify the interaction between different devices. It was designed with efficiency and ease of parsing in mind.

For more detail please check the official Harp Tech documentation.

harp-protocol provides the building blocks: message framing and the typed register/payload DSL. Each register builds its frames with format and decodes them with parse.

import numpy as np
from harp.protocol import HarpMessage, RegisterU16

class WhoAmI(RegisterU16):
    address = 0

frame = WhoAmI.format(np.uint16(1216))           # build a Write frame
value = WhoAmI.parse(HarpMessage.parse(frame))   # -> np.uint16(1216)

Register value types#

parse returns numpy scalars rather than plain int or float, so a value carries the width its register declares. A Python int has no width and no upper bound, so it cannot distinguish a U8 from a U32, nor detect a value leaving the register range.

np.uint16(65535) + 1   # RuntimeWarning: overflow encountered in scalar add
65535 + 1              # 65536, wider than the register can hold

Numpy scalars behave like plain Python numbers in arithmetic, comparison, and formatting. Use int() or float() where a built-in type is required.

Read an address no schema describes#

A register class is normally declared with its address, as above, or generated from a device.yml. Calling a register base with an address instead builds a one-off register for that address, which is how a payload outside any schema is read and written:

from harp.protocol import RegisterU8Array, RegisterU16

uid = RegisterU8Array(0x10, length=16)   # R_UID, named by no schema
tag = RegisterU8Array(0x11, length=16)   # R_TAG, the firmware git hash
version = RegisterU16(0x08)              # any address, as a scalar

The result is an ordinary register, so it goes through read and write on a device exactly as a declared one does. length is keyword-only for the array form, and it is the element count rather than a byte count. An already-addressed register rejects the call, so WhoAmI(44) raises rather than quietly producing a register at another address.

It carries no transport or device logic. See harp-device for the device layer.

harp-protocol is released as open source under the MIT license. Bug reports and contributions are welcome at the GitHub repository.

harp.protocol #

AnonymousPayload #

Bases: PayloadBase[NpStructT]

Payload backed by a single unnamed numpy dtype (scalar or sub-array).

Subclasses declare the dtype via class kwargs, not descriptors:

class PayloadU16(AnonymousPayload, scalar_dtype="<u2"): ...

Used for scalar/array Harp registers whose payload has no internal structure. RegisterBase.parse unwraps these to a raw numpy scalar for 0-D, or an ndarray for a sub-array or batch. There is no .value accessor and the slot name value is free for use by struct payloads.

For a register that carries one decoded value (a codec, enum, or flag), a subclass declares a single __value__ descriptor field. The descriptor may be a :class:Field (any :class:Converter codec), a :class:GroupMask (enum), or a :class:BitMask (flag)::

class DeviceNamePayload(AnonymousPayload[np.uint8]):
    __value__: str = Field(StringConverter(25))

class EncoderModePayload(AnonymousPayload[np.uint8]):
    __value__: EncoderModeMask = GroupMask(enum=EncoderModeMask, mask=0xFF)

class ResetDevicePayload(AnonymousPayload[np.uint8]):
    __value__: ResetFlags = BitMask(enum=ResetFlags)

The __value__ field makes "this payload is one value" structural and explicit: exactly one field, named __value__, unwrapped on parse and accessed as .__value__. Declaring any other field is an error. Because the value stays a descriptor, to_columns keeps full rendering, meaning enum categoricals and demux_bit_masks flag expansion. __value__ is mutually exclusive with scalar_dtype=.

Every concrete subclass must define its single slot exactly one way: __value__ (a descriptor) or scalar_dtype= (a raw numpy dtype; an explicit dtype in the body, used by the array-register metaclass, also counts). Defining none is a definition-time error.

BitMask #

Bases: Generic[F]

Descriptor for a masked enum.IntFlag view of a payload element.

The flag counterpart of :class:GroupMask: the raw value is extracted as element & mask and mapped to an enum.IntFlag member. Decoding is permissive, so combined flag values such as A | B are valid, matching the unchecked cast of the C# generator. enum= is required and must be an IntFlag subclass.

Unlike :class:GroupMask there is no shift: IntFlag member values are absolute bit positions, so the flags are read and written in place. mask defaults to the full base element, the common whole-register bitMask case, and may be narrowed to embed a flag set inside a wider element. The element width and storage slot are derived from the base element type of the payload, so several masked fields at the same offset share storage automatically.

__init__(*, enum, mask=None, offset=0, default=_MISSING) #

Instantiates a BitMask field for the payload

BoolConverter #

Bases: Converter[bool]

Whole-element interfaceType: bool (or a single masked bit via Field(BoolConverter(), mask=...)).

A non-zero element becomes True. Operates on a single base element.

Column dataclass #

One column of a batched payload.

data is a 1-D numpy array (one row per frame). When categories is not None the column is enum-backed: data holds integer category codes and categories the ordered labels, so a consumer can map codes to labels without copying.

eq=False keeps identity comparison, since field-wise equality would hit the numpy ambiguous-truth-value error on the data array.

name is None for an anonymous single value

Parameters:

Name Type Description Default
name str | None
required
data Any
required
categories Any | None
None

Converter #

Bases: ABC, Generic[T]

Abstract base for payload field converters.

Subclasses must set `dtype as class attribute and implement the abstract methods.

decode_batch(view) abstractmethod #

Decode a 1-D structured-array column into an array-like.

decode_scalar(view) abstractmethod #

Decode a 0-D structured-array element into a Python value.

encode_into(view, value) abstractmethod #

Write a Python value back into a structured-array element.

EnumConverter #

Bases: Converter[E]

Whole-element interfaceType: <maskType> enum (strict).

Maps a base element to an enum.IntEnum member; an unknown code raises ValueError (matching Python IntEnum semantics). For masked enum sub-fields use :class:~harp.protocol.GroupMask with enum= instead.

Field #

Bases: Generic[T]

Descriptor for a payload view decoded through a :class:Converter.

Two modes, selected by mask:

  • Whole-element, with mask=None as the default. The view reads converter.dtype.itemsize bytes starting at offset, in base-element units as described in :class:StructPayload, and runs them through converter. The converter owns its own dtype, and so its own byte layout, and is independent of the base element type of the payload, so the same converter works under any register width.
  • Masked sub-field, with mask set. The raw value is extracted as (element & mask) >> shift from the base element of the payload at offset and then run through converter, which dictates the output type. The right-shift is derived from the trailing-zero count of mask. Several masked fields at the same offset share the element slot automatically, and may share it with a :class:GroupMask or :class:BitMask on the same word.

offset defaults to 0. Omitting it suits a payload with a single member; when a payload has several distinct slots, each must declare an explicit offset= or the overlap check rejects the layout.

__init__(converter, *, mask=None, offset=0, default=_MISSING) #

Instantiates a new payload Field.

GroupMask #

Bases: Generic[E]

Descriptor for a masked, shifted enum sub-field of a payload element.

Syntactic sugar over a masked :class:Field: the raw value is extracted as (element & mask) >> shift and mapped strictly to an enum.IntEnum member, and an unknown code raises. enum= is required. For masked numeric fields use Field(converter=..., mask=...) instead.

The right-shift is always derived from the trailing-zero count of mask, so the field aligns to bit 0, and offset defaults to 0. The element width and storage slot are derived from the base element type of the payload, so several masked fields at the same offset share storage automatically.

__init__(*, mask, enum, offset=0, default=_MISSING) #

Instantiates a GroupMask field for the payload

HarpMessage #

Bases: Generic[P]

A Harp message backed by its raw frame bytes, parameterized by its payload type.

Build with the constructor or parse from wire bytes with HarpMessage.parse(). A message off the wire is a HarpMessage[Any], since a frame declares only how its payload is encoded and not which register contract it satisfies. Decoding it with a register yields a HarpMessage[P], whose payload is that contract.

address property #

Return the address byte of this message.

bytes property #

The complete raw message frame, including checksum.

has_error property #

Return True if the error flag is set in this message.

has_payload property #

Return True if a register has decoded the payload of this message.

has_timestamp property #

Return True if the timestamp flag is set in this message.

message_type property #

Return the MessageType of this message.

payload property #

The decoded payload, as the register that parsed this message defines it.

Only a register knows which contract a frame satisfies, so a message read from the wire carries no payload until one decodes it. Raises ValueError in that case; test with has_payload first, or read payload_bytes instead.

payload_bytes property #

Payload bytes, excluding timestamp and checksum.

payload_type property #

Return the PayloadType of this message.

port property #

Return the port byte of this message.

timestamp property #

Return the timestamp of this message, or None if not present.

decode(decoder) #

Return a copy of this message with its payload decoded by decoder.

The payload is derived from the frame in the same call, so the two cannot disagree. The payload type and the byte count are both checked, since together they decide whether these bytes can be read as this payload at all. The address is not, so a frame may be decoded by anything describing the same layout.

parse(data) classmethod #

Parse and validate a complete Harp Message from a byte sequence. Raises HarpParseError on failure.

HarpParseError #

Bases: Exception

An exception raised for errors encountered during message parsing

HarpVersion dataclass #

Represents a Harp version

Parameters:

Name Type Description Default
major int
required
minor int
required
patch int
required

HarpVersionConverter #

Bases: Converter[HarpVersion]

A built-in converter for a HarpVersion object.

IdentityConverter #

Bases: Converter[NpScalarT]

Pass-through converter, returning the raw numpy scalar as-is.

MessageType #

Bases: IntEnum

Represents the a message type from the harp protocol

PayloadBase #

Bases: Generic[NpStructT]

Base class for typed Harp register payloads.

A subclass declares fields via the scalar descriptors above. __init_subclass__ auto-derives a Batch sibling subclass with the same dtype but each descriptor swapped to a Batch variant returning an NDArray view. from_array routes by ndim so callers never need to mention the Batch class explicitly: 0-D records stay scalar, 1-D buffers become Batch.

payload_as_columns(*, decode_enums=True, demux_bit_masks=False) #

Returns a list of Column where each member represents a field from a payload across multiple messages.

decode_enums controls whether GroupMask enum columns become category codes and labels when True, or raw integer codes when False, which is a shape-preserving relabel. demux_bit_masks controls whether a BitMask flag column is expanded into one boolean column per flag member when True, or kept as a single raw-integer column when False, which is a shape change. The two are orthogonal and apply to different descriptor kinds.

PayloadType #

Bases: Enum

Harp scalar payload types. Each value is the corresponding numpy dtype.

numpy_dtype property #

Returns the corresponding numpy dtype

PayloadTypeInfo dataclass #

PayloadTypeInfo(has_timestamp: bool, payload_type: harp.protocol._payload_type.PayloadType, element_size: int)

Parameters:

Name Type Description Default
has_timestamp bool
required
payload_type PayloadType
required
element_size int
required

RegisterBase #

Bases: ABC, Generic[U]

Abstract base for all typed Harp registers.

The generic parameter U is the static return type of :meth:parse, the user-facing value, not necessarily payload_class, which is the wire encoding. The two coincide only for multi-member struct payloads:

  • scalar registers -> a numpy scalar, for example np.uint16;
  • array registers -> NDArray[...] of fixed length;
  • multi-member struct registers -> the payload class itself;
  • single-member registers that unwrap on parse -> the inner value type, for example RegisterBase[str] for DeviceName, RegisterBase[HarpVersion], or RegisterBase[ClockConfigurationFlags] for a whole-register BitMask or GroupMask, even though each still has a payload_class.

Subclasses must define address, payload_type, and payload_class as ClassVars. The extent of a payload is always read from payload_class.

format(value=_MISSING, *, message_type=None, timestamp=None, port=_DEFAULT_PORT) classmethod #

format(
    *,
    message_type: MessageType = MessageType.Read,
    timestamp: float | None = None,
    port: int = _DEFAULT_PORT,
) -> bytes

format(
    value: U,
    *,
    message_type: MessageType = MessageType.Write,
    timestamp: float | None = None,
    port: int = _DEFAULT_PORT,
) -> bytes

Build a Harp frame for this register. No value gives a Read, a value gives a Write.

format_bulk(values, *, timestamps=None, message_type=MessageType.Event, port=_DEFAULT_PORT) classmethod #

Build a flat buffer of N frames of this register type, the inverse of :meth:parse_bulk.

values is a payload, either scalar or :class:Batch, or an ndarray of the payload_class.payload_dtype of the register. timestamps, a length-N array of seconds, makes every frame timestamped. message_type is one :class:MessageType for all frames, or a length-N array of message-type bytes or values, for example the msgtype view returned by parse_bulk.

parse(value) classmethod #

Parse a single message into the user-facing payload value.

Struct payloads return a typed wrapper (descriptor access like payload.Channel0 works). Anonymous payloads (scalar / array registers) return the raw numpy scalar or ndarray directly.

parse_bulk(source, *, parse_timestamp=True) classmethod #

Parse a bulk buffer containing one or more frames of this register type. Returns (data, timestamps, msgtype_view, payload).

RegisterFloat #

Bases: RegisterBase[float32]

A simple scalar register with a float32 payload. parse() returns np.float32.

RegisterFloatArray #

Bases: RegisterBase[NDArray[float32]]

A simple array register with a float32 array payload. It must be instantiated with a length: RegisterFloatArray(0x28, length=3). parse() returns NDArray[np.float32].

RegisterS16 #

Bases: RegisterBase[int16]

A simple scalar register with an int16 payload. parse() returns np.int16.

RegisterS16Array #

Bases: RegisterBase[NDArray[int16]]

A simple array register with an int16 array payload. It must be instantiated with a length: RegisterS16Array(0x28, length=3). parse() returns NDArray[np.int16].

RegisterS32 #

Bases: RegisterBase[int32]

A simple scalar register with an int32 payload. parse() returns np.int32.

RegisterS32Array #

Bases: RegisterBase[NDArray[int32]]

A simple array register with an int32 array payload. It must be instantiated with a length: RegisterS32Array(0x28, length=3). parse() returns NDArray[np.int32].

RegisterS64 #

Bases: RegisterBase[int64]

A simple scalar register with an int64 payload. parse() returns np.int64.

RegisterS64Array #

Bases: RegisterBase[NDArray[int64]]

A simple array register with an int64 array payload. It must be instantiated with a length: RegisterS64Array(0x28, length=3). parse() returns NDArray[np.int64].

RegisterS8 #

Bases: RegisterBase[int8]

A simple scalar register with an int8 payload. parse() returns np.int8.

RegisterS8Array #

Bases: RegisterBase[NDArray[int8]]

A simple array register with an int8 array payload. It must be instantiated with a length: RegisterS8Array(0x28, length=3). parse() returns NDArray[np.int8].

RegisterU16 #

Bases: RegisterBase[uint16]

A simple scalar register with a uint16 payload. parse() returns np.uint16.

RegisterU16Array #

Bases: RegisterBase[NDArray[uint16]]

A simple array register with a uint16 array payload. It must be instantiated with a length: RegisterU16Array(0x28, length=3). parse() returns NDArray[np.uint16].

RegisterU32 #

Bases: RegisterBase[uint32]

A simple scalar register with a uint32 payload. parse() returns np.uint32.

RegisterU32Array #

Bases: RegisterBase[NDArray[uint32]]

A simple array register with a uint32 array payload. It must be instantiated with a length: RegisterU32Array(0x28, length=3). parse() returns NDArray[np.uint32].

RegisterU64 #

Bases: RegisterBase[uint64]

A simple scalar register with a uint64 payload. parse() returns np.uint64.

RegisterU64Array #

Bases: RegisterBase[NDArray[uint64]]

A simple array register with a uint64 array payload. It must be instantiated with a length: RegisterU64Array(0x28, length=3). parse() returns NDArray[np.uint64].

RegisterU8 #

Bases: RegisterBase[uint8]

A simple scalar register with a uint8 payload. parse() returns np.uint8.

RegisterU8Array #

Bases: RegisterBase[NDArray[uint8]]

A simple array register with a uint8 array payload. It must be instantiated with a length: RegisterU8Array(0x28, length=3). parse() returns NDArray[np.uint8].

StringConverter #

Bases: Converter[str]

Converts a fixed-length byte array to/from a Python str.

StructPayload #

Bases: PayloadBase[NpStructT]

Base class for struct register payloads with typed field descriptors.

Subclasses declare fields using Field, GroupMask, or BitMask descriptors. Type checkers synthesize a keyword-only __init__ from those declarations, so constructor calls are fully type-checked and have IDE autocompletion.

The type argument (StructPayload[np.uint8]) is the base element type; it sets the unit for offset= and the integer width of masked reads. The optional length= class kwarg fixes the payload size in base elements (the register length); when omitted it defaults to the max member extent.

Example::

class MyPayload(StructPayload[np.uint8]):
    channel: np.uint16 = Field(UInt16Converter(), offset=0)
    flags: MyFlags = BitMask(enum=MyFlags, offset=2)

decode_payload_type(b) #

Decode a PayloadType byte. Raises ValueError for invalid bytes.

encode_payload_type(payload_type, *, has_timestamp=False) #

Encode a PayloadType back to a protocol byte.