Add telegram() method to both API versions (#430) (#491)

* fix typo * Add telegram method (#430) Return type is `Any` because we don't process the response from the API. * add an example usage for the telegram method * Method telegram() only works on P1 meter * Add check if device supports the telegram() method * Fix typo * Fix formatting (double newline before class) * Apply suggestions from code review * Update return type to 'str * Add some tests * Add some tests * Fix all pre-commit issues --------- Co-authored-by: Duco Sebel <74970928+DCSBL@users.noreply.github.com>
homewizard · Feb 25, 2025 · 5dd8261 · 5dd8261
1 parent bd94f32
commit 5dd8261
Show file tree

Hide file tree

Showing 14 changed files with 288 additions and 5 deletions.
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -167,6 +167,7 @@ repos:
         types: [text]
         entry: poetry run trailing-whitespace-fixer
         stages: [commit, push, manual]
+        exclude: ^.*\.ambr$
 
       - id: vulture
         name: Find unused Python code with Vulture

diff --git a/README.md b/README.md
@@ -39,6 +39,10 @@ async def main():
         measurement = await api.measurement()
         print(measurement.energy_import_kwh)
 
+        # Example of getting raw telegram data
+        telegram = await api.telegram()
+        print(telegram)  # Raw P1 meter data
+
         # Get all data and remap v1 data to new v2 structure
         print(await api.combined())
 

diff --git a/homewizard_energy/homewizard_energy.py b/homewizard_energy/homewizard_energy.py
@@ -82,6 +82,10 @@ async def measurement(self) -> Measurement:
         """Get the current measurement."""
         raise NotImplementedError
 
+    async def telegram(self) -> str:
+        """Get the latest telegram."""
+        raise NotImplementedError
+
     async def system(
         self,
         cloud_enabled: bool | None = None,

diff --git a/homewizard_energy/models.py b/homewizard_energy/models.py
@@ -142,7 +142,7 @@ def supports_cloud_enable(self) -> bool:
 
     def supports_reboot(self) -> bool:
         """Return if the device supports reboot."""
-        return self.api_version.major >= 2
+        return self.api_version.major >= 2 and self.product_type != Model.BATTERY
 
     def supports_identify(self) -> bool:
         """Return if the device supports identify."""
@@ -153,6 +153,10 @@ def supports_identify(self) -> bool:
             Model.ENERGY_METER_EASTRON_SDM630,
         )
 
+    def supports_telegram(self) -> bool:
+        """Return if the device supports telegram."""
+        return self.product_type == Model.P1_METER
+
 
 @dataclass(kw_only=True)
 class Measurement(BaseModel):

diff --git a/homewizard_energy/v1/__init__.py b/homewizard_energy/v1/__init__.py
@@ -56,6 +56,18 @@ async def measurement(self) -> Measurement:
         _, response = await self._request("api/v1/data")
         return Measurement.from_json(response)
 
+    @optional_method
+    async def telegram(self) -> str:
+        """Return the most recent, valid telegram that was given by the device.
+        The telegram is not processed in any other form.
+        If you need parsed data, use the measurement method.
+        """
+        if self._device is not None and self._device.supports_telegram() is False:
+            raise UnsupportedError("Telegram is not supported")
+
+        _, telegram = await self._request("api/v1/telegram")
+        return telegram
+
     @optional_method
     async def system(
         self,

diff --git a/homewizard_energy/v2/__init__.py b/homewizard_energy/v2/__init__.py
@@ -95,6 +95,15 @@ async def measurement(self) -> Measurement:
 
         return measurement
 
+    @authorized_method
+    async def telegram(self) -> str:
+        """Return the most recent, valid telegram that was given by the device.
+        The telegram is not processed in any other form.
+        If you need parsed data, use the measurement method.
+        """
+        _, telegram = await self._request("/api/telegram")
+        return telegram
+
     @authorized_method
     async def system(
         self,

diff --git a/tests/test_homewizard_energy.py b/tests/test_homewizard_energy.py
@@ -13,6 +13,7 @@
     [
         ("device", NotImplementedError),
         ("measurement", NotImplementedError),
+        ("telegram", NotImplementedError),
         ("system", NotImplementedError),
         ("state", UnsupportedError),
         ("identify", NotImplementedError),

diff --git a/tests/v1/__snapshots__/test_v1_homewizard_energy.ambr b/tests/v1/__snapshots__/test_v1_homewizard_energy.ambr
@@ -167,3 +167,46 @@
 # name: test_system_set[SDM630-wifi]
   System(wifi_strength_pct=None, wifi_ssid=None, wifi_rssi_db=None, cloud_enabled=True, uptime_s=None, status_led_brightness_pct=None, api_v1_enabled=None)
 # ---
+# name: test_telegram_gets_latest_telegram
+  '''
+  /ISK5\\2M550T-10111-
+
+  3:0.2.8(50)
+  0-0:1.0.0(181106140429W)
+  0-0:96.1.1(31333631353032362020202020202020)
+  1-0:1.8.1(10830.511*kWh)
+  1-0:1.8.2(002948.827*kWh)
+  1-0:2.8.1(001285.951*kWh)
+  1-0:2.8.2(002876.514*kWh)
+  0-0:96.14.0(0002)
+  1-0:1.7.0(21.100*kW)
+  1-0:2.7.0(00.000*kW)
+  0-0:96.7.21(00006)
+  0-0:96.7.9(00003)
+  1-0:99.97.0(1)(0-0:96.7.19)(180529135630S)(0000002451*s)
+  1-0:32.32.0(00003)
+  1-0:52.32.0(00002)
+  1-0:72.32.0(00002)
+  1-0:32.36.0(00001)
+  1-0:52.36.0(00001)
+  1-0:72.36.0(00001)
+  0-0:96.13.0()
+  1-0:32.7.0(236.0*V)
+  1-0:52.7.0(232.6*V)
+  1-0:72.7.0(235.1*V)
+  1-0:31.7.0(002*A)
+  1-0:51.7.0(000*A)
+  1-0:71.7.0(000*A)
+  1-0:21.7.0(00.000*kW)
+  1-0:41.7.0(00.033*kW)
+  1-0:61.7.0(00.132*kW)
+  1-0:22.7.0(00.676*kW)
+  1-0:42.7.0(00.000*kW)
+  1-0:62.7.0(00.000*kW)
+  0-1:24.1.0(003)
+  0-1:96.1.0(4730303339303031373030343630313137)
+  0-1:24.2.1(210606140010W)(02569.646*m3)
+  !1F28
+
+  '''
+# ---
diff --git a/tests/v1/fixtures/HWE-P1/telegram.txt b/tests/v1/fixtures/HWE-P1/telegram.txt
@@ -0,0 +1,38 @@
+/ISK5\\2M550T-10111-
+
+3:0.2.8(50)
+0-0:1.0.0(181106140429W)
+0-0:96.1.1(31333631353032362020202020202020)
+1-0:1.8.1(10830.511*kWh)
+1-0:1.8.2(002948.827*kWh)
+1-0:2.8.1(001285.951*kWh)
+1-0:2.8.2(002876.514*kWh)
+0-0:96.14.0(0002)
+1-0:1.7.0(21.100*kW)
+1-0:2.7.0(00.000*kW)
+0-0:96.7.21(00006)
+0-0:96.7.9(00003)
+1-0:99.97.0(1)(0-0:96.7.19)(180529135630S)(0000002451*s)
+1-0:32.32.0(00003)
+1-0:52.32.0(00002)
+1-0:72.32.0(00002)
+1-0:32.36.0(00001)
+1-0:52.36.0(00001)
+1-0:72.36.0(00001)
+0-0:96.13.0()
+1-0:32.7.0(236.0*V)
+1-0:52.7.0(232.6*V)
+1-0:72.7.0(235.1*V)
+1-0:31.7.0(002*A)
+1-0:51.7.0(000*A)
+1-0:71.7.0(000*A)
+1-0:21.7.0(00.000*kW)
+1-0:41.7.0(00.033*kW)
+1-0:61.7.0(00.132*kW)
+1-0:22.7.0(00.676*kW)
+1-0:42.7.0(00.000*kW)
+1-0:62.7.0(00.000*kW)
+0-1:24.1.0(003)
+0-1:96.1.0(4730303339303031373030343630313137)
+0-1:24.2.1(210606140010W)(02569.646*m3)
+!1F28
diff --git a/tests/v1/test_v1_homewizard_energy.py b/tests/v1/test_v1_homewizard_energy.py
@@ -740,6 +740,28 @@ async def test_identify_not_supported_with_cached_device(aresponses, model: str)
         await api.close()
 
 
+async def test_telegram_gets_latest_telegram(aresponses, snapshot: SnapshotAssertion):
+    """Telegram returns latest telegram."""
+
+    aresponses.add(
+        "example.com",
+        "/api/v1/telegram",
+        "GET",
+        aresponses.Response(
+            text=load_fixtures("HWE-P1/telegram.txt"),
+            status=200,
+            headers={"Content-Type": "application/txt; charset=utf-8"},
+        ),
+    )
+
+    async with aiohttp.ClientSession() as session:
+        api = HomeWizardEnergyV1("example.com", clientsession=session)
+
+        telegram = await api.telegram()
+        assert telegram
+        assert telegram == snapshot
+
+
 async def test_get_system_object(aresponses):
     """Test fetches system object and device object when unknown."""
 

diff --git a/tests/v2/__snapshots__/test_v2_homewizard_energy.ambr b/tests/v2/__snapshots__/test_v2_homewizard_energy.ambr
@@ -23,3 +23,46 @@
 # name: test_system_with_valid_authentication[HWE-P1-fixtures0]
   System(wifi_strength_pct=46, wifi_ssid='My Wi-Fi', wifi_rssi_db=-77, cloud_enabled=False, uptime_s=356, status_led_brightness_pct=100, api_v1_enabled=True)
 # ---
+# name: test_telegram_with_valid_authentication
+  '''
+  /ISK5\\2M550T-10111-
+
+  3:0.2.8(50)
+  0-0:1.0.0(181106140429W)
+  0-0:96.1.1(31333631353032362020202020202020)
+  1-0:1.8.1(10830.511*kWh)
+  1-0:1.8.2(002948.827*kWh)
+  1-0:2.8.1(001285.951*kWh)
+  1-0:2.8.2(002876.514*kWh)
+  0-0:96.14.0(0002)
+  1-0:1.7.0(21.100*kW)
+  1-0:2.7.0(00.000*kW)
+  0-0:96.7.21(00006)
+  0-0:96.7.9(00003)
+  1-0:99.97.0(1)(0-0:96.7.19)(180529135630S)(0000002451*s)
+  1-0:32.32.0(00003)
+  1-0:52.32.0(00002)
+  1-0:72.32.0(00002)
+  1-0:32.36.0(00001)
+  1-0:52.36.0(00001)
+  1-0:72.36.0(00001)
+  0-0:96.13.0()
+  1-0:32.7.0(236.0*V)
+  1-0:52.7.0(232.6*V)
+  1-0:72.7.0(235.1*V)
+  1-0:31.7.0(002*A)
+  1-0:51.7.0(000*A)
+  1-0:71.7.0(000*A)
+  1-0:21.7.0(00.000*kW)
+  1-0:41.7.0(00.033*kW)
+  1-0:61.7.0(00.132*kW)
+  1-0:22.7.0(00.676*kW)
+  1-0:42.7.0(00.000*kW)
+  1-0:62.7.0(00.000*kW)
+  0-1:24.1.0(003)
+  0-1:96.1.0(4730303339303031373030343630313137)
+  0-1:24.2.1(210606140010W)(02569.646*m3)
+  !1F28
+
+  '''
+# ---
diff --git a/tests/v2/fixtures/HWE-P1/telegram.txt b/tests/v2/fixtures/HWE-P1/telegram.txt
@@ -0,0 +1,38 @@
+/ISK5\\2M550T-10111-
+
+3:0.2.8(50)
+0-0:1.0.0(181106140429W)
+0-0:96.1.1(31333631353032362020202020202020)
+1-0:1.8.1(10830.511*kWh)
+1-0:1.8.2(002948.827*kWh)
+1-0:2.8.1(001285.951*kWh)
+1-0:2.8.2(002876.514*kWh)
+0-0:96.14.0(0002)
+1-0:1.7.0(21.100*kW)
+1-0:2.7.0(00.000*kW)
+0-0:96.7.21(00006)
+0-0:96.7.9(00003)
+1-0:99.97.0(1)(0-0:96.7.19)(180529135630S)(0000002451*s)
+1-0:32.32.0(00003)
+1-0:52.32.0(00002)
+1-0:72.32.0(00002)
+1-0:32.36.0(00001)
+1-0:52.36.0(00001)
+1-0:72.36.0(00001)
+0-0:96.13.0()
+1-0:32.7.0(236.0*V)
+1-0:52.7.0(232.6*V)
+1-0:72.7.0(235.1*V)
+1-0:31.7.0(002*A)
+1-0:51.7.0(000*A)
+1-0:71.7.0(000*A)
+1-0:21.7.0(00.000*kW)
+1-0:41.7.0(00.033*kW)
+1-0:61.7.0(00.132*kW)
+1-0:22.7.0(00.676*kW)
+1-0:42.7.0(00.000*kW)
+1-0:62.7.0(00.000*kW)
+0-1:24.1.0(003)
+0-1:96.1.0(4730303339303031373030343630313137)
+0-1:24.2.1(210606140010W)(02569.646*m3)
+!1F28
diff --git a/tests/v2/test_v2_homewizard_energy.py b/tests/v2/test_v2_homewizard_energy.py
@@ -384,6 +384,58 @@ async def test_measurement_with_valid_authentication(
             assert measurement == snapshot
 
 
+### Telegram tests ###
+
+
+async def test_telegram_without_authentication():
+    """Test telegram request is rejected when no authentication is provided."""
+
+    async with HomeWizardEnergyV2("example.com") as api:
+        with pytest.raises(UnauthorizedError):
+            await api.telegram()
+
+
+async def test_telegram_with_invalid_authentication(aresponses):
+    """Test telegram request is unsuccessful when invalid authentication is provided."""
+
+    aresponses.add(
+        "example.com",
+        "/api/telegram",
+        "GET",
+        aresponses.Response(
+            status=401,
+            headers={"Content-Type": "application/text"},
+            text='{"error": "user:unauthorized"}',
+        ),
+    )
+
+    async with HomeWizardEnergyV2("example.com", token="token") as api:
+        with pytest.raises(UnauthorizedError):
+            await api.telegram()
+
+
+async def test_telegram_with_valid_authentication(
+    snapshot: SnapshotAssertion, aresponses
+):
+    """Test telegram request is successful when valid authentication is provided."""
+
+    aresponses.add(
+        "example.com",
+        "/api/telegram",
+        "GET",
+        aresponses.Response(
+            text=load_fixtures("HWE-P1/telegram.txt"),
+            status=200,
+            headers={"Content-Type": "application/txt"},
+        ),
+    )
+
+    async with HomeWizardEnergyV2("example.com", token="token") as api:
+        telegram = await api.telegram()
+        assert telegram is not None
+        assert telegram == snapshot
+
+
 ### System tests ###
 
 

diff --git a/tests/v2/test_v2_models.py b/tests/v2/test_v2_models.py
@@ -120,18 +120,29 @@ async def test_system_update_raises_when_none_set():
         update.to_dict()
 
 
+# pylint: disable=too-many-arguments
+# pylint: disable=too-many-positional-arguments
 @pytest.mark.parametrize(
-    ("model", "supports_state", "supports_identify", "supports_cloud_enable"),
+    (
+        "model",
+        "supports_state",
+        "supports_identify",
+        "supports_cloud_enable",
+        "supports_reboot",
+        "supports_telegram",
+    ),
     [
-        ("HWE-P1", False, True, True),
-        ("HWE-BAT", False, True, False),
+        ("HWE-P1", False, True, True, True, True),
+        ("HWE-BAT", False, True, False, False, False),
     ],
 )
 async def test_device_support_functions(
     model: str,
     supports_state: bool,
     supports_identify: bool,
     supports_cloud_enable: bool,
+    supports_reboot: bool,
+    supports_telegram: bool,
 ):
     """Test Device model support functions."""
     device = Device.from_dict(json.loads(load_fixtures(f"{model}/device.json")))
@@ -140,4 +151,5 @@ async def test_device_support_functions(
     assert device.supports_state() == supports_state
     assert device.supports_identify() == supports_identify
     assert device.supports_cloud_enable() == supports_cloud_enable
-    assert device.supports_reboot() is True  # Always True for v2
+    assert device.supports_reboot() == supports_reboot
+    assert device.supports_telegram() == supports_telegram