Add a SoC module for Nissan Leaf till MY 2019

packages/modules/conftest.py

@@ -17,6 +17,7 @@
 sys.modules['skodaconnect.Connection'] = type(sys)('skodaconnect.Connection')
 sys.modules['socketserver'] = type(sys)('socketserver')
 sys.modules['grpc'] = type(sys)('grpc')
+sys.modules['pycarwings3'] = type(sys)('pycarwings3')
 
 
 # sys.modules['telnetlib3'] = type(sys)('telnetlib3')

packages/modules/vehicles/leaf/api.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python3
+import asyncio
+import logging
+from datetime import datetime, timezone
+
+from modules.common.component_state import CarState
+
+import pycarwings3
+
+log = logging.getLogger(__name__)
+
+
+async def _fetch_soc(username, password, region, vehicle) -> CarState:
+
+    async def getNissanSession():
+        # open HTTPS session with Nissan server
+        log.debug("vehicle%s: login = %s, region = %s" % (vehicle, username, region))
+        session = pycarwings3.Session(username, password, region)
+        leaf = await session.get_leaf()
+        await asyncio.sleep(1)
+        return leaf
+
+    async def readSoc(leaf) -> CarState:
+        # get SoC & range & time stamp from Nissan server
+        leaf_info = await leaf.get_latest_battery_status()
+        soc = float(leaf_info.battery_percent)
+        log.debug("vehicle%s: Battery State of Charge %s" % (vehicle, soc))
+        range = int(leaf_info.answer["BatteryStatusRecords"]["CruisingRangeAcOff"])/1000
+        log.debug("vehicle%s: Cruising range AC Off   %s" % (vehicle, range))
+        time_stamp_str_utc = leaf_info.answer["BatteryStatusRecords"]["NotificationDateAndTime"]
+        soc_time = datetime.strptime(f"{time_stamp_str_utc}", "%Y/%m/%d %H:%M").replace(tzinfo=timezone.utc)
+        log.debug("vehicle%s: Date&Time of SoC (UTC)  %s" % (vehicle, soc_time))
+        soc_timestamp = soc_time.timestamp()
+        log.debug("vehicle%s: soc_timestamp           %s" % (vehicle, soc_timestamp))
+        log.debug("vehicle%s: local Date&Time of SoC  %s" % (vehicle, datetime.fromtimestamp(soc_timestamp)))
+        return CarState(soc, range, soc_timestamp)
+
+    async def requestSoc(leaf: pycarwings3.Leaf):
+        # request Nissan server to request last SoC from car
+        log.debug("vehicle%s: Request SoC Update from vehicle" % (vehicle))
+        key = await leaf.request_update()
+        sleepsecs = 20
+        for _ in range(0, 3):
+            log.debug("Waiting {0} seconds".format(sleepsecs))
+            await asyncio.sleep(sleepsecs)
+            status = await leaf.get_status_from_update(key)
+            if status is not None:
+                log.debug("vehicle%s: Update successful" % (vehicle))
+                return status
+        log.debug("vehicle%s: Update not successful" % (vehicle))
+        return status
+
+    try:
+        leaf = await getNissanSession()       # start HTTPS session with Nissan server
+        soc_range = await readSoc(leaf)       # read old SoC & range values from server
+        await asyncio.sleep(1)                # give Nissan server some time
+        status = await requestSoc(leaf)       # Nissan server to request new values from vehicle
+        if status is not None:                # was update of values successful?
+            await asyncio.sleep(1)            # give Nissan server some time
+            soc_range = await readSoc(leaf)   # final read of SoC & range from server
+    except pycarwings3.CarwingsError as e:
+        log.info("vehicle%s: SoC & range request not successful" % (vehicle))
+        log.info(e)
+        soc_range = CarState(0.0, 0.0)
+    return soc_range
+
+
+# main entry - _fetch_soc needs to be run async
+def fetch_soc(user_id: str, password: str, region: str, vehicle: int) -> CarState:
+
+    loop = asyncio.new_event_loop()   # prepare and call async method
+    asyncio.set_event_loop(loop)
+
+    # get SoC and range from vehicle via server
+    soc_range = loop.run_until_complete(_fetch_soc(user_id, password, region, vehicle))
+
+    return soc_range

packages/modules/vehicles/leaf/config.py

@@ -0,0 +1,18 @@
+from typing import Optional
+
+
+class LeafConfiguration:
+    def __init__(self, user_id: Optional[str] = None, password: Optional[str] = None, region: Optional[str] = None):
+        self.user_id = user_id
+        self.password = password
+        self.region = region
+
+
+class LeafSoc:
+    def __init__(self,
+                 name: str = "Nissan Leaf/NV200 -05.2019 (experimental)",
+                 type: str = "leaf",
+                 configuration: LeafConfiguration = None) -> None:
+        self.name = name
+        self.type = type
+        self.configuration = configuration or LeafConfiguration()

packages/modules/vehicles/leaf/soc.py

@@ -0,0 +1,47 @@
+from typing import List
+
+import logging
+
+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_vehicle import VehicleUpdateData
+from modules.common.component_state import CarState
+from modules.common.configurable_vehicle import ConfigurableVehicle
+from modules.vehicles.leaf import api
+from modules.vehicles.leaf.config import LeafSoc, LeafConfiguration
+
+
+log = logging.getLogger(__name__)
+
+
+def create_vehicle(vehicle_config: LeafSoc, vehicle: int):
+    def updater(vehicle_update_data: VehicleUpdateData) -> CarState:
+        return api.fetch_soc(
+            vehicle_config.configuration.user_id,
+            vehicle_config.configuration.password,
+            vehicle_config.configuration.region,
+            vehicle)
+    return ConfigurableVehicle(vehicle_config=vehicle_config,
+                               component_updater=updater,
+                               vehicle=vehicle)
+
+
+def leaf_update(user_id: str, password: str, region: str, charge_point: int):
+    log.debug("leaf: user_id="+user_id+" region="+region+" charge_point="+str(charge_point))
+    vehicle_config = LeafSoc(configuration=LeafConfiguration(charge_point,
+                                                             user_id,
+                                                             password,
+                                                             region))
+    store.get_car_value_store(charge_point).store.set(api.fetch_soc(
+        vehicle_config.configuration.user_id,
+        vehicle_config.configuration.password,
+        vehicle_config.configuration.region,
+        charge_point))
+
+
+def main(argv: List[str]):
+    run_using_positional_cli_args(leaf_update, argv)
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=LeafSoc)

requirements.txt

@@ -23,3 +23,4 @@ protobuf==4.25.3
 bimmer_connected==0.17.2
 ocpp==1.0.0
 websockets==12.0
+pycarwings3==0.7.13