Implement Powermeter TQ EM

README.md

@@ -381,6 +381,15 @@ PASSWORD = pass (Optional)
 HEADERS = Authorization: Bearer token
 ```
 
+### TQ Energy Manager
+
+```ini
+[TQ_EM]
+IP = 192.168.1.100
+#PASSWORD = pass
+#TIMEOUT = 5.0 (Optional)
+```
+
 ### Modbus
 
 ```ini

config.ini.example

@@ -134,6 +134,11 @@ THROTTLE_INTERVAL = 0
 #USERNAME = user
 #PASSWORD = pass
 #HEADERS = Authorization: Bearer token
+
+#[TQ_EM]
+#IP = 192.168.1.100
+#PASSWORD = secret (Optional)
+#TIMEOUT = 5.0 (Optional)
 #THROTTLE_INTERVAL = 1
 
 #[SCRIPT]

config/config_loader.py

@@ -22,6 +22,7 @@
     Script,
     ESPHome,
     JsonHttpPowermeter,
+    TQEnergyManager,
     ThrottledPowermeter,
 )
 
@@ -37,6 +38,7 @@
 AMIS_READER_SECTION = "AMIS_READER"
 MODBUS_SECTION = "MODBUS"
 JSON_HTTP_SECTION = "JSON_HTTP"
+TQ_EM_SECTION = "TQ_EM"
 
 
 class ClientFilter:
@@ -119,6 +121,8 @@ def create_powermeter(
         return create_amisreader_powermeter(section, config)
     elif section.startswith(MODBUS_SECTION):
         return create_modbus_powermeter(section, config)
+    elif section.startswith(TQ_EM_SECTION):
+        return create_tq_em_powermeter(section, config)
     elif section.startswith(JSON_HTTP_SECTION):
         return create_json_http_powermeter(section, config)
     elif section.startswith("MQTT"):
@@ -319,3 +323,13 @@ def create_tasmota_powermeter(
         config.get(section, "JSON_POWER_OUTPUT_MQTT_LABEL", fallback=""),
         config.getboolean(section, "JSON_POWER_CALCULATE", fallback=False),
     )
+
+
+def create_tq_em_powermeter(
+    section: str, config: configparser.ConfigParser
+) -> Powermeter:
+    return TQEnergyManager(
+        config.get(section, "IP", fallback=""),
+        config.get(section, "PASSWORD", fallback=""),
+        timeout=config.getfloat(section, "TIMEOUT", fallback=5.0),
+    )

config/config_loader_test.py

@@ -19,6 +19,7 @@
     create_modbus_powermeter,
     create_mqtt_powermeter,
     create_json_http_powermeter,
+    create_tq_em_powermeter,
 )
 import unittest
 from unittest.mock import patch, Mock
@@ -232,6 +233,18 @@ def test_create_json_http_powermeter():
             raise
 
 
+def test_create_tq_em_powermeter():
+    """Test TQ Energy Manager powermeter creation."""
+    config = configparser.ConfigParser()
+    config["TQ_EM"] = {"IP": "127.0.0.1"}
+
+    try:
+        create_tq_em_powermeter("TQ_EM", config)
+    except Exception as e:
+        if "Connection" not in str(e):
+            raise
+
+
 def test_create_powermeter():
     """Test the main create_powermeter function."""
     config = configparser.ConfigParser()
@@ -250,6 +263,7 @@ def test_create_powermeter():
     config["MODBUS_TEST"] = {"HOST": "127.0.0.1"}
     config["MQTT_TEST"] = {"BROKER": "127.0.0.1"}
     config["JSON_HTTP_TEST"] = {"URL": "http://localhost", "JSON_PATHS": "$.power"}
+    config["TQ_EM_TEST"] = {"IP": "127.0.0.1"}
     config["UNKNOWN_TEST"] = {"SOME_KEY": "some_value"}
 
     # Test each powermeter type

powermeter/__init__.py

@@ -13,3 +13,4 @@
 from .json_http import JsonHttpPowermeter
 from .script import Script
 from .throttling import ThrottledPowermeter
+from .tq_em import TQEnergyManager

powermeter/tq_em.py

@@ -0,0 +1,96 @@
+from typing import List
+import time
+import requests
+
+from .base import Powermeter
+
+
+class TQEnergyManager(Powermeter):
+    """Powermeter using the TQ Energy Manager JSON API."""
+
+    # OBIS codes
+    _TOTAL_KEY = "1-0:1.4.0*255"  # Σ active power
+    _PHASE_KEYS = (
+        "1-0:21.4.0*255",  # L1
+        "1-0:41.4.0*255",  # L2
+        "1-0:61.4.0*255",  # L3
+    )
+
+    _MAX_IDLE = 60 * 30  # 30 min
+
+    def __init__(self, host: str, password: str = "", *, timeout: float = 5.0) -> None:
+        self._host, self._pw, self._timeout = host.rstrip("/"), password, timeout
+        self._sess = requests.Session()
+        self._serial: str | None = None
+        self._last_use = 0.0
+
+    # ------------------------------------------------------------------ #
+    # PUBLIC                                                             #
+    # ------------------------------------------------------------------ #
+    def get_powermeter_watts(self) -> List[float]:
+        self._ensure_session()
+
+        try:
+            data = self._read_live_json()
+        except _SessionExpired:
+            self._login()
+            data = self._read_live_json()
+
+        if all(k in data for k in self._PHASE_KEYS):
+            return [float(data[k]) for k in self._PHASE_KEYS]
+        if self._TOTAL_KEY in data:
+            return [float(data[self._TOTAL_KEY])]
+
+        raise RuntimeError("Required OBIS values missing in payload")
+
+    # ------------------------------------------------------------------ #
+    # INTERNALS                                                          #
+    # ------------------------------------------------------------------ #
+    def _ensure_session(self) -> None:
+        now = time.time()
+        if self._serial is None or (now - self._last_use) > self._MAX_IDLE:
+            self._login()
+        self._last_use = now
+
+    def _login(self) -> None:
+        """Authenticate lazily with the device."""
+        r1 = self._sess.get(f"http://{self._host}/start.php", timeout=self._timeout)
+        r1.raise_for_status()
+        j1 = r1.json()
+
+        self._serial = j1.get("serial") or j1.get("ieq_serial")
+        if not self._serial:
+            raise RuntimeError("Serial number missing in /start.php response")
+
+        if j1.get("authentication") is True:
+            return
+
+        payload = {"login": self._serial, "save_login": 1}
+        if self._pw:
+            payload["password"] = self._pw
+
+        r2 = self._sess.post(
+            f"http://{self._host}/start.php", data=payload, timeout=self._timeout
+        )
+        r2.raise_for_status()
+        if r2.json().get("authentication") is not True:
+            raise RuntimeError("Authentication failed")
+
+    def _read_live_json(self) -> dict:
+        r = self._sess.get(
+            f"http://{self._host}/mum-webservice/data.php", timeout=self._timeout
+        )
+        if r.status_code in (401, 403):
+            raise _SessionExpired
+
+        r.raise_for_status()
+        data = r.json()
+        if data.get("status", 0) >= 900:
+            raise _SessionExpired
+        return data
+
+
+class _SessionExpired(RuntimeError):
+    """Internal marker – triggers transparent re-login."""
+
+    pass

powermeter/tq_em_test.py

@@ -0,0 +1,71 @@
+import unittest
+from unittest.mock import patch, MagicMock
+
+from powermeter.tq_em import TQEnergyManager
+
+
+class TestTQEnergyManager(unittest.TestCase):
+    @patch("requests.Session.post")
+    @patch("requests.Session.get")
+    def test_three_phase(self, mock_get, mock_post):
+        # login GET
+        mock_get.side_effect = [
+            MagicMock(
+                status_code=200, json=lambda: {"serial": "123", "authentication": False}
+            ),
+            MagicMock(
+                status_code=200,
+                json=lambda: {
+                    "1-0:21.4.0*255": 1,
+                    "1-0:41.4.0*255": 2,
+                    "1-0:61.4.0*255": 3,
+                },
+            ),
+        ]
+        mock_post.return_value = MagicMock(
+            status_code=200, json=lambda: {"authentication": True}
+        )
+
+        meter = TQEnergyManager("192.168.0.10")
+        self.assertEqual(meter.get_powermeter_watts(), [1.0, 2.0, 3.0])
+
+    @patch("requests.Session.post")
+    @patch("requests.Session.get")
+    def test_total_only(self, mock_get, mock_post):
+        mock_get.side_effect = [
+            MagicMock(
+                status_code=200, json=lambda: {"serial": "321", "authentication": False}
+            ),
+            MagicMock(status_code=200, json=lambda: {"1-0:1.4.0*255": 9}),
+        ]
+        mock_post.return_value = MagicMock(
+            status_code=200, json=lambda: {"authentication": True}
+        )
+
+        meter = TQEnergyManager("192.168.0.12")
+        self.assertEqual(meter.get_powermeter_watts(), [9.0])
+
+    @patch("requests.Session.post")
+    @patch("requests.Session.get")
+    def test_relogin_on_expired_session(self, mock_get, mock_post):
+        mock_get.side_effect = [
+            MagicMock(
+                status_code=200, json=lambda: {"serial": "123", "authentication": False}
+            ),
+            MagicMock(status_code=200, json=lambda: {"status": 901}),
+            MagicMock(
+                status_code=200, json=lambda: {"serial": "123", "authentication": False}
+            ),
+            MagicMock(status_code=200, json=lambda: {"1-0:1.4.0*255": 5}),
+        ]
+        mock_post.side_effect = [
+            MagicMock(status_code=200, json=lambda: {"authentication": True}),
+            MagicMock(status_code=200, json=lambda: {"authentication": True}),
+        ]
+
+        meter = TQEnergyManager("192.168.0.11")
+        self.assertEqual(meter.get_powermeter_watts(), [5.0])
+
+
+if __name__ == "__main__":
+    unittest.main()