Discussion on the what the public API should be

[ZegCricket]

On the Harp SRM from yesterday (23/01/2026), two concepts for the public API were presented and discussed, although a decision on what concept to follow is yet to be made.

First concept

The first concept consists on classes that represent a HarpMessage from a specific register. This includes the message itself as well as the correct payload for that register (especially registers with payloadSpec).

Class Example

class WhoAmI(HarpMessage[int]):
    ADDRESS: int = 0
    PAYLOAD_TYPE: PayloadType = PayloadType.U16

    def __init__(
        self,
        payload: Optional[int] = None,
        message_type: MessageType = MessageType.READ,
        # This can be removed since we can infer this through the timestamp parameter
        # payload_type: Literal[
        #    PayloadType.U16, PayloadType.TIMESTAMPED_U16
        #] = PayloadType.U16,
        *,
        timestamp: Optional[float] = None,
        port: int = 255,
    ):
        super().__init__(
            message_type, self.ADDRESS, self.PAYLOAD_TYPE, payload, timestamp, port
        )

Usage example

from harp.serial import Device
from harp.protocol import WhoAmI

with Device("/dev/ttyUSB0") as device:
    # Always returns HarpMessage 
    reply_msg = device.send(WhoAmI())
    print(reply_msg) # This prints "Type: READ \r\n Length....." (__str__ method from HarpMessage)
    print(reply_msg.payload) # This prints "1236"

    # or alternatively, get the value itself
    reply_value = device.read_who_am_i()
    print(reply_value) # This prints "1236" 

Second concept

The second concept consists on creating an instance of a Spec (or another name) that has the register's address, payload_type and a "decode" and "encode" methods.

Spec example

@dataclass(frozen=True)
class Spec(Generic[T]):
    addr: int
    payload_type: PayloadType
    decode: Callable[[Any], T]  # from payload (int|float|list) -> T
    encode: Callable[[T], Any]  # from T -> payload (int|float|list

The creation of a Spec for the WhoAmI register is as follows:

def _id(x):
    return x

COMMON_SPECS = {
    CommonRegisters.WHO_AM_I: Spec(
        addr=0,
        payload_type=PayloadType.U16,
        decode=int,
        encode=_id,
    ),
    ...
}

Usage example

from harp.serial import Device
from harp.protocol import WhoAmI

with Device("/dev/ttyUSB0") as device:
    # There's a read_reg and write_reg that return the HarpMessage
    reply_msg = device.read_reg(CommonRegisters.WHO_AM_I)
    print(reply_msg) # This prints "Type: READ \r\n Length....." (__str__ method from HarpMessage)
    print(reply_msg.payload) # This prints "1236"

    # or alternatively, get the value itself
    # internally, this method will call "decode" over the corresponding Spec
    reply_value = device.read_who_am_i()
    print(reply_value) # This prints "1236" 

Things to consider:

• Try to eventually unify with harp-python, with focus on vectorization on load of .bin files

• When developing a new device, when there's still no specific device package, we can simply use the send method to send a custom HarpMessage for a register of that new device.

• Ease of conversion from and to payloadSpec

Any further suggestions? Which approach seems better so that we can direct our efforts to that one?

[bruno-f-cruz]

@ZegCricket thanks for the summary!
I am going to add a few more notes with what I had in mind:

The examples below should be taken as pseudo-code, I made very little effort into making sure the syntax is correct, sorry!

The main goal was to have a parsing/formatting internal class that allows for vectorized/bulk reads from files/memory spans.

I am not super happy with the "write" interface, but i think this is the cost from abstracting the write into the device class instead of the register.

Most of it is pretty similar to what you presented, but I think adopting this internal representation for the data may help everyone.
There are also non-trivial code that must be generated to parse the layouts, I have left a few more examples in this gist in case you are curious
.

@glopesdev I think the layout representation is all we need for the bulk reads, what do you think?

from typing import TypeVar, Generic, Protocol, ClassVar, Dict, overload
import numpy as np
from numpy.typing import NDArray
import pandas as pd

TPayload = TypeVar("TPayload")


# =============================================================================
# Core Message Types
# =============================================================================

class HarpMessageType:
    """Enum-like class for message types (Read, Write, Event, etc.)"""
    ...


class HarpMessage(Generic[TPayload]):
    """
    Represents a Harp protocol message with typed payload.
    """
    
    _bytes: bytes  # Raw bytes representing the complete message

    @property
    def seconds(self) -> float:
        """Timestamp in seconds extracted from the message header."""
        ...

    @property
    def payload(self) -> bytes:
        """Raw payload bytes (excluding header)."""
        ...

    @classmethod
    def from_bytes(cls, data: bytes) -> "HarpMessage":
        """Construct a message from raw bytes."""
        ...

    @classmethod
    def from_payload(
        cls,
        payload: TPayload | NDArray[np.void] | bytes,
        message_type: HarpMessageType,
        timestamp: float,
        address: int,
    ) -> HarpMessage[TPayload]:
        """
        Construct a message from a typed payload or raw numpy array.
        
        Accepts either:
        - A typed payload object with a `to_bytes()` method
        - A numpy array with structured dtype (uses `tobytes()` directly)
        - Raw bytes representing the payload
        """
        ...

    # Convenience factory methods for primitive types
    @classmethod
    def from_u16(cls, value: int, **kwargs) -> "HarpMessage[int]":
        ...

    @classmethod
    def from_float(cls, value: float, **kwargs) -> "HarpMessage[float]":
        ...

    def to_bytes(self) -> bytes:
        """Return the complete message as bytes for transmission."""
        return self._bytes


# =============================================================================
# Register Protocol
# =============================================================================

class IRegister(Protocol[TPayload]):
    """
    Protocol defining the interface for a Harp register.
    """
    
    address: ClassVar[int]

    @classmethod
    def format(
        cls,
        payload: TPayload,
        message_type: HarpMessageType,
        timestamp: float,
    ) -> HarpMessage[TPayload]:
        """Format a payload into a complete Harp message."""
        ...

    @classmethod
    def parse(cls, message: HarpMessage) -> TPayload:
        """Parse a Harp message into a typed payload."""
        ...

    @classmethod
    def create_default(
        cls,
        message_type: HarpMessageType,
        timestamp: float,
    ) -> HarpMessage[TPayload]:
        """Create a message with default payload values."""
        ...


# =============================================================================
# Base Device Class
# =============================================================================

class HarpDevice:
    """
    Base class for Harp device communication.
    
    Handles serial transport and provides typed read/write operations
    through the register interface.
    """
    
    _serial_transport: "SomeSerialTransportClass"

    def __init__(self, com_port: str, **kwargs):
        ...

    def connect(self) -> None:
        """Establish connection to the device."""
        ...

    def send_message(self, message: HarpMessage[TPayload]) -> HarpMessage[TPayload]:
        """Send a message and return the device's response."""
        self._serial_transport.write(message.to_bytes())
        return self.wait_for_response()

    def wait_for_response(self) -> HarpMessage[TPayload]:
        """Wait for and return the next message from the device."""
        ...

    def read(self, register: IRegister[TPayload] | HarpMessage[TPayload]) -> TPayload:
        """
        Read from a register and return the parsed payload.
        """

    def write(self, message: HarpMessage[TPayload]) -> TPayload:
        """
        Write a payload to a register and return the acknowledged value.
        """
        ...
    
    # The annoying thing about not having the "write" in the register interface is that the device needs to know which register to write to and parse
    def write(self, register: IRegister[TPayload], payload: TPayload, seconds: float, message_type: HarpMessageType) -> TPayload:
        message = register.format(payload, message_type, seconds)
        reponse = self.send_message(message)
        return register.parse(reponse)


# =============================================================================
# Device-Specific Implementation Example
# =============================================================================

class _AnalogDataLayout:
    """
    Internal layout handler for structured numpy array data.
    
    Manages the memory layout and provides named field access.
    """
    
    _arr: NDArray[np.void]
    _dtype: ClassVar[np.dtype] = np.dtype([
        ("analog_input0", "i2"),  # S16 = signed 16-bit
        ("encoder", "i2"),
        ("analog_input1", "i2"),
    ])

    def __init__(
        self,
        *,
        analog_input0: NDArray[np.int16],
        encoder: NDArray[np.int16],
        analog_input1: NDArray[np.int16],
    ):
        n = len(analog_input0)
        if not (len(encoder) == len(analog_input1) == n):
            raise ValueError("All input arrays must have the same length")
        
        arr = np.zeros(n, dtype=self._dtype)
        arr["analog_input0"] = analog_input0
        arr["encoder"] = encoder
        arr["analog_input1"] = analog_input1
        self._arr = arr

    @classmethod
    def from_buffer(cls, buf: bytes | memoryview) -> "_AnalogDataLayout":
        arr = np.frombuffer(buf, dtype=cls._dtype)
        obj = cls.__new__(cls)
        obj._arr = arr
        return obj

    @property
    def analog_input0(self) -> NDArray[np.int16]:
        return self._arr["analog_input0"]

    @property
    def encoder(self) -> NDArray[np.int16]:
        return self._arr["encoder"]

    @property
    def analog_input1(self) -> NDArray[np.int16]:
        return self._arr["analog_input1"]

    def to_array(self) -> NDArray[np.void]:
        return self._arr

    def to_dataframe(self) -> pd.DataFrame:
        return pd.DataFrame({
            "analog_input0": self.analog_input0,
            "encoder": self.encoder,
            "analog_input1": self.analog_input1,
        })


class AnalogDataPayload:
    """
    Typed payload for the AnalogData register.
    
    Wraps the layout class and provides scalar access for single-sample reads.
    """
    
    _layout: _AnalogDataLayout

    @overload
    def __init__(
        self,
        *,
        analog_input0: np.int16,
        encoder: np.int16,
        analog_input1: np.int16,
    ) -> None: ...

    @overload
    def __init__(self, *, buffer: bytes) -> None: ...

    def __init__(
        self,
        *,
        analog_input0: np.int16 | None = None,
        encoder: np.int16 | None = None,
        analog_input1: np.int16 | None = None,
        buffer: bytes | None = None,
    ):
        if buffer is not None:
            self._layout = _AnalogDataLayout.from_buffer(buffer)
        else:
            self._layout = _AnalogDataLayout(
                analog_input0=np.array([analog_input0], dtype=np.int16),
                encoder=np.array([encoder], dtype=np.int16),
                analog_input1=np.array([analog_input1], dtype=np.int16),
            )

    @property
    def analog_input0(self) -> np.int16:
        return self._layout.analog_input0[0]

    @property
    def encoder(self) -> np.int16:
        return self._layout.encoder[0]

    @property
    def analog_input1(self) -> np.int16:
        return self._layout.analog_input1[0]

    def to_bytes(self) -> bytes:
        return self._layout.to_array()[0].tobytes()


class AnalogDataRegister(IRegister[AnalogDataPayload]):
    """
    Register definition for AnalogData (address 44).
    
    Provides formatting, parsing, and convenience creation methods.
    """
    
    address: ClassVar[int] = 44
    dtype: ClassVar[np.dtype] = _AnalogDataLayout._dtype
    primitive_dtype: ClassVar[type] = np.int16
    array_size = 3
    
    @classmethod
    def format(
        cls,
        payload: AnalogDataPayload,
        message_type: HarpMessageType,
        timestamp: float,
    ) -> HarpMessage[AnalogDataPayload]:
        return HarpMessage.from_payload(
            payload,
            message_type=message_type,
            timestamp=timestamp,
            address=cls.address,
        )

    @classmethod
    def create(
        cls,
        analog_input0: np.int16,
        encoder: np.int16,
        analog_input1: np.int16,
        message_type: HarpMessageType,
        timestamp: float,
    ) -> HarpMessage[AnalogDataPayload]:
        """Convenience method to create a message with field values directly."""
        payload = AnalogDataPayload(
            analog_input0=analog_input0,
            encoder=encoder,
            analog_input1=analog_input1,
        )
        return cls.format(payload, message_type, timestamp)

    @classmethod
    def parse(cls, message: HarpMessage) -> AnalogDataPayload:
        return AnalogDataPayload(buffer=message.payload)

    @classmethod
    def create_default(
        cls,
        message_type: HarpMessageType,
        timestamp: float,
    ) -> HarpMessage[AnalogDataPayload]:
        return HarpMessage.from_payload(
            np.zeros((), dtype=cls.dtype),
            message_type,
            timestamp,
            cls.address,
        )

class SpecificDevice(HarpDevice):
    """
    Concrete device implementation with typed register access.
    """
    
    # Cache the register instances
    _registers: ClassVar[Dict[int, IRegister]] = {
        AnalogDataRegister.address: AnalogDataRegister(),
    }

    @property
    def AnalogData(self) -> AnalogDataRegister:
        return self._registers[AnalogDataRegister.address]


# =============================================================================
# Usage Example
# =============================================================================

def example_usage():
    # Connect to device
    my_device = SpecificDevice(com_port="COM3")
    my_device.connect()

    data: AnalogDataPayload = my_device.read(my_device.AnalogData)
    data: AnalogDataPayload = my_device.read(AnalogDataRegister())
    data: HarpMessage = my_device.write(my_device.AnalogData.create(
        analog_input0=123,encoder=456,analog_input1=789, timestamp=0.0, message_type=HarpMessageType.Write)) #returns raw Harp Message
    data: AnalogDataPayload = my_device.write(my_device.AnalogData, AnalogDataPayload(
        analog_input0=123,
        encoder=456,
        analog_input1=789,
    ), seconds=0.0, message_type=HarpMessageType.Write)

    # Access fields with IDE autocompletion
    print(f"Analog Input 0: {data.analog_input0}")
    print(f"Encoder: {data.encoder}")
    print(f"Analog Input 1: {data.analog_input1}")

[bruno-f-cruz]

Another point is that message information from reads and writes, if only the payload is returned, loses HarpMessage data (timestamp, port, etc...) it may be better to always return HarpMessage[TPayload] since everything is available from that interface