jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-26 10:17:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add DataUpdateCoordinator to bmw_connected_drive (#67003)

Browse Source 
Co-authored-by: rikroe <rikroe@users.noreply.github.com>
This commit is contained in:
rikroe 2022-04-21 10:13:09 +02:00 committed by GitHub
parent 089f7279bc
commit 80653463bf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 344 additions and 411 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -118,6 +118,7 @@ omit =
homeassistant/components/bmw_connected_drive/__init__.py
homeassistant/components/bmw_connected_drive/binary_sensor.py
homeassistant/components/bmw_connected_drive/button.py
homeassistant/components/bmw_connected_drive/coordinator.py
homeassistant/components/bmw_connected_drive/device_tracker.py
homeassistant/components/bmw_connected_drive/lock.py
homeassistant/components/bmw_connected_drive/notify.py

200
homeassistant/components/bmw_connected_drive/__init__.py
View File

@ -1,47 +1,30 @@
"""Reads vehicle status from BMW connected drive portal."""
from __future__ import annotations
from collections.abc import Callable
import logging
from typing import Any
from bimmer_connected.account import ConnectedDriveAccount
from bimmer_connected.country_selector import get_region_from_name
from bimmer_connected.vehicle import ConnectedDriveVehicle
import voluptuous as vol
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_DEVICE_ID,
CONF_ENTITY_ID,
CONF_NAME,
CONF_PASSWORD,
CONF_REGION,
CONF_USERNAME,
Platform,
)
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import DeviceInfo, Entity
from homeassistant.helpers.event import track_utc_time_change
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import slugify
import homeassistant.util.dt as dt_util
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import (
ATTRIBUTION,
CONF_ACCOUNT,
CONF_READ_ONLY,
DATA_ENTRIES,
DATA_HASS_CONFIG,
)
_LOGGER = logging.getLogger(__name__)
DOMAIN = "bmw_connected_drive"
ATTR_VIN = "vin"
from .const import ATTR_VIN, ATTRIBUTION, CONF_READ_ONLY, DATA_HASS_CONFIG, DOMAIN
from .coordinator import BMWDataUpdateCoordinator
CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False)
@ -64,12 +47,9 @@ PLATFORMS = [
Platform.NOTIFY,
Platform.SENSOR,
]
UPDATE_INTERVAL = 5 # in minutes
SERVICE_UPDATE_STATE = "update_state"
UNDO_UPDATE_LISTENER = "undo_update_listener"
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the BMW Connected Drive component from configuration.yaml."""
@ -96,37 +76,23 @@ def _async_migrate_options_from_data_if_missing(
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up BMW Connected Drive from a config entry."""
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN].setdefault(DATA_ENTRIES, {})
_async_migrate_options_from_data_if_missing(hass, entry)
try:
account = await hass.async_add_executor_job(
setup_account, entry, hass, entry.data[CONF_USERNAME]
)
except OSError as ex:
raise ConfigEntryNotReady from ex
# Set up one data coordinator per account/config entry
coordinator = BMWDataUpdateCoordinator(
hass,
username=entry.data[CONF_USERNAME],
password=entry.data[CONF_PASSWORD],
region=entry.data[CONF_REGION],
read_only=entry.options[CONF_READ_ONLY],
)
await coordinator.async_config_entry_first_refresh()
async def _async_update_all(service_call: ServiceCall | None = None) -> None:
"""Update all BMW accounts."""
await hass.async_add_executor_job(_update_all)
def _update_all() -> None:
"""Update all BMW accounts."""
for entry in hass.data[DOMAIN][DATA_ENTRIES].copy().values():
entry[CONF_ACCOUNT].update()
# Add update listener for config entry changes (options)
undo_listener = entry.add_update_listener(update_listener)
hass.data[DOMAIN][DATA_ENTRIES][entry.entry_id] = {
CONF_ACCOUNT: account,
UNDO_UPDATE_LISTENER: undo_listener,
}
await _async_update_all()
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = coordinator
# Set up all platforms except notify
hass.config_entries.async_setup_platforms(
entry, [platform for platform in PLATFORMS if platform != Platform.NOTIFY]
)
@ -138,11 +104,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass,
Platform.NOTIFY,
DOMAIN,
{CONF_NAME: DOMAIN},
{CONF_NAME: DOMAIN, CONF_ENTITY_ID: entry.entry_id},
hass.data[DOMAIN][DATA_HASS_CONFIG],
)
)
# Add event listener for option flow changes
entry.async_on_unload(entry.add_update_listener(async_update_options))
return True
@ -152,140 +121,45 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
entry, [platform for platform in PLATFORMS if platform != Platform.NOTIFY]
)
for vehicle in hass.data[DOMAIN][DATA_ENTRIES][entry.entry_id][
CONF_ACCOUNT
].account.vehicles:
hass.services.async_remove(NOTIFY_DOMAIN, slugify(f"{DOMAIN}_{vehicle.name}"))
if unload_ok:
hass.data[DOMAIN][DATA_ENTRIES][entry.entry_id][UNDO_UPDATE_LISTENER]()
hass.data[DOMAIN][DATA_ENTRIES].pop(entry.entry_id)
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
async def async_update_options(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(config_entry.entry_id)
def setup_account(
entry: ConfigEntry, hass: HomeAssistant, name: str
) -> BMWConnectedDriveAccount:
"""Set up a new BMWConnectedDriveAccount based on the config."""
username: str = entry.data[CONF_USERNAME]
password: str = entry.data[CONF_PASSWORD]
region: str = entry.data[CONF_REGION]
read_only: bool = entry.options[CONF_READ_ONLY]
_LOGGER.debug("Adding new account %s", name)
pos = (hass.config.latitude, hass.config.longitude)
cd_account = BMWConnectedDriveAccount(
username, password, region, name, read_only, *pos
)
# update every UPDATE_INTERVAL minutes, starting now
# this should even out the load on the servers
now = dt_util.utcnow()
track_utc_time_change(
hass,
cd_account.update,
minute=range(now.minute % UPDATE_INTERVAL, 60, UPDATE_INTERVAL),
second=now.second,
)
# Initialize
cd_account.update()
return cd_account
class BMWConnectedDriveAccount:
"""Representation of a BMW vehicle."""
def __init__(
self,
username: str,
password: str,
region_str: str,
name: str,
read_only: bool,
lat: float | None = None,
lon: float | None = None,
) -> None:
"""Initialize account."""
region = get_region_from_name(region_str)
self.read_only = read_only
self.account = ConnectedDriveAccount(username, password, region)
self.name = name
self._update_listeners: list[Callable[[], None]] = []
# Set observer position once for older cars to be in range for
# GPS position (pre-7/2014, <2km) and get new data from API
if lat and lon:
self.account.set_observer_position(lat, lon)
self.account.update_vehicle_states()
def update(self, *_: Any) -> None:
"""Update the state of all vehicles.
Notify all listeners about the update.
"""
_LOGGER.debug(
"Updating vehicle state for account %s, notifying %d listeners",
self.name,
len(self._update_listeners),
)
try:
self.account.update_vehicle_states()
for listener in self._update_listeners:
listener()
except OSError as exception:
_LOGGER.error(
"Could not connect to the BMW Connected Drive portal. "
"The vehicle state could not be updated"
)
_LOGGER.exception(exception)
def add_update_listener(self, listener: Callable[[], None]) -> None:
"""Add a listener for update notifications."""
self._update_listeners.append(listener)
class BMWConnectedDriveBaseEntity(Entity):
class BMWConnectedDriveBaseEntity(CoordinatorEntity, Entity):
"""Common base for BMW entities."""
_attr_should_poll = False
coordinator: BMWDataUpdateCoordinator
_attr_attribution = ATTRIBUTION
def __init__(
self,
account: BMWConnectedDriveAccount,
coordinator: BMWDataUpdateCoordinator,
vehicle: ConnectedDriveVehicle,
) -> None:
"""Initialize sensor."""
self._account = account
self._vehicle = vehicle
"""Initialize entity."""
super().__init__(coordinator)
self.vehicle = vehicle
self._attrs: dict[str, Any] = {
"car": self._vehicle.name,
"vin": self._vehicle.vin,
"car": self.vehicle.name,
"vin": self.vehicle.vin,
}
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, vehicle.vin)},
identifiers={(DOMAIN, self.vehicle.vin)},
manufacturer=vehicle.brand.name,
model=vehicle.name,
name=f"{vehicle.brand.name} {vehicle.name}",
)
def update_callback(self) -> None:
"""Schedule a state update."""
self.schedule_update_ha_state(True)
async def async_added_to_hass(self) -> None:
"""Add callback after being added to hass.
Show latest data after startup.
"""
self._account.add_update_listener(self.update_callback)
"""When entity is added to hass."""
await super().async_added_to_hass()
self._handle_coordinator_update()

181
homeassistant/components/bmw_connected_drive/binary_sensor.py
View File

@ -20,100 +20,37 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util.unit_system import UnitSystem
from . import (
DOMAIN as BMW_DOMAIN,
BMWConnectedDriveAccount,
BMWConnectedDriveBaseEntity,
)
from .const import CONF_ACCOUNT, DATA_ENTRIES, UNIT_MAP
from . import BMWConnectedDriveBaseEntity
from .const import DOMAIN, UNIT_MAP
from .coordinator import BMWDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
def _are_doors_closed(
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
) -> bool:
# device class opening: On means open, Off means closed
_LOGGER.debug("Status of lid: %s", vehicle_state.all_lids_closed)
for lid in vehicle_state.lids:
extra_attributes[lid.name] = lid.state.value
return not vehicle_state.all_lids_closed
def _are_windows_closed(
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
) -> bool:
# device class opening: On means open, Off means closed
for window in vehicle_state.windows:
extra_attributes[window.name] = window.state.value
return not vehicle_state.all_windows_closed
def _are_doors_locked(
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
) -> bool:
# device class lock: On means unlocked, Off means locked
# Possible values: LOCKED, SECURED, SELECTIVE_LOCKED, UNLOCKED
extra_attributes["door_lock_state"] = vehicle_state.door_lock_state.value
extra_attributes["last_update_reason"] = vehicle_state.last_update_reason
return vehicle_state.door_lock_state not in {LockState.LOCKED, LockState.SECURED}
def _are_parking_lights_on(
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
) -> bool:
# device class light: On means light detected, Off means no light
extra_attributes["lights_parking"] = vehicle_state.parking_lights.value
return cast(bool, vehicle_state.are_parking_lights_on)
def _are_problems_detected(
vehicle_state: VehicleStatus,
extra_attributes: dict[str, Any],
unit_system: UnitSystem,
) -> bool:
# device class problem: On means problem detected, Off means no problem
def _condition_based_services(
vehicle_state: VehicleStatus, unit_system: UnitSystem
) -> dict[str, Any]:
extra_attributes = {}
for report in vehicle_state.condition_based_services:
extra_attributes.update(_format_cbs_report(report, unit_system))
return not vehicle_state.are_all_cbs_ok
return extra_attributes
def _check_control_messages(
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
) -> bool:
# device class problem: On means problem detected, Off means no problem
check_control_messages = vehicle_state.check_control_messages
has_check_control_messages = vehicle_state.has_check_control_messages
if has_check_control_messages:
cbs_list = [message.description_short for message in check_control_messages]
def _check_control_messages(vehicle_state: VehicleStatus) -> dict[str, Any]:
extra_attributes: dict[str, Any] = {}
if vehicle_state.has_check_control_messages:
cbs_list = [
message.description_short
for message in vehicle_state.check_control_messages
]
extra_attributes["check_control_messages"] = cbs_list
else:
extra_attributes["check_control_messages"] = "OK"
return cast(bool, vehicle_state.has_check_control_messages)
def _is_vehicle_charging(
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
) -> bool:
# device class power: On means power detected, Off means no power
extra_attributes["charging_status"] = vehicle_state.charging_status.value
extra_attributes[
"last_charging_end_result"
] = vehicle_state.last_charging_end_result
return cast(bool, vehicle_state.charging_status == ChargingState.CHARGING)
def _is_vehicle_plugged_in(
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
) -> bool:
# device class plug: On means device is plugged in,
# Off means device is unplugged
extra_attributes["connection_status"] = vehicle_state.connection_status
return cast(str, vehicle_state.connection_status) == "CONNECTED"
return extra_attributes
def _format_cbs_report(
@ -139,7 +76,7 @@ def _format_cbs_report(
class BMWRequiredKeysMixin:
"""Mixin for required keys."""
value_fn: Callable[[VehicleStatus, dict[str, Any], UnitSystem], bool]
value_fn: Callable[[VehicleStatus], bool]
@dataclass
@ -148,6 +85,8 @@ class BMWBinarySensorEntityDescription(
):
"""Describes BMW binary_sensor entity."""
attr_fn: Callable[[VehicleStatus, UnitSystem], dict[str, Any]] | None = None
SENSOR_TYPES: tuple[BMWBinarySensorEntityDescription, ...] = (
BMWBinarySensorEntityDescription(
@ -155,42 +94,59 @@ SENSOR_TYPES: tuple[BMWBinarySensorEntityDescription, ...] = (
name="Doors",
device_class=BinarySensorDeviceClass.OPENING,
icon="mdi:car-door-lock",
value_fn=_are_doors_closed,
# device class opening: On means open, Off means closed
value_fn=lambda s: not s.all_lids_closed,
attr_fn=lambda s, u: {lid.name: lid.state.value for lid in s.lids},
),
BMWBinarySensorEntityDescription(
key="windows",
name="Windows",
device_class=BinarySensorDeviceClass.OPENING,
icon="mdi:car-door",
value_fn=_are_windows_closed,
# device class opening: On means open, Off means closed
value_fn=lambda s: not s.all_windows_closed,
attr_fn=lambda s, u: {window.name: window.state.value for window in s.windows},
),
BMWBinarySensorEntityDescription(
key="door_lock_state",
name="Door lock state",
device_class=BinarySensorDeviceClass.LOCK,
icon="mdi:car-key",
value_fn=_are_doors_locked,
# device class lock: On means unlocked, Off means locked
# Possible values: LOCKED, SECURED, SELECTIVE_LOCKED, UNLOCKED
value_fn=lambda s: s.door_lock_state
not in {LockState.LOCKED, LockState.SECURED},
attr_fn=lambda s, u: {
"door_lock_state": s.door_lock_state.value,
"last_update_reason": s.last_update_reason,
},
),
BMWBinarySensorEntityDescription(
key="lights_parking",
name="Parking lights",
device_class=BinarySensorDeviceClass.LIGHT,
icon="mdi:car-parking-lights",
value_fn=_are_parking_lights_on,
# device class light: On means light detected, Off means no light
value_fn=lambda s: cast(bool, s.are_parking_lights_on),
attr_fn=lambda s, u: {"lights_parking": s.parking_lights.value},
),
BMWBinarySensorEntityDescription(
key="condition_based_services",
name="Condition based services",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:wrench",
value_fn=_are_problems_detected,
# device class problem: On means problem detected, Off means no problem
value_fn=lambda s: not s.are_all_cbs_ok,
attr_fn=_condition_based_services,
),
BMWBinarySensorEntityDescription(
key="check_control_messages",
name="Control messages",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:car-tire-alert",
value_fn=_check_control_messages,
# device class problem: On means problem detected, Off means no problem
value_fn=lambda s: cast(bool, s.has_check_control_messages),
attr_fn=lambda s, u: _check_control_messages(s),
),
# electric
BMWBinarySensorEntityDescription(
@ -198,14 +154,20 @@ SENSOR_TYPES: tuple[BMWBinarySensorEntityDescription, ...] = (
name="Charging status",
device_class=BinarySensorDeviceClass.BATTERY_CHARGING,
icon="mdi:ev-station",
value_fn=_is_vehicle_charging,
# device class power: On means power detected, Off means no power
value_fn=lambda s: cast(bool, s.charging_status == ChargingState.CHARGING),
attr_fn=lambda s, u: {
"charging_status": s.charging_status.value,
"last_charging_end_result": s.last_charging_end_result,
},
),
BMWBinarySensorEntityDescription(
key="connection_status",
name="Connection status",
device_class=BinarySensorDeviceClass.PLUG,
icon="mdi:car-electric",
value_fn=_is_vehicle_plugged_in,
value_fn=lambda s: cast(str, s.connection_status) == "CONNECTED",
attr_fn=lambda s, u: {"connection_status": s.connection_status},
),
)
@ -216,17 +178,15 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the BMW ConnectedDrive binary sensors from config entry."""
account: BMWConnectedDriveAccount = hass.data[BMW_DOMAIN][DATA_ENTRIES][
config_entry.entry_id
][CONF_ACCOUNT]
coordinator: BMWDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
entities = [
BMWConnectedDriveSensor(account, vehicle, description, hass.config.units)
for vehicle in account.account.vehicles
BMWConnectedDriveSensor(coordinator, vehicle, description, hass.config.units)
for vehicle in coordinator.account.vehicles
for description in SENSOR_TYPES
if description.key in vehicle.available_attributes
]
async_add_entities(entities, True)
async_add_entities(entities)
class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, BinarySensorEntity):
@ -236,26 +196,35 @@ class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, BinarySensorEntity):
def __init__(
self,
account: BMWConnectedDriveAccount,
coordinator: BMWDataUpdateCoordinator,
vehicle: ConnectedDriveVehicle,
description: BMWBinarySensorEntityDescription,
unit_system: UnitSystem,
) -> None:
"""Initialize sensor."""
super().__init__(account, vehicle)
super().__init__(coordinator, vehicle)
self.entity_description = description
self._unit_system = unit_system
self._attr_name = f"{vehicle.name} {description.key}"
self._attr_unique_id = f"{vehicle.vin}-{description.key}"
def update(self) -> None:
"""Read new state data from the library."""
_LOGGER.debug("Updating binary sensors of %s", self._vehicle.name)
vehicle_state = self._vehicle.status
result = self._attrs.copy()
self._attr_is_on = self.entity_description.value_fn(
vehicle_state, result, self._unit_system
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
_LOGGER.debug(
"Updating binary sensor '%s' of %s",
self.entity_description.key,
self.vehicle.name,
)
self._attr_extra_state_attributes = result
vehicle_state = self.vehicle.status
self._attr_is_on = self.entity_description.value_fn(vehicle_state)
if self.entity_description.attr_fn:
self._attr_extra_state_attributes = dict(
self._attrs,
**self.entity_description.attr_fn(vehicle_state, self._unit_system),
)
super()._handle_coordinator_update()

76
homeassistant/components/bmw_connected_drive/button.py
View File

@ -1,10 +1,11 @@
"""Support for BMW connected drive button entities."""
from __future__ import annotations
from collections.abc import Callable
from collections.abc import Callable, Coroutine
from dataclasses import dataclass
import logging
from typing import TYPE_CHECKING
from bimmer_connected.remote_services import RemoteServiceStatus
from bimmer_connected.vehicle import ConnectedDriveVehicle
from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
@ -12,12 +13,13 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import (
DOMAIN as BMW_DOMAIN,
BMWConnectedDriveAccount,
BMWConnectedDriveBaseEntity,
)
from .const import CONF_ACCOUNT, DATA_ENTRIES
from . import BMWConnectedDriveBaseEntity
from .const import DOMAIN
if TYPE_CHECKING:
from .coordinator import BMWDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
@dataclass
@ -25,10 +27,8 @@ class BMWButtonEntityDescription(ButtonEntityDescription):
"""Class describing BMW button entities."""
enabled_when_read_only: bool = False
remote_function: Callable[
[ConnectedDriveVehicle], RemoteServiceStatus
] | None = None
account_function: Callable[[BMWConnectedDriveAccount], None] | None = None
remote_function: str | None = None
account_function: Callable[[BMWDataUpdateCoordinator], Coroutine] | None = None
BUTTON_TYPES: tuple[BMWButtonEntityDescription, ...] = (
@ -36,37 +36,37 @@ BUTTON_TYPES: tuple[BMWButtonEntityDescription, ...] = (
key="light_flash",
icon="mdi:car-light-alert",
name="Flash Lights",
remote_function=lambda vehicle: vehicle.remote_services.trigger_remote_light_flash(),
remote_function="trigger_remote_light_flash",
),
BMWButtonEntityDescription(
key="sound_horn",
icon="mdi:bullhorn",
name="Sound Horn",
remote_function=lambda vehicle: vehicle.remote_services.trigger_remote_horn(),
remote_function="trigger_remote_horn",
),
BMWButtonEntityDescription(
key="activate_air_conditioning",
icon="mdi:hvac",
name="Activate Air Conditioning",
remote_function=lambda vehicle: vehicle.remote_services.trigger_remote_air_conditioning(),
remote_function="trigger_remote_air_conditioning",
),
BMWButtonEntityDescription(
key="deactivate_air_conditioning",
icon="mdi:hvac-off",
name="Deactivate Air Conditioning",
remote_function=lambda vehicle: vehicle.remote_services.trigger_remote_air_conditioning_stop(),
remote_function="trigger_remote_air_conditioning_stop",
),
BMWButtonEntityDescription(
key="find_vehicle",
icon="mdi:crosshairs-question",
name="Find Vehicle",
remote_function=lambda vehicle: vehicle.remote_services.trigger_remote_vehicle_finder(),
remote_function="trigger_remote_vehicle_finder",
),
BMWButtonEntityDescription(
key="refresh",
icon="mdi:refresh",
name="Refresh from cloud",
account_function=lambda account: account.update(),
account_function=lambda coordinator: coordinator.async_request_refresh(),
enabled_when_read_only=True,
),
)
@ -78,18 +78,17 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the BMW ConnectedDrive buttons from config entry."""
account: BMWConnectedDriveAccount = hass.data[BMW_DOMAIN][DATA_ENTRIES][
config_entry.entry_id
][CONF_ACCOUNT]
coordinator: BMWDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
entities: list[BMWButton] = []
for vehicle in account.account.vehicles:
for vehicle in coordinator.account.vehicles:
entities.extend(
[
BMWButton(account, vehicle, description)
BMWButton(coordinator, vehicle, description)
for description in BUTTON_TYPES
if not account.read_only
or (account.read_only and description.enabled_when_read_only)
if not coordinator.read_only
or (coordinator.read_only and description.enabled_when_read_only)
]
)
@ -103,20 +102,35 @@ class BMWButton(BMWConnectedDriveBaseEntity, ButtonEntity):
def __init__(
self,
account: BMWConnectedDriveAccount,
coordinator: BMWDataUpdateCoordinator,
vehicle: ConnectedDriveVehicle,
description: BMWButtonEntityDescription,
) -> None:
"""Initialize BMW vehicle sensor."""
super().__init__(account, vehicle)
super().__init__(coordinator, vehicle)
self.entity_description = description
self._attr_name = f"{vehicle.name} {description.name}"
self._attr_unique_id = f"{vehicle.vin}-{description.key}"
def press(self) -> None:
"""Process the button press."""
async def async_press(self) -> None:
"""Press the button."""
if self.entity_description.remote_function:
self.entity_description.remote_function(self._vehicle)
await self.hass.async_add_executor_job(
getattr(
self.vehicle.remote_services,
self.entity_description.remote_function,
)
)
elif self.entity_description.account_function:
self.entity_description.account_function(self._account)
_LOGGER.warning(
"The 'Refresh from cloud' button is deprecated. Use the 'homeassistant.update_entity' "
"service with any BMW entity for a full reload. See https://www.home-assistant.io/"
"integrations/bmw_connected_drive/#update-the-state--refresh-from-api for details"
)
await self.entity_description.account_function(self.coordinator)
# Always update HA states after a button was executed.
# BMW remote services that change the vehicle's state update the local object
# when executing the service, so only the HA state machine needs further updates.
self.coordinator.notify_listeners()

4
homeassistant/components/bmw_connected_drive/const.py
View File

@ -6,17 +6,17 @@ from homeassistant.const import (
VOLUME_LITERS,
)
DOMAIN = "bmw_connected_drive"
ATTRIBUTION = "Data provided by BMW Connected Drive"
ATTR_DIRECTION = "direction"
ATTR_VIN = "vin"
CONF_ALLOWED_REGIONS = ["china", "north_america", "rest_of_world"]
CONF_READ_ONLY = "read_only"
CONF_ACCOUNT = "account"
DATA_HASS_CONFIG = "hass_config"
DATA_ENTRIES = "entries"
UNIT_MAP = {
"KILOMETERS": LENGTH_KILOMETERS,

74
homeassistant/components/bmw_connected_drive/coordinator.py Normal file
View File

@ -0,0 +1,74 @@
"""Coordinator for BMW."""
from __future__ import annotations
from datetime import timedelta
import logging
import async_timeout
from bimmer_connected.account import ConnectedDriveAccount
from bimmer_connected.country_selector import get_region_from_name
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
SCAN_INTERVAL = timedelta(seconds=300)
_LOGGER = logging.getLogger(__name__)
class BMWDataUpdateCoordinator(DataUpdateCoordinator):
"""Class to manage fetching BMW data."""
account: ConnectedDriveAccount
def __init__(
self,
hass: HomeAssistant,
*,
username: str,
password: str,
region: str,
read_only: bool = False,
) -> None:
"""Initialize account-wide BMW data updater."""
# Storing username & password in coordinator is needed until a new library version
# that does not do blocking IO on init.
self._username = username
self._password = password
self._region = get_region_from_name(region)
self.account = None
self.read_only = read_only
super().__init__(
hass,
_LOGGER,
name=f"{DOMAIN}-{username}",
update_interval=SCAN_INTERVAL,
)
async def _async_update_data(self) -> None:
"""Fetch data from BMW."""
try:
async with async_timeout.timeout(15):
if isinstance(self.account, ConnectedDriveAccount):
# pylint: disable=protected-access
await self.hass.async_add_executor_job(self.account._get_vehicles)
else:
self.account = await self.hass.async_add_executor_job(
ConnectedDriveAccount,
self._username,
self._password,
self._region,
)
self.account.set_observer_position(
self.hass.config.latitude, self.hass.config.longitude
)
except OSError as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
def notify_listeners(self) -> None:
"""Notify all listeners to refresh HA state machine."""
for update_callback in self._listeners:
update_callback()

53
homeassistant/components/bmw_connected_drive/device_tracker.py
View File

@ -12,12 +12,9 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import (
DOMAIN as BMW_DOMAIN,
BMWConnectedDriveAccount,
BMWConnectedDriveBaseEntity,
)
from .const import ATTR_DIRECTION, CONF_ACCOUNT, DATA_ENTRIES
from . import BMWConnectedDriveBaseEntity
from .const import ATTR_DIRECTION, DOMAIN
from .coordinator import BMWDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
@ -28,20 +25,18 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the BMW ConnectedDrive tracker from config entry."""
account: BMWConnectedDriveAccount = hass.data[BMW_DOMAIN][DATA_ENTRIES][
config_entry.entry_id
][CONF_ACCOUNT]
coordinator: BMWDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
entities: list[BMWDeviceTracker] = []
for vehicle in account.account.vehicles:
entities.append(BMWDeviceTracker(account, vehicle))
for vehicle in coordinator.account.vehicles:
entities.append(BMWDeviceTracker(coordinator, vehicle))
if not vehicle.is_vehicle_tracking_enabled:
_LOGGER.info(
"Tracking is (currently) disabled for vehicle %s (%s), defaulting to unknown",
vehicle.name,
vehicle.vin,
)
async_add_entities(entities, True)
async_add_entities(entities)
class BMWDeviceTracker(BMWConnectedDriveBaseEntity, TrackerEntity):
@ -52,39 +47,39 @@ class BMWDeviceTracker(BMWConnectedDriveBaseEntity, TrackerEntity):
def __init__(
self,
account: BMWConnectedDriveAccount,
coordinator: BMWDataUpdateCoordinator,
vehicle: ConnectedDriveVehicle,
) -> None:
"""Initialize the Tracker."""
super().__init__(account, vehicle)
super().__init__(coordinator, vehicle)
self._attr_unique_id = vehicle.vin
self._location = pos if (pos := vehicle.status.gps_position) else None
self._attr_name = vehicle.name
@property
def extra_state_attributes(self) -> dict:
"""Return entity specific state attributes."""
return dict(self._attrs, **{ATTR_DIRECTION: self.vehicle.status.gps_heading})
@property
def latitude(self) -> float | None:
"""Return latitude value of the device."""
return self._location[0] if self._location else None
return (
self.vehicle.status.gps_position[0]
if self.vehicle.is_vehicle_tracking_enabled
else None
)
@property
def longitude(self) -> float | None:
"""Return longitude value of the device."""
return self._location[1] if self._location else None
return (
self.vehicle.status.gps_position[1]
if self.vehicle.is_vehicle_tracking_enabled
else None
)
@property
def source_type(self) -> Literal["gps"]:
"""Return the source type, eg gps or router, of the device."""
return SOURCE_TYPE_GPS
def update(self) -> None:
"""Update state of the device tracker."""
_LOGGER.debug("Updating device tracker of %s", self._vehicle.name)
state_attrs = self._attrs
state_attrs[ATTR_DIRECTION] = self._vehicle.status.gps_heading
self._attr_extra_state_attributes = state_attrs
self._location = (
self._vehicle.status.gps_position
if self._vehicle.is_vehicle_tracking_enabled
else None
)

90
homeassistant/components/bmw_connected_drive/lock.py
View File

@ -1,4 +1,6 @@
"""Support for BMW car locks with BMW ConnectedDrive."""
from __future__ import annotations
import logging
from typing import Any
@ -7,15 +9,12 @@ from bimmer_connected.vehicle_status import LockState
from homeassistant.components.lock import LockEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import (
DOMAIN as BMW_DOMAIN,
BMWConnectedDriveAccount,
BMWConnectedDriveBaseEntity,
)
from .const import CONF_ACCOUNT, DATA_ENTRIES
from . import BMWConnectedDriveBaseEntity
from .const import DOMAIN
from .coordinator import BMWDataUpdateCoordinator
DOOR_LOCK_STATE = "door_lock_state"
_LOGGER = logging.getLogger(__name__)
@ -27,16 +26,14 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the BMW ConnectedDrive binary sensors from config entry."""
account: BMWConnectedDriveAccount = hass.data[BMW_DOMAIN][DATA_ENTRIES][
config_entry.entry_id
][CONF_ACCOUNT]
coordinator: BMWDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
if not account.read_only:
entities = [
BMWLock(account, vehicle, "lock", "BMW lock")
for vehicle in account.account.vehicles
]
async_add_entities(entities, True)
entities: list[BMWLock] = []
for vehicle in coordinator.account.vehicles:
if not coordinator.read_only:
entities.append(BMWLock(coordinator, vehicle, "lock", "BMW lock"))
async_add_entities(entities)
class BMWLock(BMWConnectedDriveBaseEntity, LockEntity):
@ -44,13 +41,13 @@ class BMWLock(BMWConnectedDriveBaseEntity, LockEntity):
def __init__(
self,
account: BMWConnectedDriveAccount,
coordinator: BMWDataUpdateCoordinator,
vehicle: ConnectedDriveVehicle,
attribute: str,
sensor_name: str,
) -> None:
"""Initialize the lock."""
super().__init__(account, vehicle)
super().__init__(coordinator, vehicle)
self._attribute = attribute
self._attr_name = f"{vehicle.name} {attribute}"
@ -60,38 +57,43 @@ class BMWLock(BMWConnectedDriveBaseEntity, LockEntity):
def lock(self, **kwargs: Any) -> None:
"""Lock the car."""
_LOGGER.debug("%s: locking doors", self._vehicle.name)
# Optimistic state set here because it takes some time before the
# update callback response
self._attr_is_locked = True
self.schedule_update_ha_state()
self._vehicle.remote_services.trigger_remote_door_lock()
_LOGGER.debug("%s: locking doors", self.vehicle.name)
# Only update the HA state machine if the vehicle reliably reports its lock state
if self.door_lock_state_available:
# Optimistic state set here because it takes some time before the
# update callback response
self._attr_is_locked = True
self.schedule_update_ha_state()
self.vehicle.remote_services.trigger_remote_door_lock()
def unlock(self, **kwargs: Any) -> None:
"""Unlock the car."""
_LOGGER.debug("%s: unlocking doors", self._vehicle.name)
# Optimistic state set here because it takes some time before the
# update callback response
self._attr_is_locked = False
self.schedule_update_ha_state()
self._vehicle.remote_services.trigger_remote_door_unlock()
_LOGGER.debug("%s: unlocking doors", self.vehicle.name)
# Only update the HA state machine if the vehicle reliably reports its lock state
if self.door_lock_state_available:
# Optimistic state set here because it takes some time before the
# update callback response
self._attr_is_locked = False
self.schedule_update_ha_state()
self.vehicle.remote_services.trigger_remote_door_unlock()
def update(self) -> None:
"""Update state of the lock."""
_LOGGER.debug(
"Updating lock data for '%s' of %s", self._attribute, self._vehicle.name
)
vehicle_state = self._vehicle.status
if not self.door_lock_state_available:
self._attr_is_locked = None
else:
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
_LOGGER.debug("Updating lock data of %s", self.vehicle.name)
# Only update the HA state machine if the vehicle reliably reports its lock state
if self.door_lock_state_available:
vehicle_state = self.vehicle.status
self._attr_is_locked = vehicle_state.door_lock_state in {
LockState.LOCKED,
LockState.SECURED,
}
self._attr_extra_state_attributes = dict(
self._attrs,
**{
"door_lock_state": vehicle_state.door_lock_state.value,
"last_update_reason": vehicle_state.last_update_reason,
},
)
result = self._attrs.copy()
if self.door_lock_state_available:
result["door_lock_state"] = vehicle_state.door_lock_state.value
result["last_update_reason"] = vehicle_state.last_update_reason
self._attr_extra_state_attributes = result
super()._handle_coordinator_update()

34
homeassistant/components/bmw_connected_drive/notify.py
View File

@ -11,12 +11,18 @@ from homeassistant.components.notify import (
ATTR_TARGET,
BaseNotificationService,
)
from homeassistant.const import ATTR_LATITUDE, ATTR_LOCATION, ATTR_LONGITUDE, ATTR_NAME
from homeassistant.const import (
ATTR_LATITUDE,
ATTR_LOCATION,
ATTR_LONGITUDE,
ATTR_NAME,
CONF_ENTITY_ID,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DOMAIN as BMW_DOMAIN, BMWConnectedDriveAccount
from .const import CONF_ACCOUNT, DATA_ENTRIES
from .const import DOMAIN
from .coordinator import BMWDataUpdateCoordinator
ATTR_LAT = "lat"
ATTR_LOCATION_ATTRIBUTES = ["street", "city", "postal_code", "country"]
@ -33,26 +39,22 @@ def get_service(
discovery_info: DiscoveryInfoType | None = None,
) -> BMWNotificationService:
"""Get the BMW notification service."""
accounts: list[BMWConnectedDriveAccount] = [
e[CONF_ACCOUNT] for e in hass.data[BMW_DOMAIN][DATA_ENTRIES].values()
coordinator: BMWDataUpdateCoordinator = hass.data[DOMAIN][
(discovery_info or {})[CONF_ENTITY_ID]
]
_LOGGER.debug("Found BMW accounts: %s", ", ".join([a.name for a in accounts]))
svc = BMWNotificationService()
svc.setup(accounts)
return svc
targets = {}
if not coordinator.read_only:
targets.update({v.name: v for v in coordinator.account.vehicles})
return BMWNotificationService(targets)
class BMWNotificationService(BaseNotificationService):
"""Send Notifications to BMW."""
def __init__(self) -> None:
def __init__(self, targets: dict[str, ConnectedDriveVehicle]) -> None:
"""Set up the notification service."""
self.targets: dict[str, ConnectedDriveVehicle] = {}
def setup(self, accounts: list[BMWConnectedDriveAccount]) -> None:
"""Get the BMW vehicle(s) for the account(s)."""
for account in accounts:
self.targets.update({v.name: v for v in account.account.vehicles})
self.targets: dict[str, ConnectedDriveVehicle] = targets
def send_message(self, message: str = "", **kwargs: Any) -> None:
"""Send a message or POI to the car."""

42
homeassistant/components/bmw_connected_drive/sensor.py
View File

@ -22,17 +22,14 @@ from homeassistant.const import (
VOLUME_GALLONS,
VOLUME_LITERS,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.util.unit_system import UnitSystem
from . import (
DOMAIN as BMW_DOMAIN,
BMWConnectedDriveAccount,
BMWConnectedDriveBaseEntity,
)
from .const import CONF_ACCOUNT, DATA_ENTRIES, UNIT_MAP
from . import BMWConnectedDriveBaseEntity
from .const import DOMAIN, UNIT_MAP
from .coordinator import BMWDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
@ -135,21 +132,20 @@ async def async_setup_entry(
) -> None:
"""Set up the BMW ConnectedDrive sensors from config entry."""
unit_system = hass.config.units
account: BMWConnectedDriveAccount = hass.data[BMW_DOMAIN][DATA_ENTRIES][
config_entry.entry_id
][CONF_ACCOUNT]
coordinator: BMWDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
entities: list[BMWConnectedDriveSensor] = []
for vehicle in account.account.vehicles:
for vehicle in coordinator.account.vehicles:
entities.extend(
[
BMWConnectedDriveSensor(account, vehicle, description, unit_system)
BMWConnectedDriveSensor(coordinator, vehicle, description, unit_system)
for attribute_name in vehicle.available_attributes
if (description := SENSOR_TYPES.get(attribute_name))
]
)
async_add_entities(entities, True)
async_add_entities(entities)
class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, SensorEntity):
@ -159,13 +155,13 @@ class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, SensorEntity):
def __init__(
self,
account: BMWConnectedDriveAccount,
coordinator: BMWDataUpdateCoordinator,
vehicle: ConnectedDriveVehicle,
description: BMWSensorEntityDescription,
unit_system: UnitSystem,
) -> None:
"""Initialize BMW vehicle sensor."""
super().__init__(account, vehicle)
super().__init__(coordinator, vehicle)
self.entity_description = description
self._attr_name = f"{vehicle.name} {description.key}"
@ -176,8 +172,14 @@ class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, SensorEntity):
else:
self._attr_native_unit_of_measurement = description.unit_metric
@property
def native_value(self) -> StateType:
"""Return the state."""
state = getattr(self._vehicle.status, self.entity_description.key)
return cast(StateType, self.entity_description.value(state, self.hass))
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
_LOGGER.debug(
"Updating sensor '%s' of %s", self.entity_description.key, self.vehicle.name
)
state = getattr(self.vehicle.status, self.entity_description.key)
self._attr_native_value = cast(
StateType, self.entity_description.value(state, self.hass)
)
super()._handle_coordinator_update()