|
| 1 | +import logging |
| 2 | + |
| 3 | +from typing import List, Union |
| 4 | +from datetime import datetime |
| 5 | + |
| 6 | +from helpermodules.cli import run_using_positional_cli_args |
| 7 | +from modules.common import req |
| 8 | +from modules.common import store |
| 9 | +from modules.common.abstract_device import DeviceDescriptor |
| 10 | +from modules.common.abstract_vehicle import VehicleUpdateData |
| 11 | +from modules.common.component_state import CarState |
| 12 | +from modules.common.configurable_vehicle import ConfigurableVehicle |
| 13 | +from modules.vehicles.homeassistant.config import HaVehicleSocSetup, HaVehicleSocConfiguration |
| 14 | + |
| 15 | + |
| 16 | +log = logging.getLogger(__name__) |
| 17 | + |
| 18 | + |
| 19 | +def extract_to_epoch(input_string: Union[str, int, float]) -> float: |
| 20 | + # If already an integer, return it |
| 21 | + if isinstance(input_string, int) or isinstance(input_string, float): |
| 22 | + return int(input_string) |
| 23 | + |
| 24 | + # Try parsing as UTC formatted time |
| 25 | + try: |
| 26 | + dt = datetime.fromisoformat(input_string) |
| 27 | + return int(dt.timestamp()) |
| 28 | + except ValueError: |
| 29 | + log.exception(f'Kein ISO 8601 formatiertes Datum in "{input_string}" gefunden.') |
| 30 | + return None |
| 31 | + |
| 32 | + |
| 33 | +def fetch_soc(config: HaVehicleSocSetup) -> CarState: |
| 34 | + url = config.configuration.url |
| 35 | + entity_id = config.configuration.entity_id |
| 36 | + token = config.configuration.token |
| 37 | + if url is None or url == "": |
| 38 | + raise ValueError("Keine URL zum Abrufen der Daten definiert. Bitte Konfiguration anpassen.") |
| 39 | + if entity_id is None or entity_id == "": |
| 40 | + raise ValueError("Keine Entitäts-ID definiert. Bitte Konfiguration anpassen.") |
| 41 | + if token is None or token == "": |
| 42 | + raise ValueError("Kein Token definiert. Bitte Konfiguration anpassen.") |
| 43 | + url = url + "/api/states/" + entity_id |
| 44 | + response = req.get_http_session().get(url, timeout=10, |
| 45 | + headers={ |
| 46 | + "authorization": "Bearer " + token, |
| 47 | + "content-type": "application/json"} |
| 48 | + ) |
| 49 | + json = response.json() |
| 50 | + soc = float(json['state']) |
| 51 | + soc_timestamp = extract_to_epoch(json['last_changed']) |
| 52 | + return CarState(soc=soc, soc_timestamp=soc_timestamp) |
| 53 | + |
| 54 | + |
| 55 | +def create_vehicle(vehicle_config: HaVehicleSocSetup, vehicle: int): |
| 56 | + def updater(vehicle_update_data: VehicleUpdateData) -> CarState: |
| 57 | + return fetch_soc(vehicle_config) |
| 58 | + return ConfigurableVehicle(vehicle_config=vehicle_config, |
| 59 | + component_updater=updater, |
| 60 | + vehicle=vehicle) |
| 61 | + |
| 62 | + |
| 63 | +def json_update(charge_point: int, |
| 64 | + url: str, |
| 65 | + token: str, |
| 66 | + entity_id: str |
| 67 | + ): |
| 68 | + log.debug(f'homeassistant-soc: charge_point={charge_point} url="{url}" token="{token}" ' |
| 69 | + f'entity_id="{entity_id}"') |
| 70 | + store.get_car_value_store(charge_point).store.set( |
| 71 | + fetch_soc(HaVehicleSocSetup(configuration=HaVehicleSocConfiguration(url=url, |
| 72 | + token=token, |
| 73 | + entity_id=entity_id)))) |
| 74 | + |
| 75 | + |
| 76 | +def main(argv: List[str]): |
| 77 | + run_using_positional_cli_args(json_update, argv) |
| 78 | + |
| 79 | + |
| 80 | +device_descriptor = DeviceDescriptor(configuration_factory=HaVehicleSocSetup) |
0 commit comments