harp-device#
The transport-agnostic device layer for the Harp protocol: the core Harp registers and a Device base that handles framing, request/reply and register access. It depends only on harp-protocol, with no transport dependencies. Pair it with a transport such as harp-serial.
Read/write registers#
A Device operates over a transport. read and write take a register class:
from harp.device import core
# `device` is a Device opened over some transport, see harp-serial
who = device.read(core.WhoAmI).payload # -> np.uint16
device.write(core.OperationControl, payload) # write a register
When a request fails#
An error reply raises DeviceError, which keeps the reply as reply so the frame sent by the device stays available for inspection. Pass raise_on_error=False to the constructor to receive such a reply as an ordinary return value instead. A transport failure raises TransportError, and every later request reports the same failure rather than waiting for a reply that cannot arrive. A device that never answers raises TimeoutError after REPLY_TIMEOUT, which is also what happens when close is called during a request.
Extend for a specific device#
A device is described by a module. Downstream, often generated, packages record the device identity as WHO_AM_I, declare the register classes at module level, and expand the core REGISTER_MAP beside them:
from harp.device.core import REGISTER_MAP as _CORE_REGISTER_MAP
WHO_AM_I: int = 1216
REGISTER_MAP = {**_CORE_REGISTER_MAP, 32: DigitalInputState, ...}
This is the same structure create_device_module builds from a schema, so a device reads the same way whether it was generated ahead of time or compiled at runtime. A WHO_AM_I of 0 marks an unregistered device, used while a device is in development or outside the official registry, and identity checks are skipped for it.
A device module names only what its schema declares, the registers beside the enums and payload classes built from them. The core registers and any core mask reused by the schema have a single definition, in harp.device.core, and are accessed from there rather than through the device module. The core register set is not a device, so it carries no WHO_AM_I. REGISTER_MAP covers the complete device address space, including both core and application registers.
Pass the module to Device, or to open_device, to validate identity on open:
from harp.device import behavior, client, core
with client.Device(transport, behavior) as device:
device.read(core.WhoAmI) # a core register
device.read(behavior.DigitalInputState) # declared by the schema
The WHO_AM_I in the module determines the check, and 0 skips it. Omitting the module skips validation. The module is not otherwise consulted: registers are passed to read, write and subscribe as arguments either way, and only a subscribed register is parsed on arrival. Core registers such as WhoAmI and OperationControl come from harp.device.core and are read the same way.
A new transport is just an object implementing the ITransport protocol, with open, write, read and close.
Generate registers from a device.yml#
Without a pre-generated device package, create_device_module builds the same structure at runtime from Harp device.yml text. It emits register, enum, and payload classes at module level, a REGISTER_MAP beside them, and the identity declared by the schema as WHO_AM_I. Identifiers match a generated package name for name: register, enum, and payload class names come from the yml verbatim, payload fields are snake_case, and enum members are SCREAMING_SNAKE_CASE. A maskType the schema does not declare resolves against the core masks, and a register marked private is emitted with an underscore-prefixed name.
from pathlib import Path
from harp.device import schema
behavior = schema.create_device_module(Path("device.yml").read_bytes())
reg = behavior.AnalogData # by name
reg = behavior.REGISTER_MAP[44] # or by address
The module is not registered in sys.modules, so it has to be bound rather than imported. Names come from the schema at runtime, so they don't autocomplete and aren't statically checked. A generated package on disk gives both.
For a custom interfaceType, pass its converter via converters=, keyed by {InterfaceType}Converter or {MemberName}Converter. An unresolved custom type raises UnknownConverterError, or pass require_converters=False to decode it natively:
parse_device_schema(yml_text) is also public, returning the parsed schema model without a module: registers, masks, and optional device identity.
harp-device is released as open source under the MIT license. Bug reports and contributions are welcome at the GitHub repository.
harp.device.client.Device
#
Bases: Generic[M]
Harp device protocol logic (framing, request/reply, register access)
over an :class:~harp.device.client.ITransport.
Must be opened before use, via with or :meth:open. :meth:read,
:meth:write and :meth:subscribe take a register class directly.
Pass a device_module (from :func:~harp.device.schema.create_device_module or a
statically generated device package) to validate device identity on open::
behavior = create_device_module(schema_text)
with Device(transport, behavior) as dev:
dev.read(behavior.OperationControl)
Omitting device_module skips that check. The module is not otherwise
consulted: registers are passed to :meth:read, :meth:write and :meth:subscribe
as arguments either way, and only a subscribed register is parsed on arrival.
A request fails in one of three ways. An error reply raises :class:DeviceError
carrying the reply, unless raise_on_error=False. A transport failure raises
:class:~harp.device.client.TransportError, and every later request reports the
same failure rather than waiting for a reply that cannot arrive. A device that
never answers raises :class:TimeoutError after REPLY_TIMEOUT, which is also
what happens when :meth:close is called during a request.
module
property
#
The device module injected at construction, or None if not set.
open()
#
Open the transport, start the reader thread and validate identity.
subscribe(register, handler, *, message_types=MessageType.Event)
#
Call handler with a :class:~harp.protocol.HarpMessage typed by the
payload of register, each time the device emits a message for it.
By default only unsolicited Event messages are delivered. Pass
message_types (a :class:MessageType or an iterable of them) to also
observe Read/Write replies, e.g.
message_types=(MessageType.Event, MessageType.Write).
Handlers run on a single dedicated event thread, shared by all
subscribers, so they may block or call back into :meth:read/:meth:write
without deadlocking the reader or delaying synchronous requests. However,
because that thread is shared, handlers are invoked sequentially, in
subscription order, one message at a time: a slow handler delays every
other subscriber and backs up later messages. Keep handlers quick, and
offload heavy work to a separate thread or queue.
Returns a :class:Subscription; call :meth:Subscription.unsubscribe to
stop.
subscribe_all(handler, *, message_types=MessageType.Event)
#
Call handler with the raw :class:HarpMessage for every message,
regardless of address, whose type is in message_types (default:
Event only). Pass more types for a full-traffic firehose, e.g. a
logger. See :meth:subscribe for threading and cancellation semantics.
harp.device.client.DeviceError
#
Bases: Exception
Raised when the device replies to a request with the error flag set.
The reply is kept as :attr:reply, so the frame sent by the device stays available
for inspection. Construct the device with raise_on_error=False to receive such
a reply as an ordinary return value instead.
reply = reply
instance-attribute
#
The error reply, as received.
harp.device.client.Subscription
#
Handle returned by :meth:Device.subscribe. Cancel with
:meth:unsubscribe, or use as a context manager to auto-cancel on exit.
unsubscribe()
#
Stop delivering events to this subscription. Idempotent.
harp.device.client.EventHandler = Callable[[HarpMessage[P]], None]
module-attribute
#
A callback receiving a message typed by the payload of a specific register.
harp.device.client.HarpFramer
#
Stateful Harp message stream parser.
Feed raw bytes incrementally with feed(), then drain complete frames with next_frame() or by iterating. Suitable for both file parsing and streaming sources (e.g. serial ports) where data arrives in chunks.
Recovery: on checksum or PayloadType failure, the framer skips exactly the bad MessageType byte and retries from the next byte, matching the C# StreamTransport resynchronisation strategy.
feed(data)
#
Append new bytes to the internal buffer.
frames()
#
Yield all complete frames currently available in the buffer.
next_frame()
#
Return the next complete, valid HarpMessage, or None if not enough data.
parse_bytes(data)
classmethod
#
Parse all Harp messages from a byte buffer.
parse_file(path)
classmethod
#
Parse all Harp messages from a binary file.
harp.device.client.ITransport
#
Bases: Protocol
Byte channel a :class:~harp.device.client.Device drives.
Owns no protocol logic. Failures are reported as :class:TransportError.
read()
#
Return available bytes, or b'' on idle/timeout.
harp.device.client.TransportError
#
Bases: Exception
Raised by a transport when the underlying byte channel fails.
harp.device.schema.create_device_module(text, *, name=None, converters=None, require_converters=True)
#
Emit a module of register classes from device.yml text.
The module names what the schema declares, its registers beside the enums and
payload classes they are built from, so behavior.AnalogData,
behavior.AnalogDataPayload and behavior.EncoderModeMask all resolve while a
core register such as WhoAmI is imported from :mod:harp.device.core, keeping
one definition of each. This is the same set a generated device package holds. A
name describing two declarations is rejected rather than shadowed. Beside them
it holds:
REGISTER_MAP, the device address space, so the core registers are present here even though the module does not name them;WHO_AM_I, the identity declared by the schema (0for an unregistered device);DEVICE_NAME, thedevicename of the schema, ornamewhen given, and empty for a header-less register fragment. Recordings are written under this name, so :class:~harp.data.DatasetReadermatches files by it;__name__, the same name, falling back to"Device"so the module is never anonymous. This names the module rather than the device, and is not part of what a device module promises;__doc__, the optionaldescriptionof the schema.
Because the names come from the schema at runtime they don't autocomplete, and
each resolves as Any rather than its own type. A generated device package is
a real module on disk and gives both. On an address clash the device register
replaces the core one in REGISTER_MAP.
text is the schema itself rather than a path to it, matching
:func:parse_device_schema, so read the file first. The module is not
registered in :data:sys.modules, so it cannot be imported and two
schemas may share a name without clashing. Bind it yourself::
behavior = create_device_module(Path("device.yml").read_bytes())
behavior.AnalogData
harp.device.schema.parse_device_schema(text)
#
Parse a Harp device.yml (or a header-less fragment) into a :class:DeviceModel.
A header-less fragment (just registers / bitMasks / groupMasks)
parses fine, and the identity fields such as device and whoAmI are simply
None. Read files yourself, e.g.
parse_device_schema(Path("device.yml").read_bytes()). Prefer reading bytes:
a YAML stream declares its own encoding, so the parser decodes it, whereas
read_text() without an explicit encoding uses the locale default.
Uses pydantic-yaml (ruamel-backed, YAML 1.2), so group-mask keys like
Off / On stay strings instead of being coerced to booleans.
harp.device.schema.ConverterContext
dataclass
#
The schema definition of a payload value, resolved against its register context.
Handed to every converter factory so it can construct the converter with the
right arguments, for example StringConverter(span), HarpVersionConverter(element),
or IdentityConverter(dtype).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
|
required |
interface_type
|
str | None
|
|
required |
mask
|
int | None
|
|
required |
length
|
int
|
|
required |
element
|
dtype
|
|
required |
element_size
|
int
|
|
required |
member_dtype
property
#
The numpy dtype of the value itself. A native primitive interfaceType overrides the element.
raw_dtype
property
#
Native passthrough dtype, a sub-array when the value spans more than one element.
span
property
#
Byte span of the value, its element count times the element size.
harp.device.schema.DeviceModule
#
Bases: ModuleType
The type of the module returned by :func:create_device_module.
The declarations of the schema are accessed by name and typed Any, since they
exist only at runtime. DEVICE_NAME, REGISTER_MAP, WHO_AM_I and
__all__ are declared here and carry their own types.
DEVICE_NAME
instance-attribute
#
The device name declared by the schema. Empty when absent.
REGISTER_MAP
instance-attribute
#
Address -> register class, the core Harp registers merged with those of the schema.
WHO_AM_I
instance-attribute
#
The device identity declared by the schema. 0 when absent.
__all__
instance-attribute
#
The declarations of the schema, beside REGISTER_MAP and WHO_AM_I.
harp.device.schema.DeviceModuleLike
#
Bases: Protocol
Any module describing a device, however it was produced.
A generated device package is a plain module, so it cannot be named by a class;
what identifies it is describing a device. Matching structurally accepts both it
and :class:DeviceModule, and rejects the core register set, which carries
registers but is not a device.
DEVICE_NAME is required rather than optional, so a generated package always
states the name used for its recordings. A schema declaring none still
builds, since :class:DeviceModule declares the member and leaves it empty.
harp.device.core
#
The core register set every Harp device carries, and its address space.
AssemblyVersion
#
ClockConfiguration
#
Bases: RegisterBase[ClockConfigurationFlags]
Specifies the configuration for the device synchronization clock.
ClockConfigurationFlags
#
Bases: IntFlag
Specifies configuration flags for the device synchronization clock.
CLOCK_GENERATOR = 2
class-attribute
instance-attribute
#
The device resets and generates the clock synchronization signal on the clock output connector, if available.
CLOCK_LOCK = 128
class-attribute
instance-attribute
#
The device will lock the timestamp register counter and will not accept commands to set new timestamp values.
CLOCK_REPEATER = 1
class-attribute
instance-attribute
#
The device will repeat the clock synchronization signal to the clock output connector, if available.
CLOCK_UNLOCK = 64
class-attribute
instance-attribute
#
The device will unlock the timestamp register counter and will accept commands to set new timestamp values.
GENERATOR_CAPABILITY = 16
class-attribute
instance-attribute
#
Specifies the device has the capability to generate the clock synchronization signal to the clock output connector.
REPEATER_CAPABILITY = 8
class-attribute
instance-attribute
#
Specifies the device has the capability to repeat the clock synchronization signal to the clock output connector.
ClockConfigurationPayload
#
CoreVersionHigh
#
CoreVersionLow
#
DeviceName
#
DeviceNamePayload
#
EnableFlag
#
FirmwareVersionHigh
#
FirmwareVersionLow
#
HardwareVersionHigh
#
HardwareVersionLow
#
OperationControl
#
OperationControlPayload
#
Bases: StructPayload[uint8]
Represents the payload of the OperationControl register.
dump_registers = Field(BoolConverter(), mask=8)
class-attribute
instance-attribute
#
Specifies whether the device should report the content of all registers on initialization.
heartbeat = GroupMask(enum=EnableFlag, mask=128)
class-attribute
instance-attribute
#
Specifies whether the device should report the content of the seconds register each second.
mute_replies = Field(BoolConverter(), mask=16)
class-attribute
instance-attribute
#
Specifies whether the replies to all commands will be muted, i.e. not sent by the device.
operation_led = GroupMask(enum=EnableFlag, mask=64)
class-attribute
instance-attribute
#
Specifies whether the device state LED should report the operation mode of the device.
operation_mode = GroupMask(enum=OperationMode, mask=3)
class-attribute
instance-attribute
#
Specifies the operation mode of the device.
visual_indicators = GroupMask(enum=EnableFlag, mask=32)
class-attribute
instance-attribute
#
Specifies the state of all visual indicators on the device.
OperationMode
#
Bases: IntEnum
Specifies the operation mode of the device.
ResetDevice
#
ResetDevicePayload
#
ResetFlags
#
Bases: IntFlag
Specifies the behavior of the non-volatile registers when resetting the device.
BOOT_FROM_DEFAULT = 64
class-attribute
instance-attribute
#
Specifies that the device has booted from default factory values.
BOOT_FROM_EEPROM = 128
class-attribute
instance-attribute
#
Specifies that the device has booted from non-volatile values stored in EEPROM.
RESTORE_DEFAULT = 1
class-attribute
instance-attribute
#
The device will boot with all the registers reset to their default factory values.
RESTORE_EEPROM = 2
class-attribute
instance-attribute
#
The device will boot and restore all the registers to the values stored in non-volatile memory.
RESTORE_NAME = 8
class-attribute
instance-attribute
#
The device will boot with the default device name.
SAVE = 4
class-attribute
instance-attribute
#
The device will boot and save all the current register values to non-volatile memory.
UPDATE_FIRMWARE = 32
class-attribute
instance-attribute
#
The device will enter firmware update mode.