homeassistant vehicle soc

packages/modules/vehicles/homeassistant/config.py

@@ -0,0 +1,28 @@
+from helpermodules.auto_str import auto_str
+from typing import Optional
+
+
+@auto_str
+class HaVehicleSocConfiguration:
+    def __init__(
+            self,
+            url: Optional[str] = None,
+            token: Optional[str] = None,
+            entity_id: Optional[str] = None
+            ):
+        self.url = url
+        self.token = token
+        self.entity_id = entity_id
+
+
+@auto_str
+class HaVehicleSocSetup():
+    def __init__(self,
+                 name: str = "HomeAssistant",
+                 type: str = "homeassistant",
+                 official: bool = True,
+                 configuration: HaVehicleSocConfiguration = None) -> None:
+        self.name = name
+        self.type = type
+        self.official = official
+        self.configuration = configuration or HaVehicleSocConfiguration()

packages/modules/vehicles/homeassistant/soc.py

@@ -0,0 +1,80 @@
+import logging
+
+from typing import List, Union
+from datetime import datetime
+
+from helpermodules.cli import run_using_positional_cli_args
+from modules.common import req
+from modules.common import store
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.abstract_vehicle import VehicleUpdateData
+from modules.common.component_state import CarState
+from modules.common.configurable_vehicle import ConfigurableVehicle
+from modules.vehicles.homeassistant.config import HaVehicleSocSetup, HaVehicleSocConfiguration
+
+
+log = logging.getLogger(__name__)
+
+
+def extract_to_epoch(input_string: Union[str, int, float]) -> float:
+    # If already an integer, return it
+    if isinstance(input_string, int) or isinstance(input_string, float):
+        return int(input_string)
+
+    # Try parsing as UTC formatted time
+    try:
+        dt = datetime.fromisoformat(input_string)
+        return int(dt.timestamp())
+    except ValueError:
+        log.exception(f'Kein ISO 8601 formatiertes Datum in "{input_string}" gefunden.')
+        return None
+
+
+def fetch_soc(config: HaVehicleSocSetup) -> CarState:
+    url = config.configuration.url
+    entity_id = config.configuration.entity_id
+    token = config.configuration.token
+    if url is None or url == "":
+        raise ValueError("Keine URL zum Abrufen der Daten definiert. Bitte Konfiguration anpassen.")
+    if entity_id is None or entity_id == "":
+        raise ValueError("Keine Entitäts-ID definiert. Bitte Konfiguration anpassen.")
+    if token is None or token == "":
+        raise ValueError("Kein Token definiert. Bitte Konfiguration anpassen.")
+    url = url + "/api/states/" + entity_id
+    response = req.get_http_session().get(url, timeout=10,
+                                          headers={
+                                              "authorization": "Bearer " + token,
+                                              "content-type": "application/json"}
+                                          )
+    json = response.json()
+    soc = float(json['state'])
+    soc_timestamp = extract_to_epoch(json['last_changed'])
+    return CarState(soc=soc, soc_timestamp=soc_timestamp)
+
+
+def create_vehicle(vehicle_config: HaVehicleSocSetup, vehicle: int):
+    def updater(vehicle_update_data: VehicleUpdateData) -> CarState:
+        return fetch_soc(vehicle_config)
+    return ConfigurableVehicle(vehicle_config=vehicle_config,
+                               component_updater=updater,
+                               vehicle=vehicle)
+
+
+def json_update(charge_point: int,
+                url: str,
+                token: str,
+                entity_id: str
+                ):
+    log.debug(f'homeassistant-soc: charge_point={charge_point} url="{url}" token="{token}" '
+              f'entity_id="{entity_id}"')
+    store.get_car_value_store(charge_point).store.set(
+        fetch_soc(HaVehicleSocSetup(configuration=HaVehicleSocConfiguration(url=url,
+                                                                            token=token,
+                                                                            entity_id=entity_id))))
+
+
+def main(argv: List[str]):
+    run_using_positional_cli_args(json_update, argv)
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=HaVehicleSocSetup)

packages/modules/vehicles/homeassistant/test_ha_soc.py

@@ -0,0 +1,52 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from modules.vehicles.homeassistant.soc import fetch_soc, HaVehicleSocSetup, HaVehicleSocConfiguration
+
+
+class TestSoc(unittest.TestCase):
+
+    def setUp(self):
+        self.test_cases = [{
+            "sample_data": {
+                "entity_id": "sensor.ioniq_ev_battery_level",
+                "state": "84",
+                "attributes": {
+                    "state_class": "measurement",
+                    "unit_of_measurement": "%",
+                    "device_class": "battery",
+                    "friendly_name": "IONIQ EV Battery Level"
+                },
+                "last_changed": "2025-09-29T17:48:02.754865+00:00",
+                "last_reported": "2025-09-29T17:48:02.754865+00:00",
+                "last_updated": "2025-09-29T17:48:02.754865+00:00",
+                "context": {
+                    "id": "05K6B4GTT29YSKIDVJV2R7V8KY",
+                    "parent_id": None,
+                    "user_id": None
+                }
+            },
+            "url": "http://1.1.1.1:4711",
+            "entity_id": "sensor.ioniq_ev_battery_level",
+            "token": "testtoken",
+            "expected_soc": 84,
+            "expected_range": None,
+            "expected_timestamp": 1759168082
+        }]
+
+    @patch('soc.req.get_http_session')
+    def test_fetch_soc(self, mock_get_http_session):
+        for case in self.test_cases:
+            mock_response = MagicMock()
+            mock_response.json.return_value = case['sample_data']
+            mock_get_http_session.return_value.get.return_value = mock_response
+
+            vehicle_config = HaVehicleSocSetup(configuration=HaVehicleSocConfiguration(
+                url=case['url'],
+                token=case['token'],
+                entity_id=case['entity_id']
+            ))
+            car_state = fetch_soc(vehicle_config)
+
+            self.assertEqual(car_state.soc, case['expected_soc'])
+            self.assertEqual(car_state.range, case['expected_range'])
+            self.assertEqual(car_state.soc_timestamp, case['expected_timestamp'])

packages/modules/vehicles/json/test_soc.py → packages/modules/vehicles/json/test_json_soc.py

@@ -1,6 +1,6 @@
 import unittest
 from unittest.mock import patch, MagicMock
-from soc import initialize_vehicle, fetch_soc, JsonSocSetup, JsonSocConfiguration
+from modules.vehicles.json.soc import initialize_vehicle, fetch_soc, JsonSocSetup, JsonSocConfiguration
 
 
 class TestSoc(unittest.TestCase):