Add JSON HTTP powermeter

README.md

@@ -365,9 +365,22 @@ PASSWORD = mqtt_pass (Optional)
 # THROTTLE_INTERVAL = 2
 ```
 
-The `JSON_PATH` option is used to extract the power value from a JSON payload. The path must be a [valid JSONPath expression](https://goessner.net/articles/JsonPath/). 
+The `JSON_PATH` option is used to extract the power value from a JSON payload. The path must be a [valid JSONPath expression](https://goessner.net/articles/JsonPath/).
 If the payload is a simple integer value, you can omit this option.
 
+### JSON HTTP
+
+```ini
+[JSON_HTTP]
+URL = http://example.com/api
+# Comma separated JSON paths - single path for 1-phase or three for 3-phase
+JSON_PATHS = $.power
+USERNAME = user (Optional)
+PASSWORD = pass (Optional)
+# Additional headers separated by ';' using 'Key: Value'
+HEADERS = Authorization: Bearer token
+```
+
 ### Modbus
 
 ```ini

config.ini.example

@@ -128,6 +128,14 @@ THROTTLE_INTERVAL = 0
 ## MQTT can have variable latency depending on broker and network
 #THROTTLE_INTERVAL = 1
 
+# [JSON_HTTP]
+#URL = http://example.com/api
+#JSON_PATHS = $.power
+#USERNAME = user
+#PASSWORD = pass
+#HEADERS = Authorization: Bearer token
+#THROTTLE_INTERVAL = 1
+
 #[SCRIPT]
 #COMMAND = /path/to/your/script.sh
 ## Per-powermeter throttling override (optional)

config/config_loader.py

@@ -21,6 +21,7 @@
     MqttPowermeter,
     Script,
     ESPHome,
+    JsonHttpPowermeter,
     ThrottledPowermeter,
 )
 
@@ -35,6 +36,7 @@
 ESPHOME_SECTION = "ESPHOME"
 AMIS_READER_SECTION = "AMIS_READER"
 MODBUS_SECTION = "MODBUS"
+JSON_HTTP_SECTION = "JSON_HTTP"
 
 
 class ClientFilter:
@@ -117,6 +119,8 @@ def create_powermeter(
         return create_amisreader_powermeter(section, config)
     elif section.startswith(MODBUS_SECTION):
         return create_modbus_powermeter(section, config)
+    elif section.startswith(JSON_HTTP_SECTION):
+        return create_json_http_powermeter(section, config)
     elif section.startswith("MQTT"):
         return create_mqtt_powermeter(section, config)
     else:
@@ -168,6 +172,36 @@ def create_mqtt_powermeter(
     )
 
 
+def create_json_http_powermeter(
+    section: str, config: configparser.ConfigParser
+) -> Powermeter:
+    json_paths = config.get(section, "JSON_PATHS", fallback="").split(",")
+    json_paths = [p.strip() for p in json_paths if p.strip()]
+    json_path_value = json_paths[0] if len(json_paths) == 1 else json_paths
+    return JsonHttpPowermeter(
+        config.get(section, "URL", fallback=""),
+        json_path_value,
+        config.get(section, "USERNAME", fallback=None),
+        config.get(section, "PASSWORD", fallback=None),
+        (
+            {
+                k.strip(): v.strip()
+                for k, v in (
+                    [
+                        item.split(":", 1)
+                        for item in config.get(section, "HEADERS", fallback="").split(
+                            ";"
+                        )
+                        if ":" in item
+                    ]
+                )
+            }
+            if config.get(section, "HEADERS", fallback="")
+            else None
+        ),
+    )
+
+
 def create_modbus_powermeter(
     section: str, config: configparser.ConfigParser
 ) -> Powermeter:

config/config_loader_test.py

@@ -18,6 +18,7 @@
     create_amisreader_powermeter,
     create_modbus_powermeter,
     create_mqtt_powermeter,
+    create_json_http_powermeter,
 )
 import unittest
 from unittest.mock import patch, Mock
@@ -219,6 +220,18 @@ def test_create_mqtt_powermeter():
             raise
 
 
+def test_create_json_http_powermeter():
+    """Test JSON HTTP powermeter creation."""
+    config = configparser.ConfigParser()
+    config["JSON_HTTP"] = {"URL": "http://localhost", "JSON_PATHS": "$.power"}
+
+    try:
+        create_json_http_powermeter("JSON_HTTP", 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()
@@ -236,6 +249,7 @@ def test_create_powermeter():
     config["AMIS_READER_TEST"] = {"IP": "127.0.0.1"}
     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["UNKNOWN_TEST"] = {"SOME_KEY": "some_value"}
 
     # Test each powermeter type

powermeter/__init__.py

@@ -10,5 +10,6 @@
 from .amisreader import AmisReader
 from .modbus import ModbusPowermeter
 from .mqtt import MqttPowermeter
+from .json_http import JsonHttpPowermeter
 from .script import Script
 from .throttling import ThrottledPowermeter

powermeter/json_http.py

@@ -0,0 +1,54 @@
+from .base import Powermeter
+import requests
+from typing import Union, List, Dict, Optional
+import json
+from jsonpath_ng import parse
+from requests.auth import HTTPBasicAuth
+from config.logger import logger
+
+
+def extract_json_value(data, path):
+    jsonpath_expr = parse(path)
+    match = jsonpath_expr.find(data)
+    if match:
+        return float(match[0].value)
+    else:
+        raise ValueError("No match found for the JSON path")
+
+
+class JsonHttpPowermeter(Powermeter):
+    def __init__(
+        self,
+        url: str,
+        json_path: Union[str, List[str]],
+        username: str = None,
+        password: str = None,
+        headers: Optional[Dict[str, str]] = None,
+    ):
+        self.url = url
+        self.json_paths = [json_path] if isinstance(json_path, str) else list(json_path)
+        self.auth = HTTPBasicAuth(username, password) if username or password else None
+        self.headers = headers or {}
+        self.session = requests.Session()
+
+    def get_json(self):
+        try:
+            response = self.session.get(
+                self.url, headers=self.headers, auth=self.auth, timeout=10
+            )
+            response.raise_for_status()
+            return response.json()
+        except json.JSONDecodeError as e:
+            logger.error(f"Failed to decode JSON: {e}")
+            logger.error(f"Response content: {response.text[:200]}...")
+            raise ValueError(f"Invalid JSON response: {e}")
+        except requests.exceptions.RequestException as e:
+            logger.error(f"HTTP request error: {e}")
+            raise ValueError(f"HTTP request error: {e}")
+
+    def get_powermeter_watts(self) -> List[float]:
+        data = self.get_json()
+        values = []
+        for path in self.json_paths:
+            values.append(extract_json_value(data, path))
+        return values

powermeter/json_http_test.py

@@ -0,0 +1,56 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from powermeter import JsonHttpPowermeter
+from requests.auth import HTTPBasicAuth
+
+
+class TestJsonHttpPowermeter(unittest.TestCase):
+    @patch("requests.Session.get")
+    def test_single_phase(self, mock_get):
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"power": 100}
+        mock_get.return_value = mock_response
+
+        meter = JsonHttpPowermeter("http://localhost", "$.power")
+        self.assertEqual(meter.get_powermeter_watts(), [100.0])
+
+    @patch("requests.Session.get")
+    def test_three_phase(self, mock_get):
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "p1": 100,
+            "p2": 200,
+            "p3": 300,
+        }
+        mock_get.return_value = mock_response
+
+        meter = JsonHttpPowermeter(
+            "http://localhost",
+            ["$.p1", "$.p2", "$.p3"],
+        )
+        self.assertEqual(meter.get_powermeter_watts(), [100.0, 200.0, 300.0])
+
+    @patch("requests.Session.get")
+    def test_headers_and_auth(self, mock_get):
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"power": 50}
+        mock_get.return_value = mock_response
+
+        meter = JsonHttpPowermeter(
+            "http://localhost",
+            "$.power",
+            username="user",
+            password="pass",
+            headers={"X-Test": "1"},
+        )
+        meter.get_powermeter_watts()
+        mock_get.assert_called_with(
+            "http://localhost",
+            headers={"X-Test": "1"},
+            auth=HTTPBasicAuth("user", "pass"),
+            timeout=10,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()