Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 56 additions & 6 deletions src/packages/harp-device/src/harp/device/schema/_emit.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,15 @@

from harp.device import core

from ._model import DeviceModel, PayloadMember, PayloadType, Register, Registers, Visibility
from ._model import (
DeviceModel,
MaskValueItem,
PayloadMember,
PayloadType,
Register,
Registers,
Visibility,
)
from ._naming import enum_member_name, field_name


Expand Down Expand Up @@ -294,17 +302,49 @@ def _rename(
return renamed

# -- enums ------------------------------------------------------------
@staticmethod
def _mask_value_doc(value: Any) -> Optional[str]:
"""The per-value ``description``, when the mask value is the object form."""
return value.root.description if isinstance(value.root, MaskValueItem) else None

@staticmethod
def _document_enum(
enum_cls: Any, description: Optional[str], docs: Mapping[str, Optional[str]]
) -> None:
"""Set the enum class ``__doc__`` and every member's, ``None`` included.

A member is always assigned, even when it has no description: an ``Enum``
member with no ``__doc__`` of its own falls back to the *class* docstring via
attribute lookup, which would misrepresent it as carrying the mask's own
description.
"""
enum_cls.__doc__ = description
for member_name, doc in docs.items():
getattr(enum_cls, member_name).__doc__ = doc

def _build_enums(self) -> dict[str, Any]:
# Enum type names stay verbatim; members take the SCREAMING_SNAKE of the generator.
enums: dict[str, Any] = {}
for name, spec in self.bit_masks.items():
# IntFlag has no zero-valued member; drop it if present.
bits = {k: v for k, v in spec.bits.items() if int(v) != 0}
renamed = self._rename("bit", name, bits, enum_member_name)
enums[name] = enum.IntFlag(name, {renamed[k]: int(v) for k, v in bits.items()})
enum_cls = enum.IntFlag(name, {renamed[k]: int(v) for k, v in bits.items()})
self._document_enum(
enum_cls,
spec.description,
{renamed[k]: self._mask_value_doc(v) for k, v in bits.items()},
)
enums[name] = enum_cls
for name, spec in self.group_masks.items():
renamed = self._rename("value", name, spec.values, enum_member_name)
enums[name] = enum.IntEnum(name, {renamed[k]: int(v) for k, v in spec.values.items()})
enum_cls = enum.IntEnum(name, {renamed[k]: int(v) for k, v in spec.values.items()})
self._document_enum(
enum_cls,
spec.description,
{renamed[k]: self._mask_value_doc(v) for k, v in spec.values.items()},
)
enums[name] = enum_cls
return enums

# -- converter resolution (one uniform factory pipeline) -------------
Expand Down Expand Up @@ -388,12 +428,16 @@ def _build_field(self, key: str, member: PayloadMember, reg: Register) -> Any:
if group_mask is not None and issubclass(group_mask, enum.IntEnum):
full = (1 << (elem_size * 8)) - 1
mask = member.mask if member.mask is not None else full
return GroupMask(enum=group_mask, mask=mask, offset=offset, **default_kwarg)
field = GroupMask(enum=group_mask, mask=mask, offset=offset, **default_kwarg)
field.__doc__ = member.description
return field

field_kwargs: dict[str, Any] = {"offset": offset, **default_kwarg}
if member.mask is not None:
field_kwargs["mask"] = member.mask
return Field(self._resolve_converter(ctx), **field_kwargs)
field = Field(self._resolve_converter(ctx), **field_kwargs)
field.__doc__ = member.description
return field

# -- payloads ---------------------------------------------------------
def _payload_name(self, name: str, reg: Register) -> str:
Expand Down Expand Up @@ -477,8 +521,13 @@ def _build_register(self, name: str, class_name: str, reg: Register) -> type[Reg
if reg.length is not None: # plain array register
cls = _ARRAY_REGISTER[reg.type](reg.address, length=reg.length)
cls.__name__ = cls.__qualname__ = class_name
cls.__doc__ = reg.description
return cls
return _new_class(class_name, (_SCALAR_REGISTER[reg.type],), {"address": reg.address})
return _new_class(
class_name,
(_SCALAR_REGISTER[reg.type],),
{"address": reg.address, "__doc__": reg.description},
)

payload_cls = self._build_payload(name, reg)
return _new_class(
Expand All @@ -488,6 +537,7 @@ def _build_register(self, name: str, class_name: str, reg: Register) -> type[Reg
"address": reg.address,
"payload_type": ProtoPayloadType[reg.type.name],
"payload_class": payload_cls,
"__doc__": reg.description,
},
)

Expand Down
133 changes: 132 additions & 1 deletion tests/device/test_emit.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
import numpy as np
import pytest
from harp.data import parse_to_dataframe
from harp.protocol import GroupMask, HarpMessage
from harp.protocol import Field, GroupMask, HarpMessage

from harp.device import core
from harp.device.schema._emit import (
Expand Down Expand Up @@ -595,3 +595,134 @@ def test_emitted_register_bulk_matches_oracle(name, device_registers):
df_oracle = parse_to_dataframe(oracle, buf, time_index=False)
assert list(df_emitted.columns) == list(df_oracle.columns)
assert df_emitted.equals(df_oracle)


# ---------------------------------------------------------------------------
# Schema descriptions surfacing as __doc__ on the emitted runtime objects
# ---------------------------------------------------------------------------

_DESCRIBED_YML = (
"registers:\n"
" Scalar:\n"
" address: 40\n"
" type: U16\n"
" access: Read\n"
" description: A scalar register.\n"
" Array:\n"
" address: 41\n"
" type: U16\n"
" length: 2\n"
" access: Read\n"
" description: An array register.\n"
" Undescribed:\n"
" address: 42\n"
" type: U16\n"
" access: Read\n"
" StartPulse:\n"
" address: 100\n"
" type: U16\n"
" access: Write\n"
" description: Starts a PWM pulse.\n"
" payloadSpec:\n"
" DigitalOutput:\n"
" offset: 0\n"
" mask: 0xC00\n"
" maskType: PwmPort\n"
" description: Which output channel drives the pulse.\n"
" PulseWidth:\n"
" offset: 0\n"
" mask: 0x3FF\n"
" interfaceType: ushort\n"
" Indicators:\n"
" address: 43\n"
" type: U8\n"
" access: Write\n"
" maskType: PortDigitalIOS\n"
"groupMasks:\n"
" PwmPort:\n"
" description: Which PWM channel is selected.\n"
" values:\n"
" Pwm0:\n"
" value: 0x1\n"
" description: PWM channel 0.\n"
" Pwm1: 0x2\n"
"bitMasks:\n"
" PortDigitalIOS:\n"
" description: Bitmask of digital IO ports.\n"
" bits:\n"
" DIO0:\n"
" value: 0x1\n"
" description: Digital output 0.\n"
" DIO1: 0x2\n"
)


@pytest.fixture
def described_registers():
return create_registers(_DESCRIBED_YML)


def test_scalar_register_description_becomes_class_doc(described_registers):
assert described_registers["Scalar"].__doc__ == "A scalar register."


def test_array_register_description_becomes_class_doc(described_registers):
assert described_registers["Array"].__doc__ == "An array register."


def test_payload_wrapped_register_description_becomes_class_doc(described_registers):
assert described_registers["StartPulse"].__doc__ == "Starts a PWM pulse."


def test_absent_register_description_leaves_doc_none(described_registers):
assert described_registers["Undescribed"].__doc__ is None


def test_payload_member_description_becomes_field_doc(described_registers):
# DigitalOutput has a maskType, so it is emitted as a GroupMask descriptor.
payload_cls = described_registers["StartPulse"].payload_class
field = payload_cls.__dict__["digital_output"]
assert field.__doc__ == "Which output channel drives the pulse."


def test_undescribed_payload_member_does_not_inherit_descriptor_class_doc(described_registers):
# Regression: an unset instance __doc__ falls back to the owning class's, which
# would misrepresent PulseWidth (no description) as carrying Field's own generic
# docstring.
payload_cls = described_registers["StartPulse"].payload_class
field = payload_cls.__dict__["pulse_width"]
assert field.__doc__ is None
assert field.__doc__ != Field.__doc__


def test_group_mask_type_description_becomes_enum_doc(described_registers):
enum_cls = _enum_of(described_registers["StartPulse"], "digital_output")
assert enum_cls.__doc__ == "Which PWM channel is selected."


def test_bit_mask_type_description_becomes_enum_doc(described_registers):
enum_cls = _enum_of(described_registers["Indicators"], "__value__")
assert enum_cls.__doc__ == "Bitmask of digital IO ports."


def test_bit_mask_value_description_becomes_member_doc(described_registers):
enum_cls = _enum_of(described_registers["Indicators"], "__value__")
assert enum_cls.DIO0.__doc__ == "Digital output 0."


def test_group_mask_value_description_becomes_member_doc(described_registers):
enum_cls = _enum_of(described_registers["StartPulse"], "digital_output")
assert enum_cls.PWM0.__doc__ == "PWM channel 0."


def test_undescribed_bit_mask_value_does_not_inherit_enum_type_doc(described_registers):
# Regression: Enum attribute lookup on a member with no __doc__ of its own falls
# back to the class __doc__, which would misrepresent DIO1 as carrying the
# bitmask's own description once PortDigitalIOS.__doc__ is set.
enum_cls = _enum_of(described_registers["Indicators"], "__value__")
assert enum_cls.DIO1.__doc__ is None


def test_undescribed_group_mask_value_does_not_inherit_enum_type_doc(described_registers):
enum_cls = _enum_of(described_registers["StartPulse"], "digital_output")
assert enum_cls.PWM1.__doc__ is None