diff --git a/README.md b/README.md index 7d2731e..019aa9c 100644 --- a/README.md +++ b/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 diff --git a/config.ini.example b/config.ini.example index b9735c0..0353c99 100644 --- a/config.ini.example +++ b/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] diff --git a/config/config_loader.py b/config/config_loader.py index 1d1c592..6728bf0 100644 --- a/config/config_loader.py +++ b/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), + ) diff --git a/config/config_loader_test.py b/config/config_loader_test.py index ad86099..1bea407 100644 --- a/config/config_loader_test.py +++ b/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 diff --git a/powermeter/__init__.py b/powermeter/__init__.py index d366497..3fa91af 100644 --- a/powermeter/__init__.py +++ b/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 diff --git a/powermeter/tq_em.py b/powermeter/tq_em.py new file mode 100644 index 0000000..93b06c5 --- /dev/null +++ b/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 diff --git a/powermeter/tq_em_test.py b/powermeter/tq_em_test.py new file mode 100644 index 0000000..2d055c3 --- /dev/null +++ b/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()