API Reference

Core Classes

OCMF

The main class for parsing, validating, and verifying OCMF data.

OCMF

Bases: BaseModel

OCMF data model with three pipe-separated sections: header, payload, and signature.

A public key appended as a fourth section is kept as embedded_public_key.

Source code in src/pyocmf/core/ocmf.py

class OCMF(pydantic.BaseModel):
    """OCMF data model with three pipe-separated sections: header, payload, and signature.

    A public key appended as a fourth section is kept as ``embedded_public_key``.
    """

    header: Literal["OCMF"]
    payload: Payload
    signature: Signature
    _original_payload_json: str | None = pydantic.PrivateAttr(default=None)
    _embedded_public_key: str | None = pydantic.PrivateAttr(default=None)

    @classmethod
    def from_string(cls, ocmf_string: str, *, strict: bool = False) -> OCMF:
        """Parse an OCMF string into an OCMF model.

        Automatically detects whether the input is plain text (starts with "OCMF|")
        or hex-encoded and handles both formats.

        By default, input that deviates from the OCMF spec but is accepted by the
        Transparenzsoftware is parsed and reported as a SpecWarning.

        Args:
            ocmf_string: OCMF string, plain or hex-encoded
            strict: Reject spec deviations instead of warning

        Raises:
            OcmfFormatError: If the string is not structured as OCMF
            OcmfPayloadError: If the payload is invalid (or deviates from the spec
                in strict mode)
            OcmfSignatureError: If the signature section is invalid (or deviates from
                the spec in strict mode)

        """
        with spec_mode(strict=strict):
            return cls._parse(ocmf_string)

    @classmethod
    def _parse(cls, ocmf_string: str) -> OCMF:
        ocmf_text = ocmf_string.strip()

        if not ocmf_text.startswith(OCMF_PREFIX):
            try:
                decoded_bytes = bytes.fromhex(ocmf_text)
                ocmf_text = decoded_bytes.decode("utf-8")
            except ValueError as e:
                msg = (
                    f"Invalid OCMF string: must start with '{OCMF_PREFIX}' or be "
                    f"valid hex-encoded. {e}"
                )
                raise HexDecodingError(msg) from e
        parts = ocmf_text.split(OCMF_SEPARATOR, 3)

        if len(parts) < 3 or parts[0] != OCMF_HEADER:
            msg = (
                f"String does not match expected OCMF format "
                f"'{OCMF_HEADER}{OCMF_SEPARATOR}{{payload}}{OCMF_SEPARATOR}{{signature}}'."
            )
            raise OcmfFormatError(msg)

        payload_json = parts[1]
        signature_json = parts[2]
        embedded_public_key = None
        if len(parts) == 4:
            # The Transparenzsoftware (and QR codes for it) append the public key as a
            # fourth section
            try:
                warn_spec(
                    "Public keys must be transmitted separately, not as a fourth OCMF section"
                )
            except SpecViolationError as e:
                raise OcmfFormatError(str(e)) from e
            embedded_public_key = "".join(parts[3].split()) or None

        # parse_float=Decimal builds Decimals from the raw JSON literals, preserving
        # decimal places (e.g. 2935.600) that pydantic's own JSON parser would drop
        try:
            payload = Payload.model_validate(json.loads(payload_json, parse_float=decimal.Decimal))
        except (json.JSONDecodeError, pydantic.ValidationError) as e:
            violation = _spec_violation(e)
            msg = (
                f"Payload deviates from the OCMF spec: {violation}"
                if violation
                else f"Invalid payload JSON: {e}"
            )
            raise OcmfPayloadError(msg) from e

        try:
            signature = Signature.model_validate_json(signature_json)
        except pydantic.ValidationError as e:
            violation = _spec_violation(e)
            msg = (
                f"Signature deviates from the OCMF spec: {violation}"
                if violation
                else f"Invalid signature JSON: {e}"
            )
            raise OcmfSignatureError(msg) from e

        # Payload and signature are validated above; constructing the model normally would
        # re-run the payload's validators (and report every spec deviation twice)
        ocmf = cls.model_construct(header=OCMF_HEADER, payload=payload, signature=signature)
        ocmf._original_payload_json = payload_json
        ocmf._embedded_public_key = embedded_public_key
        return ocmf

    def to_string(self, hex: bool = False) -> str:
        """Convert the OCMF model to string format "OCMF|{payload}|{signature}".

        Set hex=True to return hex-encoded string instead of plain text.
        """
        payload_json = model_to_ocmf_json(self.payload)
        signature_json = model_to_ocmf_json(self.signature)
        ocmf_string = OCMF_SEPARATOR.join([OCMF_HEADER, payload_json, signature_json])

        if hex:
            return ocmf_string.encode("utf-8").hex()
        return ocmf_string

    @property
    def embedded_public_key(self) -> str | None:
        """Public key appended to the parsed string as a fourth section, if any.

        The OCMF spec requires keys to be transmitted separately, but the
        Transparenzsoftware accepts them appended (``OCMF|payload|signature|key``),
        for example in QR codes. Parsing such a string emits a SpecWarning.
        """
        return self._embedded_public_key

    def verify_signature(self, public_key: PublicKey | str | None = None) -> bool:
        """Verify the cryptographic signature of the OCMF data.

        Per OCMF spec, public keys must be transmitted out-of-band (separately from OCMF data).
        Without ``public_key``, the key embedded in the parsed string is used.
        Requires that the OCMF was parsed from a string (not constructed programmatically)
        because signature verification needs the exact original payload bytes.

        Args:
            public_key: Public key as a PublicKey, or a hex or base64 DER string

        Returns:
            Whether the signature matches the payload

        Raises:
            SignatureVerificationError: If no public key is given or embedded, or the
                signature cannot be checked

        """
        key = public_key if public_key is not None else self._embedded_public_key
        if key is None:
            msg = "No public key given and none embedded in the OCMF string"
            raise SignatureVerificationError(msg)
        if self._original_payload_json is None:
            msg = (
                "Cannot verify signature: original payload JSON not available. "
                "Signature verification requires the exact original payload bytes. "
                "Use OCMF.from_string() to parse OCMF data for signature verification."
            )
            raise SignatureVerificationError(msg)

        return verification.verify_signature(
            payload_json=self._original_payload_json,
            signature_data=self.signature.SD,
            signature_method=self.signature.SA,
            signature_encoding=self.signature.SE,
            public_key=key.key if isinstance(key, PublicKey) else key,
        )

    def check_eichrecht(
        self, other: OCMF | None = None, *, errors_only: bool = False
    ) -> list[EichrechtIssue]:
        """Check German calibration law (Eichrecht) compliance.

        Applies the rules of the Transparenzsoftware to the law-relevant readings,
        preferring loss-compensated registers. Errors are what the Transparenzsoftware
        rejects; spec rules it does not enforce are warnings.

        Without ``other``, this record is checked on its own, as a complete transaction
        if it holds both the begin and the end reading. With ``other``, this record is
        the begin and ``other`` the end of a transaction.
        Set errors_only=True to filter out warnings.
        """
        if other is None:
            issues = compliance.check_eichrecht_payload(self.payload)
        else:
            issues = compliance.check_eichrecht_transaction(self.payload, other.payload)

        if errors_only:
            issues = [issue for issue in issues if issue.severity == IssueSeverity.ERROR]

        return issues

    @property
    def is_eichrecht_compliant(self) -> bool:
        """Whether this record on its own passes the Eichrecht checks; warnings are ignored."""
        issues = self.check_eichrecht(errors_only=True)
        return len(issues) == 0

    def verify(
        self,
        public_key: PublicKey | str | None = None,
        other: OCMF | None = None,
        eichrecht: bool = True,
    ) -> tuple[bool, list[EichrechtIssue]]:
        """Verify both cryptographic signature and legal compliance.

        Combines signature verification and Eichrecht compliance checking.
        Without ``public_key``, the key embedded in the parsed string is used.
        Returns (signature_valid, compliance_issues).
        Set eichrecht=False to skip compliance checking.
        """
        signature_valid = self.verify_signature(public_key)
        compliance_issues = self.check_eichrecht(other) if eichrecht else []
        return signature_valid, compliance_issues

embedded_public_key: str | None property

Public key appended to the parsed string as a fourth section, if any.

The OCMF spec requires keys to be transmitted separately, but the Transparenzsoftware accepts them appended (OCMF|payload|signature|key), for example in QR codes. Parsing such a string emits a SpecWarning.

is_eichrecht_compliant: bool property

Whether this record on its own passes the Eichrecht checks; warnings are ignored.

from_string(ocmf_string: str, *, strict: bool = False) -> OCMF classmethod

Parse an OCMF string into an OCMF model.

Automatically detects whether the input is plain text (starts with "OCMF|") or hex-encoded and handles both formats.

By default, input that deviates from the OCMF spec but is accepted by the Transparenzsoftware is parsed and reported as a SpecWarning.

Parameters:

Name	Type	Description	Default
ocmf_string	str	OCMF string, plain or hex-encoded	required
strict	bool	Reject spec deviations instead of warning	False

Raises:

Type	Description
OcmfFormatError	If the string is not structured as OCMF
OcmfPayloadError	If the payload is invalid (or deviates from the spec in strict mode)
OcmfSignatureError	If the signature section is invalid (or deviates from the spec in strict mode)

Source code in src/pyocmf/core/ocmf.py

@classmethod
def from_string(cls, ocmf_string: str, *, strict: bool = False) -> OCMF:
    """Parse an OCMF string into an OCMF model.

    Automatically detects whether the input is plain text (starts with "OCMF|")
    or hex-encoded and handles both formats.

    By default, input that deviates from the OCMF spec but is accepted by the
    Transparenzsoftware is parsed and reported as a SpecWarning.

    Args:
        ocmf_string: OCMF string, plain or hex-encoded
        strict: Reject spec deviations instead of warning

    Raises:
        OcmfFormatError: If the string is not structured as OCMF
        OcmfPayloadError: If the payload is invalid (or deviates from the spec
            in strict mode)
        OcmfSignatureError: If the signature section is invalid (or deviates from
            the spec in strict mode)

    """
    with spec_mode(strict=strict):
        return cls._parse(ocmf_string)

to_string(hex: bool = False) -> str

Convert the OCMF model to string format "OCMF|{payload}|{signature}".

Set hex=True to return hex-encoded string instead of plain text.

Source code in src/pyocmf/core/ocmf.py

def to_string(self, hex: bool = False) -> str:
    """Convert the OCMF model to string format "OCMF|{payload}|{signature}".

    Set hex=True to return hex-encoded string instead of plain text.
    """
    payload_json = model_to_ocmf_json(self.payload)
    signature_json = model_to_ocmf_json(self.signature)
    ocmf_string = OCMF_SEPARATOR.join([OCMF_HEADER, payload_json, signature_json])

    if hex:
        return ocmf_string.encode("utf-8").hex()
    return ocmf_string

verify_signature(public_key: PublicKey | str | None = None) -> bool

Verify the cryptographic signature of the OCMF data.

Per OCMF spec, public keys must be transmitted out-of-band (separately from OCMF data). Without public_key, the key embedded in the parsed string is used. Requires that the OCMF was parsed from a string (not constructed programmatically) because signature verification needs the exact original payload bytes.

Parameters:

Name	Type	Description	Default
public_key	PublicKey | str | None	Public key as a PublicKey, or a hex or base64 DER string	None

Returns:

Type	Description
bool	Whether the signature matches the payload

Raises:

Type	Description
SignatureVerificationError	If no public key is given or embedded, or the signature cannot be checked

Source code in src/pyocmf/core/ocmf.py

def verify_signature(self, public_key: PublicKey | str | None = None) -> bool:
    """Verify the cryptographic signature of the OCMF data.

    Per OCMF spec, public keys must be transmitted out-of-band (separately from OCMF data).
    Without ``public_key``, the key embedded in the parsed string is used.
    Requires that the OCMF was parsed from a string (not constructed programmatically)
    because signature verification needs the exact original payload bytes.

    Args:
        public_key: Public key as a PublicKey, or a hex or base64 DER string

    Returns:
        Whether the signature matches the payload

    Raises:
        SignatureVerificationError: If no public key is given or embedded, or the
            signature cannot be checked

    """
    key = public_key if public_key is not None else self._embedded_public_key
    if key is None:
        msg = "No public key given and none embedded in the OCMF string"
        raise SignatureVerificationError(msg)
    if self._original_payload_json is None:
        msg = (
            "Cannot verify signature: original payload JSON not available. "
            "Signature verification requires the exact original payload bytes. "
            "Use OCMF.from_string() to parse OCMF data for signature verification."
        )
        raise SignatureVerificationError(msg)

    return verification.verify_signature(
        payload_json=self._original_payload_json,
        signature_data=self.signature.SD,
        signature_method=self.signature.SA,
        signature_encoding=self.signature.SE,
        public_key=key.key if isinstance(key, PublicKey) else key,
    )

check_eichrecht(other: OCMF | None = None, *, errors_only: bool = False) -> list[EichrechtIssue]

Check German calibration law (Eichrecht) compliance.

Applies the rules of the Transparenzsoftware to the law-relevant readings, preferring loss-compensated registers. Errors are what the Transparenzsoftware rejects; spec rules it does not enforce are warnings.

Without other, this record is checked on its own, as a complete transaction if it holds both the begin and the end reading. With other, this record is the begin and other the end of a transaction. Set errors_only=True to filter out warnings.

Source code in src/pyocmf/core/ocmf.py

def check_eichrecht(
    self, other: OCMF | None = None, *, errors_only: bool = False
) -> list[EichrechtIssue]:
    """Check German calibration law (Eichrecht) compliance.

    Applies the rules of the Transparenzsoftware to the law-relevant readings,
    preferring loss-compensated registers. Errors are what the Transparenzsoftware
    rejects; spec rules it does not enforce are warnings.

    Without ``other``, this record is checked on its own, as a complete transaction
    if it holds both the begin and the end reading. With ``other``, this record is
    the begin and ``other`` the end of a transaction.
    Set errors_only=True to filter out warnings.
    """
    if other is None:
        issues = compliance.check_eichrecht_payload(self.payload)
    else:
        issues = compliance.check_eichrecht_transaction(self.payload, other.payload)

    if errors_only:
        issues = [issue for issue in issues if issue.severity == IssueSeverity.ERROR]

    return issues

verify(public_key: PublicKey | str | None = None, other: OCMF | None = None, eichrecht: bool = True) -> tuple[bool, list[EichrechtIssue]]

Verify both cryptographic signature and legal compliance.

Combines signature verification and Eichrecht compliance checking. Without public_key, the key embedded in the parsed string is used. Returns (signature_valid, compliance_issues). Set eichrecht=False to skip compliance checking.

Source code in src/pyocmf/core/ocmf.py

def verify(
    self,
    public_key: PublicKey | str | None = None,
    other: OCMF | None = None,
    eichrecht: bool = True,
) -> tuple[bool, list[EichrechtIssue]]:
    """Verify both cryptographic signature and legal compliance.

    Combines signature verification and Eichrecht compliance checking.
    Without ``public_key``, the key embedded in the parsed string is used.
    Returns (signature_valid, compliance_issues).
    Set eichrecht=False to skip compliance checking.
    """
    signature_valid = self.verify_signature(public_key)
    compliance_issues = self.check_eichrecht(other) if eichrecht else []
    return signature_valid, compliance_issues

Payload

The payload section containing meter readings and metadata.

Payload

Bases: BaseModel

Payload section of an OCMF record: gateway, meter, user assignment and readings.

Reading fields left out after the first reading are inherited from the previous reading, as the OCMF spec allows.

Source code in src/pyocmf/core/payload.py

class Payload(pydantic.BaseModel):
    """Payload section of an OCMF record: gateway, meter, user assignment and readings.

    Reading fields left out after the first reading are inherited from the previous
    reading, as the OCMF spec allows.
    """

    model_config = pydantic.ConfigDict(extra="allow")

    FV: str | None = pydantic.Field(default=None, description="Format Version")
    GI: str | None = pydantic.Field(default=None, description="Gateway Identification")
    GS: str | None = pydantic.Field(default=None, description="Gateway Serial")
    GV: str | None = pydantic.Field(default=None, description="Gateway Version")

    PG: str | None = pydantic.Field(default=None, description="Pagination")

    MV: str | None = pydantic.Field(default=None, description="Meter Vendor")
    MM: str | None = pydantic.Field(default=None, description="Meter Model")
    MS: str | None = pydantic.Field(default=None, description="Meter Serial")
    MF: str | None = pydantic.Field(default=None, description="Meter Firmware")

    IS: LenientBool | None = pydantic.Field(default=None, description="Identification Status")
    IL: LenientUserAssignmentStatus | None = pydantic.Field(
        default=None, description="Identification Level"
    )
    IF: list[LenientIdentificationFlag] = pydantic.Field(
        default=[], description="Identification Flags (0..4 per OCMF spec Table 4)"
    )
    IT: LenientIdentificationType | None = pydantic.Field(
        default=IdentificationType.NONE, description="Identification Type"
    )
    ID: IdentificationData | None = pydantic.Field(default=None, description="Identification Data")
    TT: str | None = pydantic.Field(default=None, description="Tariff Text (0..250)")

    CF: str | None = pydantic.Field(
        default=None, description="Charge Controller Firmware Version (0..25)"
    )
    LC: CableLossCompensation | None = pydantic.Field(default=None, description="Loss Compensation")

    CT: LenientChargePointIdentificationType | None = pydantic.Field(
        default=None, description="Charge Point Identification Type"
    )
    CI: str | None = pydantic.Field(default=None, description="Charge Point Identification")

    RD: list[Reading] = pydantic.Field(description="Readings")

    @pydantic.model_validator(mode="before")
    @classmethod
    def apply_reading_inheritance(cls, data: dict[str, Any]) -> dict[str, Any]:
        """Apply field inheritance for readings.

        Per OCMF spec, some reading fields can be inherited from the previous reading
        if not specified.
        """
        if not isinstance(data, dict):
            return data

        readings_data = data.get("RD", [])
        if not readings_data:
            return data

        if isinstance(readings_data[0], Reading):
            return data

        inheritable_fields = ["TM", "TX", "RI", "RU", "RT", "EF", "ST"]
        last_values: dict[str, str] = {}
        processed_readings = []

        for rd in readings_data:
            reading_dict = {
                field: rd.get(field, last_values.get(field))
                for field in inheritable_fields
                if field in rd or field in last_values
            }
            reading_dict.update({k: v for k, v in rd.items() if k not in inheritable_fields})

            last_values.update({k: v for k, v in reading_dict.items() if k in inheritable_fields})

            processed_readings.append(reading_dict)

        return {**data, "RD": processed_readings}

    @pydantic.model_validator(mode="after")
    def warn_on_spec_deviations(self) -> Payload:
        """Warn about OCMF spec violations the Transparenzsoftware tolerates."""
        if self.PG is None or not _PAGINATION_PATTERN.match(self.PG):
            warn_spec(f"Pagination (PG) '{self.PG}' must be 'T' or 'F' followed by a number")
        if self.IS is None:
            warn_spec("Identification Status (IS) is mandatory")
        # GS is optional (0..1) and MS mandatory (1..1), but either one identifies the
        # signature component
        if self.GS is None and self.MS is None:
            warn_spec("Either Gateway Serial (GS) or Meter Serial (MS) must be provided")
        if len(self.IF) > _MAX_IDENTIFICATION_FLAGS:
            warn_spec(
                f"At most {_MAX_IDENTIFICATION_FLAGS} identification flags (IF) are allowed, "
                f"got {len(self.IF)}"
            )
        if self.TT is not None and len(self.TT) > _MAX_TARIFF_TEXT_LENGTH:
            warn_spec(f"Tariff Text (TT) exceeds {_MAX_TARIFF_TEXT_LENGTH} characters")
        if self.CF is not None and len(self.CF) > _MAX_CHARGE_CONTROLLER_FIRMWARE_LENGTH:
            warn_spec(
                f"Charge Controller Firmware (CF) exceeds "
                f"{_MAX_CHARGE_CONTROLLER_FIRMWARE_LENGTH} characters"
            )
        return self

    @pydantic.field_validator("FV", mode="before")
    @classmethod
    def convert_fv_to_string(cls, v: int | float | decimal.Decimal | str | None) -> str | None:
        if isinstance(v, (int, float, decimal.Decimal)):
            return str(v)
        return v

    @pydantic.field_validator("CT", mode="before")
    @classmethod
    def convert_ct_empty_to_none(cls, v: str | int | None) -> str | None:
        if v == "" or v == 0:
            return None
        if isinstance(v, int):
            return str(v)
        return v

    @pydantic.model_validator(mode="after")
    def validate_id_format_by_type(self) -> Payload:
        """Warn when the ID does not match the format of its Identification Type (IT).

        Types without a defined format (LOCAL, CENTRAL, KEY_CODE, ...) accept any value.
        """
        if not self.ID or self.IT is None:
            return self

        adapter = _ID_FORMAT_ADAPTERS.get(self.IT)
        if adapter is None:
            return self

        try:
            adapter.validate_python(self.ID)
        except pydantic.ValidationError:
            warn_spec(
                f"ID value '{self.ID}' does not match expected format for identification "
                f"type '{self.IT}'"
            )
        return self

apply_reading_inheritance(data: dict[str, Any]) -> dict[str, Any] classmethod

Apply field inheritance for readings.

Per OCMF spec, some reading fields can be inherited from the previous reading if not specified.

Source code in src/pyocmf/core/payload.py

@pydantic.model_validator(mode="before")
@classmethod
def apply_reading_inheritance(cls, data: dict[str, Any]) -> dict[str, Any]:
    """Apply field inheritance for readings.

    Per OCMF spec, some reading fields can be inherited from the previous reading
    if not specified.
    """
    if not isinstance(data, dict):
        return data

    readings_data = data.get("RD", [])
    if not readings_data:
        return data

    if isinstance(readings_data[0], Reading):
        return data

    inheritable_fields = ["TM", "TX", "RI", "RU", "RT", "EF", "ST"]
    last_values: dict[str, str] = {}
    processed_readings = []

    for rd in readings_data:
        reading_dict = {
            field: rd.get(field, last_values.get(field))
            for field in inheritable_fields
            if field in rd or field in last_values
        }
        reading_dict.update({k: v for k, v in rd.items() if k not in inheritable_fields})

        last_values.update({k: v for k, v in reading_dict.items() if k in inheritable_fields})

        processed_readings.append(reading_dict)

    return {**data, "RD": processed_readings}

warn_on_spec_deviations() -> Payload

Warn about OCMF spec violations the Transparenzsoftware tolerates.

Source code in src/pyocmf/core/payload.py

@pydantic.model_validator(mode="after")
def warn_on_spec_deviations(self) -> Payload:
    """Warn about OCMF spec violations the Transparenzsoftware tolerates."""
    if self.PG is None or not _PAGINATION_PATTERN.match(self.PG):
        warn_spec(f"Pagination (PG) '{self.PG}' must be 'T' or 'F' followed by a number")
    if self.IS is None:
        warn_spec("Identification Status (IS) is mandatory")
    # GS is optional (0..1) and MS mandatory (1..1), but either one identifies the
    # signature component
    if self.GS is None and self.MS is None:
        warn_spec("Either Gateway Serial (GS) or Meter Serial (MS) must be provided")
    if len(self.IF) > _MAX_IDENTIFICATION_FLAGS:
        warn_spec(
            f"At most {_MAX_IDENTIFICATION_FLAGS} identification flags (IF) are allowed, "
            f"got {len(self.IF)}"
        )
    if self.TT is not None and len(self.TT) > _MAX_TARIFF_TEXT_LENGTH:
        warn_spec(f"Tariff Text (TT) exceeds {_MAX_TARIFF_TEXT_LENGTH} characters")
    if self.CF is not None and len(self.CF) > _MAX_CHARGE_CONTROLLER_FIRMWARE_LENGTH:
        warn_spec(
            f"Charge Controller Firmware (CF) exceeds "
            f"{_MAX_CHARGE_CONTROLLER_FIRMWARE_LENGTH} characters"
        )
    return self

validate_id_format_by_type() -> Payload

Warn when the ID does not match the format of its Identification Type (IT).

Types without a defined format (LOCAL, CENTRAL, KEY_CODE, ...) accept any value.

Source code in src/pyocmf/core/payload.py

@pydantic.model_validator(mode="after")
def validate_id_format_by_type(self) -> Payload:
    """Warn when the ID does not match the format of its Identification Type (IT).

    Types without a defined format (LOCAL, CENTRAL, KEY_CODE, ...) accept any value.
    """
    if not self.ID or self.IT is None:
        return self

    adapter = _ID_FORMAT_ADAPTERS.get(self.IT)
    if adapter is None:
        return self

    try:
        adapter.validate_python(self.ID)
    except pydantic.ValidationError:
        warn_spec(
            f"ID value '{self.ID}' does not match expected format for identification "
            f"type '{self.IT}'"
        )
    return self

Reading

Individual meter reading data.

Reading

Bases: BaseModel

A meter reading (OCMF spec Table 7).

Source code in src/pyocmf/core/reading.py

class Reading(pydantic.BaseModel):
    """A meter reading (OCMF spec Table 7)."""

    # Re-validate on assignment so mutated readings keep parsed types (e.g. TM).
    # Extra keys (e.g. EI from pre-1.0 formats) survive parse/serialize roundtrips.
    model_config = pydantic.ConfigDict(validate_assignment=True, extra="allow")

    TM: OCMFTimestamp = pydantic.Field(description="Time (ISO 8601 + time status) - REQUIRED")
    TX: LenientMeterReadingReason | None = pydantic.Field(default=None, description="Transaction")
    RV: OCMFNumber | None = pydantic.Field(
        default=None,
        description="Reading Value - omitted only for pure error-event readings",
    )
    RI: OBISCode | None = pydantic.Field(
        default=None, description="Reading Identification (OBIS code) - Conditional"
    )
    RU: LenientUnit | None = pydantic.Field(
        default=None,
        description="Reading Unit (e.g. kWh, Wh, mOhm per OCMF spec Table 20)",
    )
    RT: LenientReadingType | None = pydantic.Field(
        default=None, description="Reading Current Type (AC/DC per OCMF spec Table 21)"
    )
    CL: OCMFNumber | None = pydantic.Field(default=None, description="Cumulated Losses")
    EF: str | None = pydantic.Field(
        default=None, description="Error Flags (can contain 'E', 't', or both)"
    )
    ST: LenientMeterStatus | None = pydantic.Field(default=None, description="Status - REQUIRED")

    @pydantic.field_validator("TM", mode="before")
    @classmethod
    def parse_timestamp(cls, v: str | OCMFTimestamp) -> OCMFTimestamp:
        if isinstance(v, OCMFTimestamp):
            return v
        if isinstance(v, str):
            return OCMFTimestamp.from_string(v)
        msg = f"TM must be a string or OCMFTimestamp, got {type(v)}"
        raise TypeError(msg)

    @pydantic.field_serializer("TM")
    def serialize_timestamp(self, tm: OCMFTimestamp) -> str:
        return str(tm)

    @pydantic.field_validator("EF", mode="before")
    @classmethod
    def ef_empty_string_to_none(cls, v: str | None) -> str | None:
        if v == "":
            return None
        if v is not None and not _ERROR_FLAGS_PATTERN.match(v):
            warn_spec(f"Error flags (EF) '{v}' contain characters other than 'E' and 't'")
        return v

    @pydantic.model_validator(mode="after")
    def warn_on_missing_fields(self) -> Reading:
        """Warn about OCMF spec Table 7 violations the Transparenzsoftware tolerates.

        RV/RI/RU/RT may all be omitted only when the reading merely signals an
        error event of the meter.
        """
        if self.ST is None:
            warn_spec("Status (ST) is mandatory")
        if (self.RI is None) != (self.RU is None):
            warn_spec(
                "RI (Reading Identification) and RU (Reading Unit) must both be "
                "present or both absent"
            )
        if self.RV is not None and self.RU is None:
            warn_spec("RU (Reading Unit) is required when RV (Reading Value) is present")
        return self

    @property
    def timestamp(self) -> datetime.datetime:
        """Reading time (TM) without its time status."""
        return self.TM.timestamp

    @property
    def time_status(self) -> TimeStatus | str:
        """Time status of TM; an unknown status stays a plain string."""
        return self.TM.status

timestamp: datetime.datetime property

Reading time (TM) without its time status.

time_status: TimeStatus | str property

Time status of TM; an unknown status stays a plain string.

warn_on_missing_fields() -> Reading

Warn about OCMF spec Table 7 violations the Transparenzsoftware tolerates.

RV/RI/RU/RT may all be omitted only when the reading merely signals an error event of the meter.

Source code in src/pyocmf/core/reading.py

@pydantic.model_validator(mode="after")
def warn_on_missing_fields(self) -> Reading:
    """Warn about OCMF spec Table 7 violations the Transparenzsoftware tolerates.

    RV/RI/RU/RT may all be omitted only when the reading merely signals an
    error event of the meter.
    """
    if self.ST is None:
        warn_spec("Status (ST) is mandatory")
    if (self.RI is None) != (self.RU is None):
        warn_spec(
            "RI (Reading Identification) and RU (Reading Unit) must both be "
            "present or both absent"
        )
    if self.RV is not None and self.RU is None:
        warn_spec("RU (Reading Unit) is required when RV (Reading Value) is present")
    return self

Signature

Cryptographic signature data and verification.

Signature

Bases: BaseModel

Signature section of an OCMF record (OCMF spec Table 8).

Source code in src/pyocmf/core/signature.py

class Signature(pydantic.BaseModel):
    """Signature section of an OCMF record (OCMF spec Table 8)."""

    # OCMF spec reserves extension points (keys starting with U-Z and A-F) in the
    # signature section; allow them so they survive parse/serialize roundtrips.
    model_config = pydantic.ConfigDict(extra="allow")

    SA: LenientSignatureMethod | None = pydantic.Field(
        default=SignatureMethod.SECP256R1_SHA256, description="Signature Algorithm"
    )
    SE: LenientSignatureEncoding | None = pydantic.Field(
        default=SignatureEncodingType.HEX, description="Signature Encoding"
    )
    SM: LenientSignatureMimeType | None = pydantic.Field(
        default=SignatureMimeType.APPLICATION_X_DER, description="Signature Mime Type"
    )
    SD: str = pydantic.Field(description="Signature Data (hex or base64)")

    @pydantic.field_validator("SD")
    @classmethod
    def warn_on_undecodable_signature_data(cls, v: str) -> str:
        for validate in (validate_hex_string, validate_base64_string):
            try:
                validate(v)
            except EncodingError:
                continue
            return v
        warn_spec("Signature Data (SD) is neither hex nor base64 encoded")
        return v

Compliance Checking

Eichrecht Compliance

Functions for validating OCMF data against German Eichrecht (calibration law) requirements.

check_eichrecht_reading(reading: Reading) -> list[EichrechtIssue]

Check a single reading for Eichrecht compliance.

Parameters:

Name	Type	Description	Default
reading	Reading	The reading to check	required

Returns:

Type	Description
list[EichrechtIssue]	List of compliance issues (empty if compliant)

Source code in src/pyocmf/compliance/reading.py

def check_eichrecht_reading(reading: Reading) -> list[EichrechtIssue]:
    """Check a single reading for Eichrecht compliance.

    Args:
        reading: The reading to check

    Returns:
        List of compliance issues (empty if compliant)

    """
    issues: list[EichrechtIssue] = []

    if reading.ST != MeterStatus.OK:
        issues.append(
            EichrechtIssue(
                code=IssueCode.METER_STATUS,
                message=(
                    f"Meter status must be 'G' (OK) for billing-relevant readings, "
                    f"got '{reading.ST}'"
                ),
                field="ST",
            )
        )

    if reading.EF and "E" in reading.EF:
        issues.append(
            EichrechtIssue(
                code=IssueCode.ERROR_FLAGS,
                message=f"Energy error flag ('E') set on billing-relevant reading: '{reading.EF}'",
                field="EF",
            )
        )
    # A time error does not invalidate the energy value, matching the Transparenzsoftware
    if reading.EF and "t" in reading.EF:
        issues.append(
            EichrechtIssue(
                code=IssueCode.ERROR_FLAGS,
                message=f"Time error flag ('t') set on reading: '{reading.EF}'",
                field="EF",
                severity=IssueSeverity.WARNING,
            )
        )

    if reading.time_status != TimeStatus.SYNCHRONIZED:
        issues.append(
            EichrechtIssue(
                code=IssueCode.TIME_SYNC,
                message=(
                    f"Time should be synchronized (status 'S') for billing, "
                    f"got '{reading.time_status}'"
                ),
                field="TM",
                severity=IssueSeverity.WARNING,
            )
        )

    issues.extend(_check_cumulated_loss(reading))

    return issues

check_eichrecht_payload(payload: Payload) -> list[EichrechtIssue]

Check a single payload for Eichrecht compliance.

Like the Transparenzsoftware, only law-relevant readings are checked. A payload holding a complete transaction (begin and end reading in the transaction context) is checked like a begin/end pair; otherwise each law-relevant reading is checked on its own.

Source code in src/pyocmf/compliance/transaction.py

def check_eichrecht_payload(payload: Payload) -> list[EichrechtIssue]:
    """Check a single payload for Eichrecht compliance.

    Like the Transparenzsoftware, only law-relevant readings are checked. A payload
    holding a complete transaction (begin and end reading in the transaction context)
    is checked like a begin/end pair; otherwise each law-relevant reading is checked
    on its own.
    """
    if not payload.RD:
        return [
            EichrechtIssue(
                code=IssueCode.NO_READINGS,
                message="No readings (RD) present in payload",
                field="RD",
            )
        ]

    if not _contains_complete_transaction(payload):
        return [
            issue
            for reading in _law_relevant_readings(payload.RD)
            for issue in check_eichrecht_reading(reading)
        ]

    begin, end, issues = _select_transaction_readings(payload.RD)
    for reading in (begin, end):
        if reading is not None:
            issues.extend(check_eichrecht_reading(reading))
    if begin is not None and end is not None:
        issues.extend(_check_transaction_readings(begin, end, payload, payload))
    if issue := _check_identification_level(payload, "transaction"):
        issues.append(issue)
    return issues

check_eichrecht_transaction(begin: Payload, end: Payload) -> list[EichrechtIssue]

Check a complete charging transaction for Eichrecht compliance.

Source code in src/pyocmf/compliance/transaction.py

def check_eichrecht_transaction(
    begin: Payload,
    end: Payload,
) -> list[EichrechtIssue]:
    """Check a complete charging transaction for Eichrecht compliance."""
    if not begin.RD or not end.RD:
        return [
            EichrechtIssue(
                code=IssueCode.NO_READINGS,
                message="Both begin and end payloads must contain readings (RD)",
                field="RD",
            )
        ]

    begin_reading, end_reading, issues = _select_transaction_readings([*begin.RD, *end.RD])

    if begin_reading is not None:
        issues.extend(check_eichrecht_reading(begin_reading))
    if end_reading is not None:
        issues.extend(check_eichrecht_reading(end_reading))
    if begin_reading is not None and end_reading is not None:
        issues.extend(_check_transaction_readings(begin_reading, end_reading, begin, end))

    if issue := _check_field_match(
        begin.GS or begin.MS, end.GS or end.MS, "GS/MS", IssueCode.SERIAL_MISMATCH, "Serial numbers"
    ):
        issues.append(issue)

    for payload, context in ((begin, "begin"), (end, "end")):
        if issue := _check_identification_level(payload, context):
            issues.append(issue)

    if issue := _check_pagination(begin, end):
        issues.append(issue)

    if issue := _check_field_match(
        begin.ID, end.ID, "ID", IssueCode.ID_MISMATCH, "Identification data"
    ):
        issue.severity = IssueSeverity.WARNING
        issues.append(issue)

    return issues

validate_transaction_pair(begin: OCMF, end: OCMF) -> bool

Validate transaction pair compliance (errors only, warnings ignored).

Source code in src/pyocmf/compliance/transaction.py

def validate_transaction_pair(begin: OCMF, end: OCMF) -> bool:
    """Validate transaction pair compliance (errors only, warnings ignored)."""
    issues = check_eichrecht_transaction(begin.payload, end.payload)
    return not any(issue.severity == IssueSeverity.ERROR for issue in issues)

Compliance Models

EichrechtIssue dataclass

A single finding of an Eichrecht compliance check.

Source code in src/pyocmf/compliance/models.py

@dataclass
class EichrechtIssue:
    """A single finding of an Eichrecht compliance check."""

    code: IssueCode
    message: str
    field: str | None = None
    severity: IssueSeverity = IssueSeverity.ERROR

    def __str__(self) -> str:
        prefix = f"[{self.field}] " if self.field else ""
        return f"{prefix}{self.message} ({self.code})"

IssueCode

Bases: StrEnum

Identifies the rule an Eichrecht issue comes from.

Source code in src/pyocmf/compliance/models.py

class IssueCode(enum.StrEnum):
    """Identifies the rule an Eichrecht issue comes from."""

    # Reading-level issues
    METER_STATUS = "METER_STATUS"
    ERROR_FLAGS = "ERROR_FLAGS"
    TIME_SYNC = "TIME_SYNC"
    CL_BEGIN = "CL_BEGIN"
    CL_NEGATIVE = "CL_NEGATIVE"
    CL_REGISTER = "CL_REGISTER"

    # Transaction-level issues
    NO_READINGS = "NO_READINGS"
    BEGIN_TX = "BEGIN_TX"
    END_TX = "END_TX"
    MULTIPLE_BEGIN = "MULTIPLE_BEGIN"
    MULTIPLE_END = "MULTIPLE_END"
    SERIAL_MISMATCH = "SERIAL_MISMATCH"
    OBIS_MISMATCH = "OBIS_MISMATCH"
    UNIT_MISMATCH = "UNIT_MISMATCH"
    VALUE_REGRESSION = "VALUE_REGRESSION"
    TIME_REGRESSION = "TIME_REGRESSION"
    ID_MISMATCH = "ID_MISMATCH"
    ID_LEVEL_INVALID = "ID_LEVEL_INVALID"
    PAGINATION_INCONSISTENT = "PAGINATION_INCONSISTENT"
    EVENT_COUNTER_MISMATCH = "EVENT_COUNTER_MISMATCH"

IssueSeverity

Bases: StrEnum

Severity of an Eichrecht issue: errors fail compliance, warnings do not.

Source code in src/pyocmf/compliance/models.py

class IssueSeverity(enum.StrEnum):
    """Severity of an Eichrecht issue: errors fail compliance, warnings do not."""

    ERROR = "error"
    WARNING = "warning"

Data Models

PublicKey

Public key metadata and validation per OCMF specification.

PublicKey

Bases: BaseModel

An elliptic curve public key in DER encoding, with its curve and size.

Source code in src/pyocmf/models/public_key.py

class PublicKey(pydantic.BaseModel):
    """An elliptic curve public key in DER encoding, with its curve and size."""

    key: HexStr = pydantic.Field(description="Hex-encoded DER public key")
    curve: CurveType = pydantic.Field(description="Elliptic curve type")
    size: int = pydantic.Field(description="Key size in bits")
    block_length: int = pydantic.Field(description="Block length in bytes")

    def to_string(self, base64_encode: bool = False) -> str:
        """Return the DER key as hex, or as base64 with ``base64_encode=True``."""
        if base64_encode:
            key_bytes = bytes.fromhex(self.key)
            return base64.b64encode(key_bytes).decode("ascii")
        return self.key

    @classmethod
    def from_string(cls, key_string: str) -> Self:
        """Parse DER public key string (hex or base64) and extract metadata."""
        check_cryptography_available()

        key_string = key_string.strip()

        # Try hex first (DER-encoded keys typically start with 30 in hex)
        key_bytes: bytes | None = None
        key_hex: str

        try:
            key_bytes = bytes.fromhex(key_string)
            key_hex = key_string
        except ValueError:
            # Not valid hex, try base64
            try:
                key_bytes = base64.b64decode(key_string, validate=True)
                key_hex = key_bytes.hex()
            except Exception as e:
                msg = f"Invalid public key encoding: not valid hex or base64. {e}"
                raise Base64DecodingError(msg) from e

        try:
            public_key = serialization.load_der_public_key(key_bytes)

            if not isinstance(public_key, ec.EllipticCurvePublicKey):
                msg = "Public key is not an elliptic curve key"
                raise TypeError(msg)  # ruff: ignore[raise-within-try]

            curve_name = public_key.curve.name
            key_size = public_key.curve.key_size
            block_length = key_size // 8

            return cls(
                key=key_hex,
                curve=curve_name,
                size=key_size,
                block_length=block_length,
            )
        except UnsupportedAlgorithm as e:
            curve = fallback_curve(key_bytes)
            if curve is None:
                msg = f"Unsupported elliptic curve in public key: {e}"
                raise PublicKeyError(msg) from e
            size = _CURVE_SIZES[curve]
            return cls(key=key_hex, curve=curve, size=size, block_length=size // 8)
        except (ValueError, TypeError) as e:
            msg = f"Failed to parse public key: {e}"
            raise PublicKeyError(msg) from e

    @property
    def key_type_identifier(self) -> KeyType:
        """Key type of the curve, as listed in OCMF spec Table 23."""
        return KeyType.from_curve(self.curve)

    def matches_signature_algorithm(
        self, signature_algorithm: SignatureMethod | str | None
    ) -> bool:
        """Whether the key uses the curve of the signature algorithm (SA)."""
        if not isinstance(signature_algorithm, SignatureMethod):
            return False

        return self.curve == signature_algorithm.curve

key_type_identifier: KeyType property

Key type of the curve, as listed in OCMF spec Table 23.

to_string(base64_encode: bool = False) -> str

Return the DER key as hex, or as base64 with base64_encode=True.

Source code in src/pyocmf/models/public_key.py

def to_string(self, base64_encode: bool = False) -> str:
    """Return the DER key as hex, or as base64 with ``base64_encode=True``."""
    if base64_encode:
        key_bytes = bytes.fromhex(self.key)
        return base64.b64encode(key_bytes).decode("ascii")
    return self.key

from_string(key_string: str) -> Self classmethod

Parse DER public key string (hex or base64) and extract metadata.

Source code in src/pyocmf/models/public_key.py

@classmethod
def from_string(cls, key_string: str) -> Self:
    """Parse DER public key string (hex or base64) and extract metadata."""
    check_cryptography_available()

    key_string = key_string.strip()

    # Try hex first (DER-encoded keys typically start with 30 in hex)
    key_bytes: bytes | None = None
    key_hex: str

    try:
        key_bytes = bytes.fromhex(key_string)
        key_hex = key_string
    except ValueError:
        # Not valid hex, try base64
        try:
            key_bytes = base64.b64decode(key_string, validate=True)
            key_hex = key_bytes.hex()
        except Exception as e:
            msg = f"Invalid public key encoding: not valid hex or base64. {e}"
            raise Base64DecodingError(msg) from e

    try:
        public_key = serialization.load_der_public_key(key_bytes)

        if not isinstance(public_key, ec.EllipticCurvePublicKey):
            msg = "Public key is not an elliptic curve key"
            raise TypeError(msg)  # ruff: ignore[raise-within-try]

        curve_name = public_key.curve.name
        key_size = public_key.curve.key_size
        block_length = key_size // 8

        return cls(
            key=key_hex,
            curve=curve_name,
            size=key_size,
            block_length=block_length,
        )
    except UnsupportedAlgorithm as e:
        curve = fallback_curve(key_bytes)
        if curve is None:
            msg = f"Unsupported elliptic curve in public key: {e}"
            raise PublicKeyError(msg) from e
        size = _CURVE_SIZES[curve]
        return cls(key=key_hex, curve=curve, size=size, block_length=size // 8)
    except (ValueError, TypeError) as e:
        msg = f"Failed to parse public key: {e}"
        raise PublicKeyError(msg) from e

matches_signature_algorithm(signature_algorithm: SignatureMethod | str | None) -> bool

Whether the key uses the curve of the signature algorithm (SA).

Source code in src/pyocmf/models/public_key.py

def matches_signature_algorithm(
    self, signature_algorithm: SignatureMethod | str | None
) -> bool:
    """Whether the key uses the curve of the signature algorithm (SA)."""
    if not isinstance(signature_algorithm, SignatureMethod):
        return False

    return self.curve == signature_algorithm.curve

OBIS

OBIS code model for meter reading identifiers.

OBIS

Bases: BaseModel

OBIS code of a reading register (RI), with an optional suffix after *.

Source code in src/pyocmf/models/obis.py

class OBIS(pydantic.BaseModel):
    """OBIS code of a reading register (RI), with an optional suffix after ``*``."""

    model_config = pydantic.ConfigDict(frozen=True)

    code: str
    suffix: str | None = None

    @classmethod
    def from_string(cls, obis_str: str) -> OBIS:
        """Split an OBIS string into the code and the suffix after ``*``."""
        if not isinstance(obis_str, str):
            return obis_str

        parts = obis_str.split("*", 1)
        return cls(
            code=parts[0],
            suffix=parts[1] if len(parts) > 1 else None,
        )

    @pydantic.model_serializer
    def serialize_to_string(self) -> str:
        # OCMF spec Table 7: RI is a JSON String, not an object
        return str(self)

    @property
    def info(self) -> OBISInfo | None:
        """Registry entry of a known code, else None."""
        return get_obis_info(self.code)

    @property
    def is_billing_relevant(self) -> bool:
        """See ``pyocmf.registries.obis.is_billing_relevant``."""
        return is_billing_relevant(self.code)

    @property
    def is_law_relevant(self) -> bool:
        """See ``pyocmf.registries.obis.is_law_relevant``."""
        return is_law_relevant(str(self))

    @property
    def is_loss_compensated(self) -> bool:
        """Whether the register holds loss-compensated energy (98, B1, B3, C1, C3)."""
        return is_loss_compensated(self.code)

    @property
    def is_accumulation_register(self) -> bool:
        """See ``pyocmf.registries.obis.is_accumulation_register``."""
        return is_accumulation_register(self.code)

    @property
    def is_transaction_register(self) -> bool:
        """See ``pyocmf.registries.obis.is_transaction_register``."""
        return is_transaction_register(self.code)

    def __str__(self) -> str:
        return f"{self.code}*{self.suffix}" if self.suffix else self.code

    def __repr__(self) -> str:
        if self.suffix:
            return f"OBIS('{self.code}*{self.suffix}')"
        return f"OBIS('{self.code}')"

info: OBISInfo | None property

Registry entry of a known code, else None.

is_billing_relevant: bool property

See pyocmf.registries.obis.is_billing_relevant.

is_law_relevant: bool property

See pyocmf.registries.obis.is_law_relevant.

is_loss_compensated: bool property

Whether the register holds loss-compensated energy (98, B1, B3, C1, C3).

is_accumulation_register: bool property

See pyocmf.registries.obis.is_accumulation_register.

is_transaction_register: bool property

See pyocmf.registries.obis.is_transaction_register.

from_string(obis_str: str) -> OBIS classmethod

Split an OBIS string into the code and the suffix after *.

Source code in src/pyocmf/models/obis.py

@classmethod
def from_string(cls, obis_str: str) -> OBIS:
    """Split an OBIS string into the code and the suffix after ``*``."""
    if not isinstance(obis_str, str):
        return obis_str

    parts = obis_str.split("*", 1)
    return cls(
        code=parts[0],
        suffix=parts[1] if len(parts) > 1 else None,
    )

OCMFTimestamp

Timestamp with time synchronization status.

OCMFTimestamp dataclass

Source code in src/pyocmf/models/timestamp.py

@dataclass(frozen=True)
class OCMFTimestamp:
    timestamp: datetime
    status: TimeStatus | str

    def __str__(self) -> str:
        return self.serialize()

    @classmethod
    def from_string(cls, timestamp_str: str) -> OCMFTimestamp:
        """Parse OCMF timestamp string to OCMFTimestamp.

        OCMF format: "2023-06-15T14:30:45,123+0200 S" (note: comma for milliseconds).
        """
        if " " in timestamp_str:
            ts_part, status_part = timestamp_str.rsplit(" ", 1)
            status = _parse_time_status(status_part)
        else:
            ts_part = timestamp_str
            status = TimeStatus.UNKNOWN_OR_UNSYNCHRONIZED

        ts_normalized = ts_part.replace(",", ".")
        dt = datetime.fromisoformat(ts_normalized)

        return cls(timestamp=dt, status=status)

    def serialize(self) -> str:
        """Serialize to OCMF timestamp format.

        Uses comma for milliseconds and a colon-free timezone offset (e.g. +0200)
        as required by the OCMF spec.
        """
        if self.timestamp.tzinfo is None:
            error_message = "Datetime must be timezone-aware for OCMF format"
            raise ValueError(error_message)

        iso_str = self.timestamp.isoformat(timespec="milliseconds")
        ocmf_str = re.sub(r"([+-]\d{2}):(\d{2})$", r"\1\2", iso_str.replace(".", ","))

        return f"{ocmf_str} {self.status}"

from_string(timestamp_str: str) -> OCMFTimestamp classmethod

Parse OCMF timestamp string to OCMFTimestamp.

OCMF format: "2023-06-15T14:30:45,123+0200 S" (note: comma for milliseconds).

Source code in src/pyocmf/models/timestamp.py

@classmethod
def from_string(cls, timestamp_str: str) -> OCMFTimestamp:
    """Parse OCMF timestamp string to OCMFTimestamp.

    OCMF format: "2023-06-15T14:30:45,123+0200 S" (note: comma for milliseconds).
    """
    if " " in timestamp_str:
        ts_part, status_part = timestamp_str.rsplit(" ", 1)
        status = _parse_time_status(status_part)
    else:
        ts_part = timestamp_str
        status = TimeStatus.UNKNOWN_OR_UNSYNCHRONIZED

    ts_normalized = ts_part.replace(",", ".")
    dt = datetime.fromisoformat(ts_normalized)

    return cls(timestamp=dt, status=status)

serialize() -> str

Serialize to OCMF timestamp format.

Uses comma for milliseconds and a colon-free timezone offset (e.g. +0200) as required by the OCMF spec.

Source code in src/pyocmf/models/timestamp.py

def serialize(self) -> str:
    """Serialize to OCMF timestamp format.

    Uses comma for milliseconds and a colon-free timezone offset (e.g. +0200)
    as required by the OCMF spec.
    """
    if self.timestamp.tzinfo is None:
        error_message = "Datetime must be timezone-aware for OCMF format"
        raise ValueError(error_message)

    iso_str = self.timestamp.isoformat(timespec="milliseconds")
    ocmf_str = re.sub(r"([+-]\d{2}):(\d{2})$", r"\1\2", iso_str.replace(".", ","))

    return f"{ocmf_str} {self.status}"

CableLossCompensation

Cable loss compensation data.

CableLossCompensation

Bases: BaseModel

Cable loss compensation parameters (OCMF spec Table 24).

The Transparenzsoftware ignores them, so the Eichrecht checks only warn about them.

Source code in src/pyocmf/models/cable_loss.py

class CableLossCompensation(pydantic.BaseModel):
    """Cable loss compensation parameters (OCMF spec Table 24).

    The Transparenzsoftware ignores them, so the Eichrecht checks only warn about them.
    """

    LN: str | None = pydantic.Field(default=None, description="Loss Compensation Naming (0..20)")
    LI: int | None = pydantic.Field(default=None, description="Loss Compensation Identification")
    LR: OCMFNumber | None = pydantic.Field(
        default=None, description="Loss Compensation Cable Resistance - REQUIRED"
    )
    LU: LenientResistanceUnit | None = pydantic.Field(
        default=None, description="Loss Compensation Unit - REQUIRED"
    )

    @pydantic.model_validator(mode="after")
    def warn_on_spec_deviations(self) -> CableLossCompensation:
        if self.LN is not None and len(self.LN) > _MAX_NAMING_LENGTH:
            warn_spec(f"Loss compensation naming (LN) exceeds {_MAX_NAMING_LENGTH} characters")
        if self.LR is None:
            warn_spec("Loss compensation cable resistance (LR) is mandatory")
        if self.LU is None:
            warn_spec("Loss compensation resistance unit (LU) is mandatory")
        return self

Registries

OBIS Code Registry

Utilities for working with OBIS codes.

get_obis_info(obis_code: str) -> OBISInfo | None

Look up a known OBIS code; None for codes not in the registry.

Source code in src/pyocmf/registries/obis.py

def get_obis_info(obis_code: str) -> OBISInfo | None:
    """Look up a known OBIS code; None for codes not in the registry."""
    groups = parse_obis(obis_code)
    return _KNOWN_OBIS_BY_KEY.get(groups.key) if groups else None

is_billing_relevant(obis_code: str) -> bool

Whether the register holds energy that can be billed.

Known codes use the registry. Other cumulative energy registers count when they are law-relevant or hold active export energy, so this is broader than is_law_relevant, which the Eichrecht checks use.

Source code in src/pyocmf/registries/obis.py

def is_billing_relevant(obis_code: str) -> bool:
    """Whether the register holds energy that can be billed.

    Known codes use the registry. Other cumulative energy registers count when they
    are law-relevant or hold active export energy, so this is broader than
    ``is_law_relevant``, which the Eichrecht checks use.
    """
    if (info := get_obis_info(obis_code)) is not None:
        return info.billing_relevant
    groups = parse_obis(obis_code)
    return (
        groups is not None
        and groups.is_cumulative_energy
        and (groups.c in _LAW_RELEVANT_C or groups.c == _ACTIVE_EXPORT)
    )

is_law_relevant(obis_code: str) -> bool

Whether the register is compared for a transaction, as in the Transparenzsoftware.

These are cumulative energy registers for active import (01), the vendor registers 98 (loss-compensated) and 9E, and the range B0-C7 that OCMF spec Table 25 reserves. An F group, if present, must be 0 or FF (or 200/255, which the Transparenzsoftware also accepts).

Source code in src/pyocmf/registries/obis.py

def is_law_relevant(obis_code: str) -> bool:
    """Whether the register is compared for a transaction, as in the Transparenzsoftware.

    These are cumulative energy registers for active import (01), the vendor registers
    98 (loss-compensated) and 9E, and the range B0-C7 that OCMF spec Table 25 reserves.
    An F group, if present, must be 0 or FF (or 200/255, which the Transparenzsoftware
    also accepts).
    """
    groups = parse_obis(obis_code)
    return (
        groups is not None
        and groups.is_cumulative_energy
        and groups.c in _LAW_RELEVANT_C
        and (groups.f is None or groups.f in _LAW_RELEVANT_F)
    )

parse_obis(obis_code: str) -> OBISGroups | None

Parse an OBIS code into its groups, read as hex as in the OCMF spec; None if malformed.

Source code in src/pyocmf/registries/obis.py

def parse_obis(obis_code: str) -> OBISGroups | None:
    """Parse an OBIS code into its groups, read as hex as in the OCMF spec; None if malformed."""
    match = _OBIS_PATTERN.match(obis_code.strip())
    if match is None:
        return None
    f = match["f"]
    return OBISGroups(
        a=int(match["a"], 16),
        b=int(match["b"], 16),
        c=int(match["c"], 16),
        d=int(match["d"], 16),
        e=int(match["e"], 16),
        f=int(f, 16) if f is not None else None,
    )

is_accumulation_register(obis_code: str) -> bool

Whether the code is an OCMF reserved energy register (B0-B3, C0-C3; spec Table 25).

Source code in src/pyocmf/registries/obis.py

def is_accumulation_register(obis_code: str) -> bool:
    """Whether the code is an OCMF reserved energy register (B0-B3, C0-C3; spec Table 25)."""
    return _is_reserved_register(obis_code, _ACCUMULATION_C)

is_transaction_register(obis_code: str) -> bool

Whether the code is an OCMF reserved per-transaction register (B2, B3, C2, C3).

Source code in src/pyocmf/registries/obis.py

def is_transaction_register(obis_code: str) -> bool:
    """Whether the code is an OCMF reserved per-transaction register (B2, B3, C2, C3)."""
    return _is_reserved_register(obis_code, _TRANSACTION_C)

OBISInfo dataclass

Description and classification of a known OBIS code.

Source code in src/pyocmf/registries/obis.py

@dataclass
class OBISInfo:
    """Description and classification of a known OBIS code."""

    code: str
    description: str
    billing_relevant: bool
    category: OBISCategory

    @staticmethod
    def normalize(obis_code: str) -> str:
        return _normalize(obis_code)

    @staticmethod
    def from_code(obis_code: str) -> OBISInfo | None:
        return get_obis_info(obis_code)

    def is_accumulation_register(self) -> bool:
        return is_accumulation_register(self.code)

    def is_transaction_register(self) -> bool:
        return is_transaction_register(self.code)

XML Utilities

OcmfContainer

Container for parsing OCMF data from XML transaction files.

OcmfContainer

OCMF records read from a Transparenzsoftware XML file, in file order.

Records that occur more than once are kept once. Supports len(), iteration and indexing.

Source code in src/pyocmf/utils/xml.py

class OcmfContainer:
    """OCMF records read from a Transparenzsoftware XML file, in file order.

    Records that occur more than once are kept once. Supports ``len()``, iteration
    and indexing.
    """

    def __init__(self, entries: list[OcmfRecord]) -> None:
        self._entries = entries

    @classmethod
    def from_xml(cls, xml_path: pathlib.Path | str, *, strict: bool = False) -> OcmfContainer:
        """Parse OCMF data from an XML file.

        Args:
            xml_path: Path to the XML file
            strict: Reject OCMF records that deviate from the spec instead of warning

        Returns:
            OcmfContainer with parsed OCMF entries

        Raises:
            XmlParsingError: If the XML file cannot be parsed
            DataNotFoundError: If no OCMF data is found

        """
        path = pathlib.Path(xml_path)

        try:
            tree = ET.parse(path)
            root = tree.getroot()
        except ET.ParseError as e:
            msg = f"Failed to parse XML file: {e}"
            raise XmlParsingError(msg) from e

        entries = []
        seen_strings: set[str] = set()

        for value_elem in root.findall("value"):
            ocmf_str = _extract_ocmf_string(value_elem)

            if ocmf_str and ocmf_str not in seen_strings:
                ocmf = OCMF.from_string(ocmf_str, strict=strict)
                entries.append(
                    OcmfRecord(
                        ocmf=ocmf,
                        public_key=_extract_public_key(value_elem),
                        transaction_id=_parse_transaction_id(value_elem.get("transactionId")),
                        context=(value_elem.get("context") or "").strip() or None,
                    )
                )
                seen_strings.add(ocmf_str)

        if not entries:
            msg = "No OCMF data found in XML file"
            raise DataNotFoundError(msg)

        return cls(entries)

    @property
    def entries(self) -> list[OcmfRecord]:
        """The records as a list."""
        return self._entries

    def __len__(self) -> int:
        return len(self._entries)

    def __iter__(self) -> Iterator[OcmfRecord]:
        return iter(self._entries)

    def __getitem__(self, index: int) -> OcmfRecord:
        return self._entries[index]

    def transactions(self) -> dict[int, list[OcmfRecord]]:
        """Group the records by their transaction ID, in file order."""
        groups: dict[int, list[OcmfRecord]] = {}
        for record in self._entries:
            if record.transaction_id is not None:
                groups.setdefault(record.transaction_id, []).append(record)
        return groups

    def check_eichrecht(self, *, errors_only: bool = False) -> list[EichrechtResult]:
        """Check Eichrecht compliance of the file, as the Transparenzsoftware does.

        Records sharing a transaction ID form a transaction. It needs exactly one
        record with the context ``Transaction.Begin`` and one with ``Transaction.End``,
        which are checked as a begin/end pair. Every other record is checked on its own.

        Args:
            errors_only: Leave out warnings

        Returns:
            One result per transaction or standalone record, in file order

        """
        transactions = {tid: rs for tid, rs in self.transactions().items() if len(rs) > 1}
        results: list[EichrechtResult] = []
        for record in self._entries:
            tid = record.transaction_id
            if tid is None or tid not in transactions:
                results.append(EichrechtResult([record], record.ocmf.check_eichrecht()))
            elif record is transactions[tid][0]:
                results.append(_check_transaction(tid, transactions[tid]))

        if errors_only:
            for result in results:
                result.issues = [i for i in result.issues if i.severity == IssueSeverity.ERROR]
        return results

entries: list[OcmfRecord] property

The records as a list.

from_xml(xml_path: pathlib.Path | str, *, strict: bool = False) -> OcmfContainer classmethod

Parse OCMF data from an XML file.

Parameters:

Name	Type	Description	Default
xml_path	Path | str	Path to the XML file	required
strict	bool	Reject OCMF records that deviate from the spec instead of warning	False

Returns:

Type	Description
OcmfContainer	OcmfContainer with parsed OCMF entries

Raises:

Type	Description
XmlParsingError	If the XML file cannot be parsed
DataNotFoundError	If no OCMF data is found

Source code in src/pyocmf/utils/xml.py

@classmethod
def from_xml(cls, xml_path: pathlib.Path | str, *, strict: bool = False) -> OcmfContainer:
    """Parse OCMF data from an XML file.

    Args:
        xml_path: Path to the XML file
        strict: Reject OCMF records that deviate from the spec instead of warning

    Returns:
        OcmfContainer with parsed OCMF entries

    Raises:
        XmlParsingError: If the XML file cannot be parsed
        DataNotFoundError: If no OCMF data is found

    """
    path = pathlib.Path(xml_path)

    try:
        tree = ET.parse(path)
        root = tree.getroot()
    except ET.ParseError as e:
        msg = f"Failed to parse XML file: {e}"
        raise XmlParsingError(msg) from e

    entries = []
    seen_strings: set[str] = set()

    for value_elem in root.findall("value"):
        ocmf_str = _extract_ocmf_string(value_elem)

        if ocmf_str and ocmf_str not in seen_strings:
            ocmf = OCMF.from_string(ocmf_str, strict=strict)
            entries.append(
                OcmfRecord(
                    ocmf=ocmf,
                    public_key=_extract_public_key(value_elem),
                    transaction_id=_parse_transaction_id(value_elem.get("transactionId")),
                    context=(value_elem.get("context") or "").strip() or None,
                )
            )
            seen_strings.add(ocmf_str)

    if not entries:
        msg = "No OCMF data found in XML file"
        raise DataNotFoundError(msg)

    return cls(entries)

transactions() -> dict[int, list[OcmfRecord]]

Group the records by their transaction ID, in file order.

Source code in src/pyocmf/utils/xml.py

def transactions(self) -> dict[int, list[OcmfRecord]]:
    """Group the records by their transaction ID, in file order."""
    groups: dict[int, list[OcmfRecord]] = {}
    for record in self._entries:
        if record.transaction_id is not None:
            groups.setdefault(record.transaction_id, []).append(record)
    return groups

check_eichrecht(*, errors_only: bool = False) -> list[EichrechtResult]

Check Eichrecht compliance of the file, as the Transparenzsoftware does.

Records sharing a transaction ID form a transaction. It needs exactly one record with the context Transaction.Begin and one with Transaction.End, which are checked as a begin/end pair. Every other record is checked on its own.

Parameters:

Name	Type	Description	Default
errors_only	bool	Leave out warnings	False

Returns:

Type	Description
list[EichrechtResult]	One result per transaction or standalone record, in file order

Source code in src/pyocmf/utils/xml.py

def check_eichrecht(self, *, errors_only: bool = False) -> list[EichrechtResult]:
    """Check Eichrecht compliance of the file, as the Transparenzsoftware does.

    Records sharing a transaction ID form a transaction. It needs exactly one
    record with the context ``Transaction.Begin`` and one with ``Transaction.End``,
    which are checked as a begin/end pair. Every other record is checked on its own.

    Args:
        errors_only: Leave out warnings

    Returns:
        One result per transaction or standalone record, in file order

    """
    transactions = {tid: rs for tid, rs in self.transactions().items() if len(rs) > 1}
    results: list[EichrechtResult] = []
    for record in self._entries:
        tid = record.transaction_id
        if tid is None or tid not in transactions:
            results.append(EichrechtResult([record], record.ocmf.check_eichrecht()))
        elif record is transactions[tid][0]:
            results.append(_check_transaction(tid, transactions[tid]))

    if errors_only:
        for result in results:
            result.issues = [i for i in result.issues if i.severity == IssueSeverity.ERROR]
    return results

OcmfRecord

A single OCMF record with its associated public key.

OcmfRecord dataclass

An OCMF record from an XML file, with the public key stored next to it, if any.

transaction_id and context come from the transactionId and context attributes of the XML value. The Transparenzsoftware uses them to pair the begin and end record of a transaction.

Source code in src/pyocmf/utils/xml.py

@dataclass
class OcmfRecord:
    """An OCMF record from an XML file, with the public key stored next to it, if any.

    ``transaction_id`` and ``context`` come from the ``transactionId`` and ``context``
    attributes of the XML value. The Transparenzsoftware uses them to pair the begin
    and end record of a transaction.
    """

    ocmf: OCMF
    public_key: PublicKey | None = None
    transaction_id: int | None = None
    context: str | None = None

    def verify_signature(self) -> bool:
        """Verify with the record's public key, or else a key embedded in the OCMF string.

        Raises:
            SignatureVerificationError: If no key is available or the signature cannot
                be checked

        """
        if self.public_key is None and self.ocmf.embedded_public_key is None:
            msg = "No public key available for signature verification"
            raise SignatureVerificationError(msg)

        return self.ocmf.verify_signature(self.public_key)

verify_signature() -> bool

Verify with the record's public key, or else a key embedded in the OCMF string.

Raises:

Type	Description
SignatureVerificationError	If no key is available or the signature cannot be checked

Source code in src/pyocmf/utils/xml.py

def verify_signature(self) -> bool:
    """Verify with the record's public key, or else a key embedded in the OCMF string.

    Raises:
        SignatureVerificationError: If no key is available or the signature cannot
            be checked

    """
    if self.public_key is None and self.ocmf.embedded_public_key is None:
        msg = "No public key available for signature verification"
        raise SignatureVerificationError(msg)

    return self.ocmf.verify_signature(self.public_key)

EichrechtResult

Eichrecht issues of one transaction or standalone record of an XML file.

EichrechtResult dataclass

Eichrecht issues of one transaction, or of one record checked on its own.

Source code in src/pyocmf/utils/xml.py

@dataclass
class EichrechtResult:
    """Eichrecht issues of one transaction, or of one record checked on its own."""

    records: list[OcmfRecord]
    issues: list[EichrechtIssue] = field(default_factory=list)
    transaction_id: int | None = None

    @property
    def is_compliant(self) -> bool:
        """Whether none of the issues is an error."""
        return not any(issue.severity == IssueSeverity.ERROR for issue in self.issues)

is_compliant: bool property

Whether none of the issues is an error.

Exceptions

All exceptions inherit from PyOCMFError.

PyOCMFError

Bases: Exception

Base class for all pyocmf errors; field names the OCMF field involved.

Source code in src/pyocmf/exceptions.py

class PyOCMFError(Exception):
    """Base class for all pyocmf errors; ``field`` names the OCMF field involved."""

    def __init__(
        self,
        message: str,
        *,
        field: str | None = None,
        details: list[dict[str, Any]] | None = None,
    ) -> None:
        super().__init__(message)
        self.field = field
        self.details = details

OCMF Parsing Errors

OcmfFormatError

Bases: PyOCMFError

The string is not structured as OCMF, or carries an appended key in strict mode.

Source code in src/pyocmf/exceptions.py

class OcmfFormatError(PyOCMFError):
    """The string is not structured as OCMF, or carries an appended key in strict mode."""

OcmfPayloadError

Bases: PyOCMFError

The payload section is not valid JSON or fails validation.

Source code in src/pyocmf/exceptions.py

class OcmfPayloadError(PyOCMFError):
    """The payload section is not valid JSON or fails validation."""

OcmfSignatureError

Bases: PyOCMFError

The signature section is not valid JSON or fails validation.

Source code in src/pyocmf/exceptions.py

class OcmfSignatureError(PyOCMFError):
    """The signature section is not valid JSON or fails validation."""

Spec Deviations

SpecWarning

Bases: UserWarning

Input deviates from the OCMF spec but is accepted, as the Transparenzsoftware does.

Source code in src/pyocmf/exceptions.py

class SpecWarning(UserWarning):
    """Input deviates from the OCMF spec but is accepted, as the Transparenzsoftware does."""

SpecViolationError

Bases: PyOCMFError, ValueError

Input deviates from the OCMF spec while parsing in strict mode.

Source code in src/pyocmf/exceptions.py

class SpecViolationError(PyOCMFError, ValueError):
    """Input deviates from the OCMF spec while parsing in strict mode."""

Encoding Errors

EncodingError

Bases: PyOCMFError, ValueError

Base class for errors decoding hex or base64 data.

Source code in src/pyocmf/exceptions.py

class EncodingError(PyOCMFError, ValueError):
    """Base class for errors decoding hex or base64 data."""

    def __init__(
        self,
        message: str,
        *,
        value: str | None = None,
        field: str | None = None,
        details: list[dict[str, Any]] | None = None,
    ) -> None:
        super().__init__(message, field=field, details=details)
        self.value = value

EncodingTypeError

Bases: PyOCMFError, TypeError

Data to decode is not a string.

Source code in src/pyocmf/exceptions.py

class EncodingTypeError(PyOCMFError, TypeError):
    """Data to decode is not a string."""

    def __init__(
        self,
        message: str,
        *,
        value: object = None,
        expected_type: str | None = None,
        field: str | None = None,
        details: list[dict[str, Any]] | None = None,
    ) -> None:
        super().__init__(message, field=field, details=details)
        self.value = value
        self.expected_type = expected_type

HexDecodingError

Bases: EncodingError

Data is not valid hex.

Source code in src/pyocmf/exceptions.py

class HexDecodingError(EncodingError):
    """Data is not valid hex."""

Base64DecodingError

Bases: EncodingError

Data is not valid base64.

Source code in src/pyocmf/exceptions.py

class Base64DecodingError(EncodingError):
    """Data is not valid base64."""

Cryptography Errors

CryptoError

Bases: PyOCMFError

Base class for public key and signature verification errors.

Source code in src/pyocmf/exceptions.py

class CryptoError(PyOCMFError):
    """Base class for public key and signature verification errors."""

SignatureVerificationError

Bases: CryptoError

The signature cannot be checked, e.g. without a key or with a mismatching curve.

A signature that does not match the payload is not an error: verification returns False.

Source code in src/pyocmf/exceptions.py

class SignatureVerificationError(CryptoError):
    """The signature cannot be checked, e.g. without a key or with a mismatching curve.

    A signature that does not match the payload is not an error: verification
    returns False.
    """

    def __init__(
        self,
        message: str,
        *,
        reason: str | None = None,
        field: str | None = None,
        details: list[dict[str, Any]] | None = None,
    ) -> None:
        super().__init__(message, field=field, details=details)
        self.reason = reason

PublicKeyError

Bases: CryptoError

The public key cannot be parsed or uses an unsupported curve.

Source code in src/pyocmf/exceptions.py

class PublicKeyError(CryptoError):
    """The public key cannot be parsed or uses an unsupported curve."""

    def __init__(
        self,
        message: str,
        *,
        key_data: str | None = None,
        field: str | None = None,
        details: list[dict[str, Any]] | None = None,
    ) -> None:
        super().__init__(message, field=field, details=details)
        self.key_data = key_data

Data Errors

DataNotFoundError

Bases: PyOCMFError

The XML file contains no OCMF data.

Source code in src/pyocmf/exceptions.py

class DataNotFoundError(PyOCMFError):
    """The XML file contains no OCMF data."""

XmlParsingError

Bases: PyOCMFError

The XML file is not well-formed.

Source code in src/pyocmf/exceptions.py

class XmlParsingError(PyOCMFError):
    """The XML file is not well-formed."""