Skip to content
Merged
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
10 changes: 8 additions & 2 deletions custom_components/zaptec/sensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,7 @@

_log_attribute = "_attr_native_value"
# This entity use several attributes from Zaptec
_log_zaptec_key = ["signed_meter_value", "completed_session"]

Check failure on line 103 in custom_components/zaptec/sensor.py

View workflow job for this annotation

GitHub Actions / Ruff validation

Ruff (RUF012)

custom_components/zaptec/sensor.py:103:23: RUF012 Mutable class attributes should be annotated with `typing.ClassVar`

@callback
def _update_from_zaptec(self) -> None:
Expand All @@ -111,14 +111,20 @@
# are OCMF (Open Charge Metering Format) data structures and must be
# parsed to get the latest reading.

# Ge the two OCMF data structures from Zaptec. The first one must exists,
# Get the two OCMF data structures from Zaptec. The first one must exists,
# the second one is optional.
meter_value = self._get_zaptec_value(key="signed_meter_value")
session = self._get_zaptec_value(key="completed_session", default={})

# Get the latest energy reading from both and use the largest value
reading = get_ocmf_max_reader_value(meter_value)
session_reading = get_ocmf_max_reader_value(session.get("SignedSession", {}))
if isinstance(session, dict):
session_reading = get_ocmf_max_reader_value(session.get("SignedSession", {}))
else:
_LOGGER.debug(
"Incorrect typing for completed_session: %s", type(session).__qualname__
)
session_reading = 0.0

self._attr_native_value = max(reading, session_reading)
self._attr_available = True
Expand Down
2 changes: 2 additions & 0 deletions custom_components/zaptec/zaptec/zconst.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,8 @@ def type_network_type(self, val: int) -> str:
def type_ocmf(self, data: str) -> dict[str, Any]:
"""Open Charge Metering Format (OCMF) type."""
# https://github.com/SAFE-eV/OCMF-Open-Charge-Metering-Format/blob/master/OCMF-en.md
if not data:
return {}
sects = data.split("|")
if len(sects) not in (2, 3) or sects[0] != "OCMF":
raise ValueError(f"Invalid OCMF data: {data}")
Expand Down
46 changes: 46 additions & 0 deletions tests/zaptec/test_zconst.py
Original file line number Diff line number Diff line change
Expand Up @@ -173,6 +173,52 @@ def test_completed_session() -> None:
assert ZCONST.type_completed_session(completed_session) == expected_completed_session_dict


def test_completed_session_corrupted_or_no_signed_session() -> None:
"""Test type conversion for completed_session if there is a corrupted or no signed session."""

expected_energy = 27.432
completed_session_signed_is_null = (
r'{"SessionId":"01234567-89ab-cdef-0123-456789abcdef","Energy":27.432,"StartDateTime":'
r'"2025-08-28T15:02:55.188048Z","EndDateTime":"2025-09-01T10:32:50.517055Z",'
r'"ReliableClock":true,"StoppedByRFID":false,"AuthenticationCode":"",'
r'"FirstAuthenticatedDateTime":"","SignedSession":null}'
)
res_signed_is_null = ZCONST.type_completed_session(completed_session_signed_is_null)
assert res_signed_is_null.get("Energy") == expected_energy

completed_session_signed_is_empty = (
r'{"SessionId":"01234567-89ab-cdef-0123-456789abcdef","Energy":27.432,"StartDateTime":'
r'"2025-08-28T15:02:55.188048Z","EndDateTime":"2025-09-01T10:32:50.517055Z",'
r'"ReliableClock":true,"StoppedByRFID":false,"AuthenticationCode":"",'
r'"FirstAuthenticatedDateTime":"","SignedSession":""}'
)
res_signed_is_empty = ZCONST.type_completed_session(completed_session_signed_is_empty)
assert res_signed_is_empty.get("Energy") == expected_energy

completed_session_no_signed = (
r'{"SessionId":"01234567-89ab-cdef-0123-456789abcdef","Energy":27.432,"StartDateTime":'
r'"2025-08-28T15:02:55.188048Z","EndDateTime":"2025-09-01T10:32:50.517055Z",'
r'"ReliableClock":true,"StoppedByRFID":false,"AuthenticationCode":"",'
r'"FirstAuthenticatedDateTime":""}'
)
res_no_signed = ZCONST.type_completed_session(completed_session_no_signed)
assert res_no_signed.get("Energy") == expected_energy

completed_session_signed_invalid_ocmf = ( # OMCF instead of OCMF
r'{"SessionId":"01234567-89ab-cdef-0123-456789abcdef","Energy":27.432,"StartDateTime":'
r'"2025-08-28T15:02:55.188048Z","EndDateTime":"2025-09-01T10:32:50.517055Z",'
r'"ReliableClock":true,"StoppedByRFID":false,"AuthenticationCode":"",'
r'"FirstAuthenticatedDateTime":"","SignedSession":"OMCF|{\"FV\":\"1.0\",\"GI\":'
r"\"ZAPTEC GO\",\"GS\":\"ZAP000042\",\"GV\":\"2.4.2.4\",\"PG\":\"T1\",\"RD\":[{\"TM\":"
r"\"2025-08-28T15:02:55,000+00:00 R\",\"TX\":\"B\",\"RV\":1472.07,\"RI\":\"1-0:1.8.0\","
r"\"RU\":\"kWh\",\"RT\":\"AC\",\"ST\":\"G\"},{\"TM\":\"2025-08-29T12:00:00,000+00:00 R\","
r"\"TX\":\"T\",\"RV\":1472.29,\"RI\":\"1-0:1.8.0\",\"RU\":\"kWh\",\"RT\":\"AC\",\"ST\":"
r'\"G\"}]}"}'
)
with pytest.raises(ValueError, match=r"Invalid OCMF data:*"):
ZCONST.type_completed_session(completed_session_signed_invalid_ocmf)


def test_update_ids_from_schema() -> None:
"""Test update_ids_from_schema."""
ZCONST.update_ids_from_schema(set("Apollo"))
Loading