moddef package¶

ModDef runtime for Python (spec v0.4): parse declarative Modbus device definitions, decode/encode points exactly, and drive devices asynchronously.

Quick start:

import asyncio
from moddef import Device, MeasurandQuery, load
from moddef.pymodbus import Options, PymodbusTransport   # extra: moddef[pymodbus]

async def main():
    doc = load("growatt-sph.moddef.yaml")
    transport = await PymodbusTransport.tcp("192.168.1.50", options=Options())
    dev = Device.create(doc, None, transport)
    print(await dev.read_point("state_of_charge"))
    print(await dev.read_measurand(MeasurandQuery("frequency")))

asyncio.run(main())

exception moddef.AmbiguousMeasurandError(query: object, matches: list[str])[source]¶

Bases: ModDefError

More than one point matches the measurand query (spec §26.4).

exception moddef.DecodeError(point_id: str, message: str)[source]¶: Bases: ModDefError

class moddef.Device(profile: DeviceProfile, transport: Transport)[source]¶

Bases: object

The untyped runtime facade over one device profile.

classmethod create(doc: ModDefDocument, device_id: str | None, transport: Transport) → Device[source]¶: Bind a transport to the named device profile in doc (or the only one).

point(point_id: str) → tuple[Point, RegisterBlock][source]¶

points() → list[Point][source]¶

async read_measurand(q: MeasurandQuery) → float | int | bool | str | bytes | list[str] | dict[str, int] | datetime | Unavailable[source]¶: Read a point by its semantic measurand tuple (spec §26.1).

async read_point(point_id: str) → float | int | bool | str | bytes | list[str] | dict[str, int] | datetime | Unavailable[source]¶: Read and decode a single point by id.

async read_points(ids: Iterable[str]) → dict[str, float | int | bool | str | bytes | list[str] | dict[str, int] | datetime | Unavailable][source]¶

async write_point(point_id: str, value: float | int | bool | str | bytes | Sequence[str] | datetime) → None[source]¶: Encode and write a value, validating access mode and §11.4 constraints.

exception moddef.DeviceNotFoundError(device_id: str)[source]¶: Bases: ModDefError

class moddef.DirResolver(roots: list[str] | None = None)[source]¶

Bases: object

Resolver over MODDEF_PACKAGE_ROOTS-style directories.

fetch(uri: str) → tuple[bytes, Literal['yaml', 'json', 'binary']][source]¶

exception moddef.EncodeError(point_id: str, message: str)[source]¶: Bases: ModDefError

exception moddef.MeasurandNotSupportedError(query: object)[source]¶: Bases: ModDefError

Bases: object

accumulation: int | None = None¶

aggregation: int | None = None¶

base_quantity: str¶

direction: int | None = None¶

location: int | None = None¶

phase_ref: int | None = None¶

exception moddef.ModDefError[source]¶

Bases: Exception

Base class for all moddef errors.

exception moddef.ParseError[source]¶

Bases: ModDefError

Document failed to parse (bad format, unknown field, bad enum value).

exception moddef.PointNotFoundError(point_id: str)[source]¶: Bases: ModDefError

class moddef.ResolvedDocument(doc: ModDefDocument, enums: dict[str, ~moddef.v1.types_pb2.EnumType]=<factory>, measurands: dict[str, ~moddef.v1.measurand_pb2.MeasurandDefinition]=<factory>, imports: dict[str, ~moddef.v1.document_pb2.ModDefDocument]=<factory>)[source]¶

Bases: object

A document’s visible symbol tables after import resolution: local symbols first, then imported ones (alias-prefixed as <alias>:<id>, never overriding an existing entry).

doc: ModDefDocument¶

enums: dict[str, EnumType]¶

imports: dict[str, ModDefDocument]¶

measurands: dict[str, MeasurandDefinition]¶

class moddef.Transport(*args, **kwargs)[source]¶

Bases: Protocol

async read_coils(offset: int, count: int) → Sequence[bool][source]¶

async read_discrete(offset: int, count: int) → Sequence[bool][source]¶

async read_holding(offset: int, count: int) → Sequence[int][source]¶

async read_input(offset: int, count: int) → Sequence[int][source]¶

async write_coil(offset: int, on: bool) → None[source]¶

async write_holding(offset: int, words: Sequence[int]) → None[source]¶

class moddef.Unavailable(meaning: str = '')[source]¶

Bases: object

The device reported “no data” for this point (spec §8.4).

meaning: str = ''¶

exception moddef.UnsupportedMappingError(subject: str, message: str)[source]¶: Bases: ModDefError

exception moddef.WriteAccessError(point_id: str, access: str)[source]¶: Bases: ModDefError

exception moddef.WriteConstraintError(point_id: str, constraint: str, value: object, message: str)[source]¶

Bases: ModDefError

§11.4 constraint violation; constraint names the violated rule.

moddef.decode_all(points: Iterable[Point], regs_by_id: Mapping[str, Sequence[int]]) → dict[str, float | int | bool | str | bytes | list[str] | dict[str, int] | datetime | Unavailable][source]¶: Decode several points, resolving refs among them first.

moddef.decode_point(p: Point, regs: Sequence[int], ctx: Mapping[str, int] | None = None) → float | int | bool | str | bytes | list[str] | dict[str, int] | datetime | Unavailable[source]¶: Decode the registers for a point into a typed value.

moddef.decode_point_raw(p: Point, regs: Sequence[int]) → tuple[int, int][source]¶: Pre-scale integer view for callers that need exactness: (raw, bits).

moddef.detect_format(path: str | PathLike[str]) → Literal['yaml', 'json', 'binary'][source]¶: Infer the format from a file path per the spec §4 extensions.

moddef.encode_point(p: Point, value: float | int | bool | str | bytes | Sequence[str] | datetime, ctx: Mapping[str, int] | None = None) → list[int][source]¶: Serialize a typed value into register words per the point’s mapping.

moddef.load(path: str | PathLike[str]) → ModDefDocument[source]¶: Load and parse a .moddef.yaml / .moddef.json / .moddef file.

moddef.measurand_matches(m: MeasurandRef | None, q: MeasurandQuery) → bool[source]¶

moddef.parse_document(data: bytes | str, format: Literal['yaml', 'json', 'binary']) → ModDefDocument[source]¶: Parse a document from text or bytes in the given format.

moddef.point_words(p: Point) → int[source]¶: Register count to read/write for p (mapping length or storage width).

moddef.resolve_context(points: Iterable[Point], regs_by_id: Mapping[str, Sequence[int]]) → dict[str, int][source]¶: Decode every point that is a scale_ref/selector_ref target into ints.

moddef.resolve_imports(doc: ModDefDocument, resolver: PackageResolver | None = None) → ResolvedDocument[source]¶: Resolve a document’s imports and build the visible symbol tables.

moddef.save(doc: ModDefDocument, path: str | PathLike[str]) → None[source]¶: Serialize and write a document; the format comes from the extension.

moddef.serialize_document(doc: ModDefDocument, format: Literal['yaml', 'json', 'binary']) → bytes | str[source]¶

Serialize a document. yaml/json return str, binary returns bytes.

Binary serialization is deterministic (ascending map keys), matching the Go-produced golden .moddef files byte for byte. It goes through canonical_bytes rather than SerializeToString(deterministic=True) because the protobuf C/upb backend orders map entries by descending key, which would not match the canonical form.

moddef.validate_constraints(p: Point, value: float | int | bool | str | bytes | Sequence[str] | datetime) → None[source]¶: Validate a write value against WriteConstraints (spec §11.4).

moddef package¶

Subpackages¶

Submodules¶