diff --git a/src/egd_dlms/discovery.py b/src/egd_dlms/discovery.py index 5080551..8d84ddf 100644 --- a/src/egd_dlms/discovery.py +++ b/src/egd_dlms/discovery.py @@ -68,6 +68,12 @@ def publish_discovery(client, logger): "unique_id": f"egd_meter_{key}", "state_topic": state_topic, "availability_topic": availability_topic, + # Záměrně BEZ .get() — chybějící klíč (payload nese jen pole, + # která se objevila v aktuálním DLMS rámci) tak template + # vykreslení "selže" a HA podrží poslední ZNÁMOU hodnotu místo + # blikání na "unavailable" každý cyklus. Garbage hodnoty už řeší + # parser (viz _is_plausible v parser.py) — tahle šablona teď řeší + # jen zobrazení, ne validitu dat. "value_template": "{{ value_json." + key + " }}", "device": device, } @@ -82,3 +88,5 @@ def publish_discovery(client, logger): client.publish(topic, json.dumps(payload), qos=0, retain=True) logger.info("MQTT Discovery odesláno") + + diff --git a/src/egd_dlms/parser.py b/src/egd_dlms/parser.py index 4bbe322..b148200 100644 --- a/src/egd_dlms/parser.py +++ b/src/egd_dlms/parser.py @@ -1,5 +1,28 @@ +from egd_dlms.config import OBIS_CONFIG from egd_dlms.models import CosemObject +# AXDR typové značky používané tímhle meterem pro hodnotu COSEM Register +# atributu (empiricky ověřeno na zachycených rámcích, viz +# project_egd_dlms_power_field_bug.md v Claude memory) -> délka hodnoty v bajtech. +VALUE_TYPE_SIZES = { + 0x06: 4, # double-long-unsigned (uint32) + 0x05: 4, # double-long (int32) + 0x12: 2, # long-unsigned (uint16) + 0x10: 2, # long (int16) + 0x11: 1, # unsigned (uint8) + 0x0F: 1, # integer (int8) +} + +# Bezpečnostní strop na "rozumnou" hodnotu podle device_class z obis.yaml. +# RS485/USB spojení občas dodá bitově poškozený rámec (typicky po USB +# re-enumeraci FT232 adaptéru) — bez týhle kontroly se to projeví jako +# hodnoty v řádu milionů/miliard wattů. Lepší takovou hodnotu zahodit +# (pole zůstane na poslední známé hodnotě) než publikovat nesmysl do HA. +MAX_PLAUSIBLE_BY_DEVICE_CLASS = { + "power": 30_000, # W — scale je u všech power OBIS kódů 1 + "energy": 100_000_000, # raw jednotky, scale 0.001 -> až 100 000 kWh +} + class CosemParser: def __init__(self, logger): @@ -10,21 +33,21 @@ def parse(self, data: bytes) -> list[CosemObject]: i = 0 while i < len(data): - if self._looks_like_register_value(data, i): - class_id = int.from_bytes(data[i + 2:i + 4], "big") - obis_raw = data[i + 4:i + 10] + register = self._register_value_at(data, i) + if register is not None: + obis_raw, value, consumed = register objects.append( CosemObject( - class_id=class_id, + class_id=3, logical_name=self._obis_short(obis_raw), full_obis=self._obis_full(obis_raw), attribute=2, - value=int.from_bytes(data[i + 12:i + 16], "big", signed=False), + value=value, ) ) - i += 16 + i += consumed continue if self._looks_like_enum_value(data, i): @@ -48,27 +71,118 @@ def parse(self, data: bytes) -> list[CosemObject]: return objects + def _register_value_at(self, data: bytes, i: int) -> tuple[bytes, int, int] | None: + """Zkusí na pozici i najít COSEM Register (class_id=3) záznam. + + Skutečná struktura na tomhle meteru (ověřeno na reálně zachycených + rámcích, viz project_egd_dlms_power_field_bug.md): 2bajtový class_id + BEZ obalové značky "02 02" (starší kód ji mylně vyžadoval, takže na + živých datech skoro nikdy nic nenašel), 6bajtový OBIS kód, 2bajtová + AXDR typová značka (první bajt vždy 0x02, druhý určuje typ/délku + hodnoty) a hodnota. + + Vrací None, pokud na pozici i nic nesedí, OBIS kód neznáme (viz + obis.yaml), nebo hodnota neprojde kontrolou rozumnosti — RS485/USB + spojení občas dodá bitově poškozený rámec a je lepší takové pole + zahodit (zůstane na poslední známé hodnotě), než publikovat + nesmysl (viz MAX_PLAUSIBLE_BY_DEVICE_CLASS). + """ + if i + 10 > len(data): + return None + if data[i] != 0x00 or data[i + 1] != 0x03: + return None + + obis_raw = data[i + 2:i + 8] + short = self._obis_short(obis_raw) + full = self._obis_full(obis_raw) + + mapping = OBIS_CONFIG.get(full) or OBIS_CONFIG.get(short) + if mapping is None: + return None + + if data[i + 8] != 0x02: + return None + + size = VALUE_TYPE_SIZES.get(data[i + 9]) + if size is None or i + 10 + size > len(data): + return None + + raw_value = int.from_bytes(data[i + 10:i + 10 + size], "big", signed=False) + + if not self._is_plausible(mapping, raw_value): + self.logger.warning( + "Zahozena nepravděpodobná hodnota OBIS %s: raw=%s (podezření na poškozený rámec)", + full, raw_value, + ) + return None + + return obis_raw, raw_value, 10 + size + + def _is_plausible(self, mapping: dict, raw_value: int) -> bool: + cap = MAX_PLAUSIBLE_BY_DEVICE_CLASS.get(mapping.get("device_class")) + if cap is None: + return True + return 0 <= raw_value <= cap + def extract_serial(self, data: bytes) -> str | None: + # OBIS 0-0:96.1.0.255 (idx 3 v EG.D dokumentaci) — čti hodnotu obecně + # podle OBIS kódu, ne podle textového prefixu konkrétního výrobce + # (starší kód hledal natvrdo b"SAG", což fungovalo jen pro Sagemcom). + value = self._find_data_string(data, "96.1.0") + if value is not None: + return value + # Fallback pro starší/jiné meterové profily se Sagemcom prefixem. idx = data.find(b"SAG") if idx == -1: return None return data[idx:idx + 16].decode("ascii", errors="ignore") def extract_tariff(self, data: bytes) -> str | None: - for tariff in [b"T1", b"T2", b"T3", b"T4"]: + # OBIS 0-0:96.14.0.255 (idx 12 v EG.D dokumentaci) — čti hodnotu obecně + # podle OBIS kódu. Starší kód hledal natvrdo velká písmena b"T1".."T4", + # ale meter může posílat i malá písmena (pozorováno "t3"), proto to + # dřív u tohoto typu meteru nikdy nenašlo shodu. + value = self._find_data_string(data, "96.14.0") + if value is not None: + return value + # Fallback na starší pevný seznam pro meterové profily bez OBIS shody. + for tariff in [b"T1", b"T2", b"T3", b"T4", b"t1", b"t2", b"t3", b"t4"]: if tariff in data: return tariff.decode("ascii") return None - def _looks_like_register_value(self, data: bytes, i: int) -> bool: - return ( - i + 16 <= len(data) - and data[i] == 0x02 - and data[i + 1] == 0x02 - and int.from_bytes(data[i + 2:i + 4], "big") == 3 - and data[i + 10] == 0x02 - and data[i + 11] == 0x06 - ) + def _find_data_string(self, data: bytes, obis_short_target: str) -> str | None: + """Projde rámec a hledá class 1 (Data) octet-string objekty podle OBIS kódu. + + Struktura (empiricky ověřeno na živém rámci 2026-08-22): + 02 02 [class_id:2]=00 01 [obis:6] [attribute:1] [axdr_type:1]=0x09 [length:1] [value:length bajtů] + """ + i = 0 + while i < len(data): + if self._looks_like_data_string_value(data, i): + obis_raw = data[i + 4:i + 10] + length = data[i + 12] + if self._obis_short(obis_raw) == obis_short_target: + value = data[i + 13:i + 13 + length] + text = value.decode("ascii", errors="ignore").strip("\x00").strip() + return text or None + i += 13 + length + continue + i += 1 + return None + + def _looks_like_data_string_value(self, data: bytes, i: int) -> bool: + if i + 13 > len(data): + return False + if data[i] != 0x02 or data[i + 1] != 0x02: + return False + class_id = int.from_bytes(data[i + 2:i + 4], "big") + if class_id != 1: + return False + if data[i + 11] != 0x09: # AXDR octet-string type tag + return False + length = data[i + 12] + return i + 13 + length <= len(data) def _looks_like_enum_value(self, data: bytes, i: int) -> bool: if i + 13 > len(data):