Add files via upload

custom_components/xt211_han/manifest.json

@@ -1,7 +1,7 @@
 {
   "domain": "xt211_han",
   "name": "XT211 HAN (RS485 via Ethernet)",
-  "version": "0.7.6",
+  "version": "0.7.7",
   "documentation": "https://github.com/nero150/xt211-han-ha",
   "issue_tracker": "https://github.com/nero150/xt211-han-ha/issues",
   "dependencies": [],

custom_components/xt211_han/sensor.py

@@ -41,14 +41,23 @@ SENSOR_META: dict[str, dict] = {
     },
 }
 
-TEXT_OBIS = {
+SERIAL_OBIS = ("0-0:96.1.1.255", "0-0:96.1.0.255")
+PRECREATED_TEXT_ENTITIES = {
+    "serial_number": {
+        "name": "Výrobní číslo",
+        "obises": SERIAL_OBIS,
+        "entity_category": EntityCategory.DIAGNOSTIC,
+    },
+    "current_tariff": {
+        "name": "Aktuální tarif",
+        "obises": ("0-0:96.14.0.255",),
+        "entity_category": EntityCategory.DIAGNOSTIC,
+    },
+}
+DYNAMIC_TEXT_OBIS = {
     "0-0:42.0.0.255",
-    "0-0:96.1.0.255",
-    "0-0:96.1.1.255",
-    "0-0:96.14.0.255",
     "0-0:96.13.0.255",
 }
-
 BINARY_OBIS = {
     "0-0:96.3.10.255",
     "0-1:96.3.10.255",
@@ -58,6 +67,7 @@ BINARY_OBIS = {
     "0-5:96.3.10.255",
     "0-6:96.3.10.255",
 }
+TEXT_OBIS = set().union(*(spec["obises"] for spec in PRECREATED_TEXT_ENTITIES.values()), DYNAMIC_TEXT_OBIS)
 
 
 def _device_info(entry: ConfigEntry) -> DeviceInfo:
@@ -76,15 +86,10 @@ def build_enabled_obis(entry: ConfigEntry) -> set[str]:
     relay_count = int(entry.data.get(CONF_RELAY_COUNT, RELAYS_4))
 
     enabled_obis: set[str] = {
-        "0-0:42.0.0.255",
-        "0-0:96.1.0.255",
-        "0-0:96.1.1.255",
-        "0-0:96.14.0.255",
-        "0-0:96.13.0.255",
-        "0-0:96.3.10.255",
         "0-0:17.0.0.255",
         "1-0:1.7.0.255",
         "1-0:1.8.0.255",
+        *TEXT_OBIS,
     }
 
     relay_obis = {
@@ -95,6 +100,7 @@ def build_enabled_obis(entry: ConfigEntry) -> set[str]:
         5: "0-5:96.3.10.255",
         6: "0-6:96.3.10.255",
     }
+    enabled_obis.add("0-0:96.3.10.255")
     for idx in range(1, relay_count + 1):
         enabled_obis.add(relay_obis[idx])
 
@@ -124,11 +130,15 @@ async def async_setup_entry(
     entities = [
         XT211SensorEntity(coordinator, entry, obis, meta)
         for obis, meta in OBIS_DESCRIPTIONS.items()
-        if obis in enabled_obis and obis not in BINARY_OBIS
+        if obis in enabled_obis and obis not in BINARY_OBIS and obis not in TEXT_OBIS
     ]
+    entities.extend(
+        XT211AliasedTextSensorEntity(coordinator, entry, key, spec)
+        for key, spec in PRECREATED_TEXT_ENTITIES.items()
+    )
     async_add_entities(entities)
 
-    registered_obis = {entity._obis for entity in entities}
+    registered_obis = {entity._obis for entity in entities if hasattr(entity, "_obis")}
 
     @callback
     def _on_update() -> None:
@@ -138,6 +148,12 @@ async def async_setup_entry(
         for obis, data in coordinator.data.items():
             if obis in registered_obis or obis not in enabled_obis or obis in BINARY_OBIS:
                 continue
+            if obis in DYNAMIC_TEXT_OBIS:
+                registered_obis.add(obis)
+                new_entities.append(XT211DynamicTextSensorEntity(coordinator, entry, obis, data))
+                continue
+            if obis in TEXT_OBIS:
+                continue
             registered_obis.add(obis)
             new_entities.append(XT211SensorEntity(coordinator, entry, obis, data))
         if new_entities:
@@ -156,14 +172,11 @@ class XT211SensorEntity(CoordinatorEntity[XT211Coordinator], SensorEntity):
         self._entry = entry
         self._obis = obis
         self._wh_to_kwh = sensor_type == "energy"
-        self._text = obis in TEXT_OBIS
         self._attr_unique_id = f"{entry.entry_id}_{obis}"
         self._attr_name = meta.get("name", obis)
-        self._attr_device_class = None if self._text else sensor_meta["device_class"]
-        self._attr_state_class = None if self._text else sensor_meta["state_class"]
-        self._attr_native_unit_of_measurement = None if self._text else (sensor_meta["unit"] or meta.get("unit"))
-        if self._text:
-            self._attr_entity_category = EntityCategory.DIAGNOSTIC
+        self._attr_device_class = sensor_meta["device_class"]
+        self._attr_state_class = sensor_meta["state_class"]
+        self._attr_native_unit_of_measurement = sensor_meta["unit"] or meta.get("unit")
 
     @property
     def device_info(self) -> DeviceInfo:
@@ -175,8 +188,6 @@ class XT211SensorEntity(CoordinatorEntity[XT211Coordinator], SensorEntity):
         if obj is None:
             return None
         value = obj.get("value")
-        if self._text:
-            return None if value is None else str(value)
         try:
             number = float(value)
         except (TypeError, ValueError):
@@ -188,3 +199,59 @@ class XT211SensorEntity(CoordinatorEntity[XT211Coordinator], SensorEntity):
     @property
     def available(self) -> bool:
         return self.coordinator.data is not None
+
+
+class XT211AliasedTextSensorEntity(CoordinatorEntity[XT211Coordinator], SensorEntity):
+    _attr_has_entity_name = True
+
+    def __init__(self, coordinator: XT211Coordinator, entry: ConfigEntry, key: str, spec: dict) -> None:
+        super().__init__(coordinator)
+        self._entry = entry
+        self._obises = tuple(spec["obises"])
+        self._attr_unique_id = f"{entry.entry_id}_{key}"
+        self._attr_name = spec["name"]
+        self._attr_entity_category = spec.get("entity_category")
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        return _device_info(self._entry)
+
+    @property
+    def native_value(self) -> str | None:
+        data = self.coordinator.data or {}
+        for obis in self._obises:
+            obj = data.get(obis)
+            if obj and obj.get("value") is not None:
+                return str(obj.get("value"))
+        return None
+
+    @property
+    def available(self) -> bool:
+        return self.coordinator.data is not None
+
+
+class XT211DynamicTextSensorEntity(CoordinatorEntity[XT211Coordinator], SensorEntity):
+    _attr_has_entity_name = True
+    _attr_entity_category = EntityCategory.DIAGNOSTIC
+
+    def __init__(self, coordinator: XT211Coordinator, entry: ConfigEntry, obis: str, meta: dict) -> None:
+        super().__init__(coordinator)
+        self._entry = entry
+        self._obis = obis
+        self._attr_unique_id = f"{entry.entry_id}_{obis}"
+        self._attr_name = meta.get("name", obis)
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        return _device_info(self._entry)
+
+    @property
+    def native_value(self) -> str | None:
+        obj = (self.coordinator.data or {}).get(self._obis)
+        if obj is None or obj.get("value") is None:
+            return None
+        return str(obj.get("value"))
+
+    @property
+    def available(self) -> bool:
+        return self.coordinator.data is not None