Extracted OpenWB Integration

Fixes #354

Author: nanomad

src/extractors/__init__.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from utils import value_in_range
+
+if TYPE_CHECKING:
+    from status_publisher.charge.chrg_mgmt_data_resp import (
+        ChrgMgmtDataRespProcessingResult,
+    )
+    from status_publisher.vehicle.vehicle_status_resp import (
+        VehicleStatusRespProcessingResult,
+    )
+
+LOG = logging.getLogger(__name__)
+
+
+def extract_electric_range(
+    vehicle_status: VehicleStatusRespProcessingResult,
+    charge_status: ChrgMgmtDataRespProcessingResult | None,
+) -> float | None:
+    if (
+        charge_status is not None
+        and (raw_fuel_range_elec := charge_status.raw_fuel_range_elec) is not None
+        and (actual_range := __validate_and_convert_electric_range(raw_fuel_range_elec))
+        is not None
+    ):
+        LOG.debug("Electric range derived from charge_status")
+        return actual_range
+
+    if (raw_range := vehicle_status.fuel_range_elec) is not None and (
+        actual_range := __validate_and_convert_electric_range(raw_range)
+    ) is not None:
+        LOG.debug("Electric range derived from vehicle_status")
+        return actual_range
+
+    LOG.warning("Could not extract a valid electric range")
+    return None
+
+
+def extract_soc(
+    vehicle_status: VehicleStatusRespProcessingResult,
+    charge_status: ChrgMgmtDataRespProcessingResult | None,
+) -> float | None:
+    if (
+        charge_status is not None
+        and (raw_soc := charge_status.raw_soc) is not None
+        and (soc := __validate_and_convert_soc(raw_soc / 10.0)) is not None
+    ):
+        LOG.debug("SoC derived from charge_status")
+        return soc
+
+    if (raw_soc := vehicle_status.raw_soc) is not None and (
+        soc := __validate_and_convert_soc(float(raw_soc))
+    ) is not None:
+        LOG.debug("SoC derived from vehicle_status")
+        return soc
+
+    LOG.warning("Could not extract a valid SoC")
+    return None
+
+
+def __validate_and_convert_electric_range(raw_value: int) -> float | None:
+    if value_in_range(raw_value, 1, 20460):
+        return raw_value / 10.0
+    return None
+
+
+def __validate_and_convert_soc(raw_value: float) -> float | None:
+    if value_in_range(raw_value, 0, 100.0, is_max_excl=False):
+        return raw_value
+    return None

src/handlers/vehicle.py

@@ -12,6 +12,7 @@
 from integrations import IntegrationException
 from integrations.abrp.api import AbrpApi
 from integrations.home_assistant.discovery import HomeAssistantDiscovery
+from integrations.openwb import OpenWBIntegration
 from integrations.osmand.api import OsmAndApi
 import mqtt_topics
 from saic_api_listener import MqttGatewayAbrpListener, MqttGatewayOsmAndListener
@@ -61,8 +62,9 @@ def __init__(
         self.vehicle_state = vehicle_state
         self.__ha_discovery = self.__setup_ha_discovery(vehicle_state, vin_info, config)
 
-        self.__setup_abrp(config, vin_info)
-        self.__setup_osmand(config, vin_info)
+        self.openwb_integration = self.__setup_openwb(config, vin_info, publisher)
+        self.abrp_api = self.__setup_abrp(config, vin_info)
+        self.osmand_api = self.__setup_osmand(config, vin_info)
         self.__vehicle_info_publisher = VehicleInfoPublisher(
             self.vin_info, self.publisher, self.vehicle_prefix
         )
@@ -74,7 +76,18 @@ def __init__(
             vehicle_prefix=self.vehicle_prefix,
         )
 
-    def __setup_abrp(self, config: Configuration, vin_info: VehicleInfo) -> None:
+    def __setup_openwb(
+        self, config: Configuration, vin_info: VehicleInfo, publisher: Publisher
+    ) -> OpenWBIntegration | None:
+        charging_station = config.charging_stations_by_vin.get(vin_info.vin, None)
+        if not charging_station:
+            return None
+        return OpenWBIntegration(
+            charging_station=charging_station,
+            publisher=publisher,
+        )
+
+    def __setup_abrp(self, config: Configuration, vin_info: VehicleInfo) -> AbrpApi:
         if vin_info.vin in self.configuration.abrp_token_map:
             abrp_user_token = self.configuration.abrp_token_map[vin_info.vin]
         else:
@@ -83,14 +96,15 @@ def __setup_abrp(self, config: Configuration, vin_info: VehicleInfo) -> None:
             abrp_api_listener = MqttGatewayAbrpListener(self.publisher)
         else:
             abrp_api_listener = None
-        self.abrp_api = AbrpApi(
+        return AbrpApi(
             self.configuration.abrp_api_key, abrp_user_token, listener=abrp_api_listener
         )
 
-    def __setup_osmand(self, config: Configuration, vin_info: VehicleInfo) -> None:
+    def __setup_osmand(
+        self, config: Configuration, vin_info: VehicleInfo
+    ) -> OsmAndApi | None:
         if not self.configuration.osmand_server_uri:
-            self.osmand_api = None
-            return
+            return None
 
         if config.publish_raw_osmand_data:
             api_listener = MqttGatewayOsmAndListener(self.publisher)
@@ -99,7 +113,7 @@ def __setup_osmand(self, config: Configuration, vin_info: VehicleInfo) -> None:
         osmand_device_id = self.configuration.osmand_device_id_map.get(
             vin_info.vin, vin_info.vin
         )
-        self.osmand_api = OsmAndApi(
+        return OsmAndApi(
             server_uri=self.configuration.osmand_server_uri,
             device_id=osmand_device_id,
             listener=api_listener,
@@ -171,7 +185,6 @@ async def __polling(self) -> None:
                 )
         else:
             LOG.debug("Skipping EV-related updates as the vehicle is not an EV")
-            charge_status = None
 
         self.vehicle_state.update_data_conflicting_in_vehicle_and_bms(
             vehicle_status_processing_result, charge_status_processing_result
@@ -180,6 +193,9 @@ async def __polling(self) -> None:
         self.vehicle_state.mark_successful_refresh()
         LOG.info("Refreshing vehicle status succeeded...")
 
+        self.__refresh_openwb(
+            vehicle_status_processing_result, charge_status_processing_result
+        )
         await self.__refresh_abrp(charge_status, vehicle_status)
         await self.__refresh_osmand(charge_status, vehicle_status)
 
@@ -196,10 +212,21 @@ def __should_complete_configuration(self, start_time: datetime.datetime) -> bool
             and datetime.datetime.now() > start_time + datetime.timedelta(seconds=10)
         )
 
+    def __refresh_openwb(
+        self,
+        vehicle_status_processing_result: VehicleStatusRespProcessingResult,
+        charge_status_processing_result: ChrgMgmtDataRespProcessingResult | None,
+    ) -> None:
+        if self.openwb_integration is None:
+            return
+        self.openwb_integration.update_openwb(
+            vehicle_status_processing_result, charge_status_processing_result
+        )
+
     async def __refresh_osmand(
         self,
         charge_status: ChrgMgmtDataResp | None,
-        vehicle_status: VehicleStatusResp | None,
+        vehicle_status: VehicleStatusResp,
     ) -> None:
         if not self.osmand_api:
             return
@@ -215,9 +242,7 @@ async def __refresh_osmand(
             LOG.info(f"OsmAnd not refreshed, reason {response}")
 
     async def __refresh_abrp(
-        self,
-        charge_status: ChrgMgmtDataResp | None,
-        vehicle_status: VehicleStatusResp | None,
+        self, charge_status: ChrgMgmtDataResp | None, vehicle_status: VehicleStatusResp
     ) -> None:
         abrp_refreshed, abrp_response = await self.abrp_api.update_abrp(
             vehicle_status, charge_status

src/integrations/openwb/__init__.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+import extractors
+from integrations.openwb.charging_station import ChargingStation
+
+if TYPE_CHECKING:
+    from publisher.core import Publisher
+    from status_publisher.charge.chrg_mgmt_data_resp import (
+        ChrgMgmtDataRespProcessingResult,
+    )
+    from status_publisher.vehicle.vehicle_status_resp import (
+        VehicleStatusRespProcessingResult,
+    )
+
+LOG = logging.getLogger(__name__)
+
+__all__ = [
+    "ChargingStation",
+    "OpenWBIntegration",
+]
+
+
+class OpenWBIntegration:
+    def __init__(
+        self, *, charging_station: ChargingStation, publisher: Publisher
+    ) -> None:
+        self.__charging_station = charging_station
+        self.__publisher = publisher
+
+    def update_openwb(
+        self,
+        vehicle_status: VehicleStatusRespProcessingResult,
+        charge_status: ChrgMgmtDataRespProcessingResult | None,
+    ) -> None:
+        range_topic = self.__charging_station.range_topic
+        electric_range = extractors.extract_electric_range(
+            vehicle_status, charge_status
+        )
+        if electric_range is not None and range_topic is not None:
+            LOG.info("OpenWB Integration published range to %f", range_topic)
+            self.__publisher.publish_float(
+                key=range_topic,
+                value=electric_range,
+                no_prefix=True,
+            )
+
+        soc_topic = self.__charging_station.soc_topic
+        soc = extractors.extract_soc(vehicle_status, charge_status)
+        if soc is not None and soc_topic is not None:
+            LOG.info("OpenWB Integration published SoC to %f", soc_topic)
+            self.__publisher.publish_float(
+                key=soc_topic,
+                value=soc,
+                no_prefix=True,
+            )

src/mqtt_gateway.py

@@ -150,25 +150,11 @@ async def setup_vehicle(
         account_prefix = (
             f"{self.configuration.saic_user}/{mqtt_topics.VEHICLES}/{vin_info.vin}"
         )
-        charging_station = self.get_charging_station(vin_info.vin)
-        if charging_station and charging_station.soc_topic:
-            LOG.debug(
-                "SoC of %s for charging station will be published over MQTT topic: %s",
-                vin_info.vin,
-                charging_station.soc_topic,
-            )
-        if charging_station and charging_station.range_topic:
-            LOG.debug(
-                "Range of %s for charging station will be published over MQTT topic: %s",
-                vin_info.vin,
-                charging_station.range_topic,
-            )
         vehicle_state = VehicleState(
             self.publisher,
             self.__scheduler,
             account_prefix,
             vin_info,
-            charging_station,
             charge_polling_min_percent=self.configuration.charge_dynamic_polling_min_percentage,
         )
         vehicle_handler = VehicleHandler(

src/publisher/mqtt_publisher.py

@@ -55,9 +55,7 @@ async def connect(self) -> None:
             ssl_context = ssl.create_default_context()
             cert_uri = self.configuration.tls_server_cert_path
             if cert_uri:
-                LOG.debug(
-                    f"Using custom CA file {cert_uri}"
-                )
+                LOG.debug(f"Using custom CA file {cert_uri}")
                 ssl_context.load_verify_locations(cafile=cert_uri)
                 ssl_context.check_hostname = False
         else: