tronity soc

packages/modules/vehicles/tronity/api.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python3
+import logging
+import jwt
+import paho.mqtt.publish as publish
+import json
+
+from modules.common import req
+from modules.vehicles.tronity.config import TronityVehicleSocConfiguration, TronityVehicleSoc
+from modules.common.abstract_soc import SocUpdateData
+from modules.common.component_state import CarState
+from datetime import datetime
+
+log = logging.getLogger(__name__)
+
+
+def fetch_soc(config: TronityVehicleSocConfiguration, soc_update_data: SocUpdateData, vehicle: int) -> CarState:
+    log.debug("Fetching Tronity SOC")
+    session = create_session(config, vehicle)
+    tronity_vehicle_id = str(config.vehicle_id)
+    tronity_data = fetch_soc_for_vehicle(tronity_vehicle_id, session)
+    return CarState(soc=tronity_data['level'], range=tronity_data['range'], soc_timestamp=tronity_data['lastUpdate'])
+
+
+def is_token_valid(access_token: str) -> bool:
+    if not access_token:
+        log.debug("No token found")
+        return False
+
+    decoded_data = jwt.decode(jwt=access_token, verify=False, algorithms=['HS256'], options={"verify_signature": False})
+    if datetime.utcfromtimestamp(decoded_data['exp']) < datetime.utcnow():
+        log.debug("Token expired: %s", decoded_data)
+        return False
+    else:
+        log.debug("Token still valid: %s", decoded_data)
+        return True
+
+
+def write_token_mqtt(topic: str, token: str, config: TronityVehicleSocConfiguration):
+    try:
+        config.access_token = token
+        value: TronityVehicleSoc = TronityVehicleSoc(configuration=config)
+        value_to_mqtt = json.dumps(value.__dict__, default=lambda o: o.__dict__)
+        publish.single(topic, value_to_mqtt, hostname="localhost", port=1883, retain=True)
+    except Exception as e:
+        log.exception('Token mqtt write exception ' + str(e))
+
+
+def create_session(config: TronityVehicleSocConfiguration, vehicle: int) -> req.Session:
+    session = req.Session()
+    data = {'grant_type': 'app',
+            'client_id': str(config.client_id),
+            'client_secret': str(config.client_secret)}
+
+    if not is_token_valid(str(config.access_token)):
+        log.debug("Requesting new Tronity Access Token")
+        response = session.post(
+            "https://api.tronity.tech/authentication",
+            data=data,
+        )
+
+        if response.status_code != 201:
+            raise Exception("Error requesting Tronity access token, please check client_id and client_secret: %s"
+                            % response.status_code)
+
+        access_token = response.json()['access_token']
+        log.debug("Retrieved Tronity Access Token: %s", access_token)
+        write_token_mqtt(
+                        "openWB/set/vehicle/" + str(vehicle) + "/soc_module/config",
+                        access_token,
+                        config)
+    else:
+        access_token = config.access_token
+
+    session.headers = {
+        'Accept': 'application/hal+json', 'Authorization': 'Bearer %s' % access_token
+    }
+    log.debug("Retrieved Tronity Access Token.")
+    return session
+
+
+def fetch_vehicles(session: req.Session) -> dict:
+    url = 'https://api.tronity.tech/tronity/vehicles'
+    response = session.get(url)
+    if response.status_code != 200:
+        raise Exception("Error requesting vehicles from Tronity: %s" % response.status_code)
+    vehicles = response.json()
+    log.debug("Retrieved Tronity vehicles: %s", vehicles["data"])
+    return vehicles["data"]
+
+
+def fetch_soc_for_vehicle(vehicle_id: str, session: req.Session) -> dict:
+    url = 'https://api.tronity.tech/tronity/vehicles/' + str(vehicle_id) + '/last_record'
+    response = session.get(url)
+    if response.status_code != 200:
+        raise Exception("Error requesting vehicle from Tronity: %s" % response.status_code)
+    data = response.json()
+    log.debug("Retrieved Tronity data: %s", data)
+    return data

packages/modules/vehicles/tronity/config.py

@@ -0,0 +1,23 @@
+from typing import Optional
+
+
+class TronityVehicleSocConfiguration:
+    def __init__(self,
+                 vehicle_id: Optional[str] = None,
+                 client_id: Optional[str] = None,
+                 client_secret: Optional[str] = None,
+                 access_token: Optional[str] = None) -> None:
+        self.vehicle_id = vehicle_id
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.access_token = access_token
+
+
+class TronityVehicleSoc:
+    def __init__(self,
+                 name: str = "Tronity",
+                 type: str = "tronity",
+                 configuration: TronityVehicleSocConfiguration = None) -> None:
+        self.name = name
+        self.type = type
+        self.configuration = configuration or TronityVehicleSocConfiguration()

packages/modules/vehicles/tronity/soc.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+import logging
+
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.abstract_soc import SocUpdateData
+from modules.common.component_state import CarState
+from modules.common.configurable_vehicle import ConfigurableVehicle
+from modules.vehicles.tronity.config import TronityVehicleSoc
+from modules.vehicles.tronity.api import fetch_soc
+
+log = logging.getLogger(__name__)
+
+
+def create_vehicle(vehicle_config: TronityVehicleSoc, vehicle: int):
+    def updater(soc_update_data: SocUpdateData) -> CarState:
+        return fetch_soc(vehicle_config.configuration, soc_update_data, vehicle)
+    return ConfigurableVehicle(vehicle_config=vehicle_config, component_updater=updater, vehicle=vehicle)
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=TronityVehicleSoc)