135 lines
4.6 KiB
Python
135 lines
4.6 KiB
Python
"""Teslemetry integration."""
|
|
|
|
import asyncio
|
|
from typing import Final
|
|
|
|
from tesla_fleet_api import EnergySpecific, Teslemetry, VehicleSpecific
|
|
from tesla_fleet_api.const import Scope
|
|
from tesla_fleet_api.exceptions import (
|
|
InvalidToken,
|
|
SubscriptionRequired,
|
|
TeslaFleetError,
|
|
)
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.const import CONF_ACCESS_TOKEN, Platform
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
from homeassistant.helpers.device_registry import DeviceInfo
|
|
|
|
from .const import DOMAIN, MODELS
|
|
from .coordinator import (
|
|
TeslemetryEnergySiteInfoCoordinator,
|
|
TeslemetryEnergySiteLiveCoordinator,
|
|
TeslemetryVehicleDataCoordinator,
|
|
)
|
|
from .models import TeslemetryData, TeslemetryEnergyData, TeslemetryVehicleData
|
|
|
|
PLATFORMS: Final = [
|
|
Platform.CLIMATE,
|
|
Platform.SENSOR,
|
|
]
|
|
|
|
|
|
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|
"""Set up Teslemetry config."""
|
|
|
|
access_token = entry.data[CONF_ACCESS_TOKEN]
|
|
session = async_get_clientsession(hass)
|
|
|
|
# Create API connection
|
|
teslemetry = Teslemetry(
|
|
session=session,
|
|
access_token=access_token,
|
|
)
|
|
try:
|
|
scopes = (await teslemetry.metadata())["scopes"]
|
|
products = (await teslemetry.products())["response"]
|
|
except InvalidToken as e:
|
|
raise ConfigEntryAuthFailed from e
|
|
except SubscriptionRequired as e:
|
|
raise ConfigEntryAuthFailed from e
|
|
except TeslaFleetError as e:
|
|
raise ConfigEntryNotReady from e
|
|
|
|
# Create array of classes
|
|
vehicles: list[TeslemetryVehicleData] = []
|
|
energysites: list[TeslemetryEnergyData] = []
|
|
for product in products:
|
|
if "vin" in product and Scope.VEHICLE_DEVICE_DATA in scopes:
|
|
# Remove the protobuff 'cached_data' that we do not use to save memory
|
|
product.pop("cached_data", None)
|
|
vin = product["vin"]
|
|
api = VehicleSpecific(teslemetry.vehicle, vin)
|
|
coordinator = TeslemetryVehicleDataCoordinator(hass, api, product)
|
|
device = DeviceInfo(
|
|
identifiers={(DOMAIN, vin)},
|
|
manufacturer="Tesla",
|
|
configuration_url="https://teslemetry.com/console",
|
|
name=product["display_name"],
|
|
model=MODELS.get(vin[3]),
|
|
serial_number=vin,
|
|
)
|
|
|
|
vehicles.append(
|
|
TeslemetryVehicleData(
|
|
api=api,
|
|
coordinator=coordinator,
|
|
vin=vin,
|
|
device=device,
|
|
)
|
|
)
|
|
elif "energy_site_id" in product and Scope.ENERGY_DEVICE_DATA in scopes:
|
|
site_id = product["energy_site_id"]
|
|
api = EnergySpecific(teslemetry.energy, site_id)
|
|
live_coordinator = TeslemetryEnergySiteLiveCoordinator(hass, api)
|
|
info_coordinator = TeslemetryEnergySiteInfoCoordinator(hass, api, product)
|
|
device = DeviceInfo(
|
|
identifiers={(DOMAIN, str(site_id))},
|
|
manufacturer="Tesla",
|
|
configuration_url="https://teslemetry.com/console",
|
|
name=product.get("site_name", "Energy Site"),
|
|
)
|
|
|
|
energysites.append(
|
|
TeslemetryEnergyData(
|
|
api=api,
|
|
live_coordinator=live_coordinator,
|
|
info_coordinator=info_coordinator,
|
|
id=site_id,
|
|
device=device,
|
|
)
|
|
)
|
|
|
|
# Run all first refreshes
|
|
await asyncio.gather(
|
|
*(
|
|
vehicle.coordinator.async_config_entry_first_refresh()
|
|
for vehicle in vehicles
|
|
),
|
|
*(
|
|
energysite.live_coordinator.async_config_entry_first_refresh()
|
|
for energysite in energysites
|
|
),
|
|
*(
|
|
energysite.info_coordinator.async_config_entry_first_refresh()
|
|
for energysite in energysites
|
|
),
|
|
)
|
|
|
|
# Setup Platforms
|
|
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = TeslemetryData(
|
|
vehicles, energysites, scopes
|
|
)
|
|
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
|
|
|
return True
|
|
|
|
|
|
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|
"""Unload Teslemetry Config."""
|
|
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
|
|
hass.data[DOMAIN].pop(entry.entry_id)
|
|
return unload_ok
|