SOC Modul über SkodaConnect

packages/dataclass_utils/_dataclass_from_dict.py

@@ -1,5 +1,6 @@
 import inspect
 from inspect import FullArgSpec
+import typing
 from typing import TypeVar, Type, Union
 
 T = TypeVar('T')
@@ -38,5 +39,16 @@ def _get_argument_value(arg_spec: FullArgSpec, index: int, parameters: dict):
 
 def _dataclass_from_dict_recurse(value, requested_type: Type[T]):
     return dataclass_from_dict(requested_type, value) \
-        if isinstance(value, dict) and not issubclass(requested_type, dict) \
+        if isinstance(value, dict) and not (
+            _is_optional_of_dict(requested_type) or
+            issubclass(requested_type, dict)) \
         else value
+
+
+def _is_optional_of_dict(requested_type):
+    # Optional[dict] is an alias for Union[dict, None]
+    if typing.get_origin(requested_type) == Union:
+        args = typing.get_args(requested_type)
+        if len(args) == 2:
+            return issubclass(args[0], dict) and issubclass(args[1], type(None))
+    return False

packages/modules/conftest.py

@@ -13,6 +13,8 @@
 sys.modules['lxml.html'] = type(sys)('lxml.html')
 sys.modules['bs4'] = type(sys)('bs4')
 sys.modules['pkce'] = type(sys)('pkce')
+sys.modules['skodaconnect'] = type(sys)('skodaconnect')
+sys.modules['skodaconnect.Connection'] = type(sys)('skodaconnect.Connection')
 
 module = type(sys)('pymodbus.client.sync')
 module.ModbusSerialClient = Mock()

packages/modules/vehicles/skodaconnect/api.py

@@ -0,0 +1,88 @@
+#!/usr/bin/env python3
+
+import aiohttp
+import asyncio
+import logging
+from helpermodules.pub import Pub
+from modules.vehicles.skodaconnect.config import SkodaConnect, SkodaConnectConfiguration
+from skodaconnect import Connection
+from typing import Union
+
+
+log = logging.getLogger("soc."+__name__)
+
+
+class SkodaConnectApi():
+
+    def __init__(self, conf: SkodaConnect, vehicle: int) -> None:
+        self.user_id = conf.configuration.user_id
+        self.password = conf.configuration.password
+        self.vin = conf.configuration.vin
+        self.refresh_token = conf.configuration.refresh_token
+        self.vehicle = vehicle
+
+    def fetch_soc(self) -> Union[int, float]:
+        # prepare and call async method
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        soc, range = loop.run_until_complete(self._fetch_soc())
+        return soc, range
+
+    async def _fetch_soc(self) -> Union[int, float]:
+        async with aiohttp.ClientSession(headers={'Connection': 'keep-alive'}) as session:
+            soc = 0
+            range = 0.0
+            login_success = False
+
+            log.debug(f"Initiating new session to Skoda Connect with {self.user_id} as username")
+            try:
+                connection = Connection(session, self.user_id, self.password)
+                if self.refresh_token is not None:
+                    log.debug("Attempting restore of tokens")
+                    if await connection.restore_tokens(self.refresh_token):
+                        log.debug("Token restore succeeded")
+                        login_success = True
+
+                if not login_success:
+                    log.debug("Attempting to login to the Skoda Connect service")
+                    login_success = await connection.doLogin()
+            except Exception:
+                log.exception("Login failed!")
+
+            if login_success:
+                log.debug('Login success!')
+                tokens = await connection.save_tokens()
+                log.debug('Fetching charging data.')
+                chargingState = await connection.getCharging(self.vin)
+                if chargingState:
+                    if 'error_description' in chargingState:
+                        log.error(f"Failed to fetch charging data: {chargingState.get('error_description')}")
+                    if 'battery' in chargingState:
+                        batteryState = chargingState.get('battery')
+                        soc = batteryState.get('stateOfChargeInPercent')
+                        log.debug(f"Battery level: {soc}")
+                        range = int(batteryState.get('cruisingRangeElectricInMeters'))/1000
+                        log.debug(f"Electric range: {range}")
+                    if tokens:
+                        self._persist_refresh_tokens(tokens)
+                elif self.refresh_token is not None:
+                    # token seems to be invalid
+                    self._persist_refresh_tokens(None)
+            return soc, range
+
+    def _persist_refresh_tokens(self, tokens: dict) -> None:
+        log.debug('Persist refresh tokens.')
+        conf = SkodaConnect(
+            configuration=SkodaConnectConfiguration(
+                self.user_id,
+                self.password,
+                self.vin,
+                tokens))
+        self._publish_refresh_tokens(conf.as_dict())
+
+    def _publish_refresh_tokens(self, config={}) -> None:
+        try:
+            Pub().pub("openWB/set/vehicle/" + self.vehicle + "/soc_module/config", config)
+        except Exception as e:
+            log.exception('Token mqtt write exception ' + str(e))

packages/modules/vehicles/skodaconnect/config.py

@@ -0,0 +1,30 @@
+from typing import Optional
+from dataclass_utils import asdict
+
+
+class SkodaConnectConfiguration:
+    def __init__(self,
+                 user_id: Optional[str] = None,        # show in UI
+                 password: Optional[str] = None,       # show in UI
+                 vin: Optional[str] = None,            # show in UI
+                 refresh_token: Optional[dict] = None   # DON'T show in UI!
+                 ):
+        self.user_id = user_id
+        self.password = password
+        self.vin = vin
+        self.refresh_token = refresh_token
+
+
+class SkodaConnect:
+    def __init__(self,
+                 name: str = "SkodaConnect",
+                 type: str = "skodaconnect",
+                 configuration: SkodaConnectConfiguration = None) -> None:
+        self.name = name
+        self.type = type
+        self.configuration = configuration or SkodaConnectConfiguration()
+
+    def as_dict(self) -> dict:
+        confDict = asdict(self)
+        confDict.pop('name')
+        return confDict

packages/modules/vehicles/skodaconnect/soc.py

@@ -0,0 +1,42 @@
+from typing import Union, List
+
+import logging
+
+from dataclass_utils import dataclass_from_dict
+from helpermodules.cli import run_using_positional_cli_args
+from modules.common import store
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.abstract_soc import AbstractSoc
+from modules.common.component_context import SingleComponentUpdateContext
+from modules.common.fault_state import ComponentInfo
+from modules.vehicles.skodaconnect.api import SkodaConnectApi
+from modules.vehicles.skodaconnect.config import SkodaConnect, SkodaConnectConfiguration
+
+
+log = logging.getLogger("soc."+__name__)
+
+
+class Soc(AbstractSoc):
+    def __init__(self, device_config: Union[dict, SkodaConnect], vehicle: int):
+        self.config = dataclass_from_dict(SkodaConnect, device_config)
+        self.vehicle = vehicle
+        self.store = store.get_car_value_store(self.vehicle)
+        self.component_info = ComponentInfo(self.vehicle, self.config.name, "vehicle")
+
+    def update(self, charge_state: bool = False) -> None:
+        with SingleComponentUpdateContext(self.component_info):
+            soc, range = SkodaConnectApi(self.config, self.vehicle).fetch_soc()
+
+
+def skodaconnect_update(user_id: str, password: str, vin: str, refresh_token: str, charge_point: int):
+    log.debug("skodaconnect: userid="+user_id+"vin="+vin+"charge_point="+str(charge_point))
+    Soc(
+        SkodaConnect(configuration=SkodaConnectConfiguration(user_id, password, vin, refresh_token)),
+        charge_point).update()
+
+
+def main(argv: List[str]):
+    run_using_positional_cli_args(skodaconnect_update, argv)
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=SkodaConnect)

requirements.txt

@@ -10,3 +10,4 @@ PyJWT==2.6.0
 ipparser==0.3.8
 bs4==0.0.1
 pkce==1.0.3
+skodaconnect==1.3.4