-
Notifications
You must be signed in to change notification settings - Fork 60
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* skodaconnect inital version * handle refresh tokens * implement api as class * implement config to dict in class * Update packages/modules/vehicles/skodaconnect/soc.py Co-authored-by: LKuemmel <76958050+LKuemmel@users.noreply.github.com> * Update packages/modules/vehicles/skodaconnect/api.py Co-authored-by: LKuemmel <76958050+LKuemmel@users.noreply.github.com> * add logging for communication errors * keep name in Json config * create JSON directly * test config with Optional of dict --------- Co-authored-by: LKuemmel <76958050+LKuemmel@users.noreply.github.com>
- Loading branch information
1 parent
75b936a
commit 84b3be7
Showing
8 changed files
with
196 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,89 @@ | ||
#!/usr/bin/env python3 | ||
|
||
import aiohttp | ||
import asyncio | ||
import logging | ||
from dataclass_utils import asdict | ||
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(asdict(conf)) | ||
|
||
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)) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,24 @@ | ||
from typing import Optional | ||
|
||
|
||
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() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -10,3 +10,4 @@ PyJWT==2.6.0 | |
ipparser==0.3.8 | ||
bs4==0.0.1 | ||
pkce==1.0.3 | ||
skodaconnect==1.3.4 |