Add second power OBIS (#4)

We added this second power OBIS to the function, because some meter incorrectly assign the power over all 3 phases to the OBIS for power on phase 1 (L1). If the total power is not present we try the next OBIS.

Author: jukrebs

iometer/reading.py

@@ -46,6 +46,7 @@ class Reading:
     TOTAL_CONSUMPTION_OBIS = "01-00:01.08.00*ff"
     TOTAL_PRODUCTION_OBIS = "01-00:02.08.00*ff"
     CURRENT_POWER_OBIS = "01-00:10.07.00*ff"
+    CURRENT_POWER_OBIS_ALT = "01-00:24.07.00*ff"
 
     @classmethod
     def from_json(cls, json_str: str) -> "Reading":
@@ -104,10 +105,21 @@ def get_total_production(self) -> float:
         register = self.meter.reading.get_register_by_obis(self.TOTAL_PRODUCTION_OBIS)
         return register.value if register else 0
 
-    def get_current_power(self) -> float:
-        """Get current power consumption in W."""
+    def get_current_power(self) -> float | None:
+        """Get current power consumption in W.
+
+        Prefer 10.07.00 OBIS, it is missing in some meters.
+        These meters falsely report 24.07.00 OBIS as current power.
+
+        Returns None if neither OBIS is found, otherwise the value in W as float.
+        """
         register = self.meter.reading.get_register_by_obis(self.CURRENT_POWER_OBIS)
-        return register.value if register else 0
+        if register:
+            return register.value
+        register_alt = self.meter.reading.get_register_by_obis(
+            self.CURRENT_POWER_OBIS_ALT
+        )
+        return register_alt.value if register_alt else None
 
     def __str__(self) -> str:
         return self.to_json()

tests/test.py

@@ -31,6 +31,43 @@ def reading_json_fixture():
     }
 
 
+@pytest.fixture(name="reading_alt_obis_json")
+def reading_alt_obis_json_fixture():
+    """Fixture reading response with alternate current power OBIS only."""
+    return {
+        "__typename": "iometer.reading.v1",
+        "meter": {
+            "number": "1ISK0000000000",
+            "reading": {
+                "time": "2024-11-11T11:11:11Z",
+                "registers": [
+                    {"obis": "01-00:01.08.00*ff", "value": 1111.1, "unit": "Wh"},
+                    {"obis": "01-00:02.08.00*ff", "value": 2222.2, "unit": "Wh"},
+                    {"obis": "01-00:24.07.00*ff", "value": 200, "unit": "W"},
+                ],
+            },
+        },
+    }
+
+
+@pytest.fixture(name="reading_no_power_obis_json")
+def reading_no_power_obis_json_fixture():
+    """Fixture reading response without any current power OBIS registers."""
+    return {
+        "__typename": "iometer.reading.v1",
+        "meter": {
+            "number": "1ISK0000000000",
+            "reading": {
+                "time": "2024-11-11T11:11:11Z",
+                "registers": [
+                    {"obis": "01-00:01.08.00*ff", "value": 3333.3, "unit": "Wh"},
+                    {"obis": "01-00:02.08.00*ff", "value": 4444.4, "unit": "Wh"},
+                ],
+            },
+        },
+    }
+
+
 @pytest.fixture(name="status_json")
 def status_json_fixture():
     """ "Fixture status response"""
@@ -184,6 +221,32 @@ async def test_get_current_reading(client_iometer, mock_aioresponse, reading_jso
     assert reading.get_current_power() == 100
 
 
+@pytest.mark.asyncio
+async def test_get_current_reading_alt_obis(
+    client_iometer, mock_aioresponse, reading_alt_obis_json
+):
+    """Test current power using alternate OBIS when primary is missing."""
+    mock_endpoint = f"http://{HOST}/v1/reading"
+    mock_aioresponse.get(mock_endpoint, status=200, payload=reading_alt_obis_json)
+
+    reading = await client_iometer.get_current_reading()
+    assert isinstance(reading, Reading)
+    assert reading.get_current_power() == 200
+
+
+@pytest.mark.asyncio
+async def test_get_current_reading_no_power_obis(
+    client_iometer, mock_aioresponse, reading_no_power_obis_json
+):
+    """Test current power returns None when no power OBIS registers are present."""
+    mock_endpoint = f"http://{HOST}/v1/reading"
+    mock_aioresponse.get(mock_endpoint, status=200, payload=reading_no_power_obis_json)
+
+    reading = await client_iometer.get_current_reading()
+    assert isinstance(reading, Reading)
+    assert reading.get_current_power() is None
+
+
 @pytest.mark.asyncio
 async def test_get_current_status(client_iometer, mock_aioresponse, status_json):
     """Test getting device status."""