909cdc2e5c
* Copy Paste Find Replace * Small progress * wip * more wip * Add SSE listen and close * More rework * Fix coordinator * Get working * Bump to 0.1.3 * Push to 0.1.4 * Lots of fixes * Remove stream * Add wakeup * Improve set temp * Be consistent with self * Increase polling until streaming * Work in progress * Move to single climate * bump to 0.2.0 * Update entity * Data handling * fixes * WIP tests * Tests * Delete other tests * Update comment * Fix init * Update homeassistant/components/teslemetry/entity.py Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com> * Add Codeowner * Update coverage * requirements * Add failure for subscription required * Add VIN to model * Add wake * Add context manager * Rename to wake_up_if_asleep * Remove context from coverage * change lock to context Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com> * Improving Logger * Add url to subscription error * Errors cannot markdown * Fix logger Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com> * rename logger * Fix error logging * Apply suggestions from code review --------- Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
"""Teslemetry Data Coordinator."""
|
|
from datetime import timedelta
|
|
from typing import Any
|
|
|
|
from tesla_fleet_api.exceptions import TeslaFleetError, VehicleOffline
|
|
from tesla_fleet_api.vehiclespecific import VehicleSpecific
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ConfigEntryNotReady
|
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
|
|
|
from .const import LOGGER, TeslemetryState
|
|
|
|
SYNC_INTERVAL = 60
|
|
|
|
|
|
class TeslemetryVehicleDataCoordinator(DataUpdateCoordinator[dict[str, Any]]):
|
|
"""Class to manage fetching data from the Teslemetry API."""
|
|
|
|
def __init__(self, hass: HomeAssistant, api: VehicleSpecific) -> None:
|
|
"""Initialize Teslemetry Data Update Coordinator."""
|
|
super().__init__(
|
|
hass,
|
|
LOGGER,
|
|
name="Teslemetry Vehicle",
|
|
update_interval=timedelta(seconds=SYNC_INTERVAL),
|
|
)
|
|
self.api = api
|
|
|
|
async def async_config_entry_first_refresh(self) -> None:
|
|
"""Perform first refresh."""
|
|
try:
|
|
response = await self.api.wake_up()
|
|
if response["response"]["state"] != TeslemetryState.ONLINE:
|
|
# The first refresh will fail, so retry later
|
|
raise ConfigEntryNotReady("Vehicle is not online")
|
|
except TeslaFleetError as e:
|
|
# The first refresh will also fail, so retry later
|
|
raise ConfigEntryNotReady from e
|
|
await super().async_config_entry_first_refresh()
|
|
|
|
async def _async_update_data(self) -> dict[str, Any]:
|
|
"""Update vehicle data using Teslemetry API."""
|
|
|
|
try:
|
|
data = await self.api.vehicle_data()
|
|
except VehicleOffline:
|
|
self.data["state"] = TeslemetryState.OFFLINE
|
|
return self.data
|
|
except TeslaFleetError as e:
|
|
raise UpdateFailed(e.message) from e
|
|
|
|
return self._flatten(data["response"])
|
|
|
|
def _flatten(
|
|
self, data: dict[str, Any], parent: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""Flatten the data structure."""
|
|
result = {}
|
|
for key, value in data.items():
|
|
if parent:
|
|
key = f"{parent}_{key}"
|
|
if isinstance(value, dict):
|
|
result.update(self._flatten(value, key))
|
|
else:
|
|
result[key] = value
|
|
return result
|