mirror of
https://github.com/home-assistant/core.git
synced 2025-07-14 16:57:10 +00:00
Update BMW connected drive to async (#71827)
* Change BMW connected drive to async * Fix coordinator exceptions, fix tests * Fix using deprecated property * Write HA state directly * Remove code that cannnot throw Exceptions from try/except * Use public async_write_ha_state Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Fix login using refresh_token if token expired * MyPy fixes * Fix pytest, bump dependency * Replace OptionFlow listener with explicit refresh * Remove uneeded async_get_entry * Update test to include change on OptionsFlow * Bump bimmer_connected to 0.9.0 * Migrate renamed entitity unique_ids * Don't replace async_migrate_entries, add tests * Rename existing_entry to existing_entry_id Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Update tests * Import full EntityRegistry * Fix comment * Increase timeout to 60s * Rely on library timeout Co-authored-by: rikroe <rikroe@users.noreply.github.com> Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
parent
f33151ff8b
commit
cd769a55c2
@ -1,31 +1,27 @@
|
||||
"""Reads vehicle status from BMW connected drive portal."""
|
||||
"""Reads vehicle status from MyBMW portal."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from bimmer_connected.vehicle import ConnectedDriveVehicle
|
||||
from bimmer_connected.vehicle import MyBMWVehicle
|
||||
import voluptuous as vol
|
||||
|
||||
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.const import CONF_DEVICE_ID, CONF_ENTITY_ID, CONF_NAME, Platform
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import discovery
|
||||
from homeassistant.helpers import discovery, entity_registry as er
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity import DeviceInfo, Entity
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import ATTR_VIN, ATTRIBUTION, CONF_READ_ONLY, DATA_HASS_CONFIG, DOMAIN
|
||||
from .coordinator import BMWDataUpdateCoordinator
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False)
|
||||
|
||||
SERVICE_SCHEMA = vol.Schema(
|
||||
@ -74,18 +70,56 @@ def _async_migrate_options_from_data_if_missing(
|
||||
hass.config_entries.async_update_entry(entry, data=data, options=options)
|
||||
|
||||
|
||||
async def _async_migrate_entries(
|
||||
hass: HomeAssistant, config_entry: ConfigEntry
|
||||
) -> bool:
|
||||
"""Migrate old entry."""
|
||||
entity_registry = er.async_get(hass)
|
||||
|
||||
@callback
|
||||
def update_unique_id(entry: er.RegistryEntry) -> dict[str, str] | None:
|
||||
replacements = {
|
||||
"charging_level_hv": "remaining_battery_percent",
|
||||
"fuel_percent": "remaining_fuel_percent",
|
||||
}
|
||||
if (key := entry.unique_id.split("-")[-1]) in replacements:
|
||||
new_unique_id = entry.unique_id.replace(key, replacements[key])
|
||||
_LOGGER.debug(
|
||||
"Migrating entity '%s' unique_id from '%s' to '%s'",
|
||||
entry.entity_id,
|
||||
entry.unique_id,
|
||||
new_unique_id,
|
||||
)
|
||||
if existing_entity_id := entity_registry.async_get_entity_id(
|
||||
entry.domain, entry.platform, new_unique_id
|
||||
):
|
||||
_LOGGER.debug(
|
||||
"Cannot migrate to unique_id '%s', already exists for '%s'",
|
||||
new_unique_id,
|
||||
existing_entity_id,
|
||||
)
|
||||
return None
|
||||
return {
|
||||
"new_unique_id": new_unique_id,
|
||||
}
|
||||
return None
|
||||
|
||||
await er.async_migrate_entries(hass, config_entry.entry_id, update_unique_id)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up BMW Connected Drive from a config entry."""
|
||||
|
||||
_async_migrate_options_from_data_if_missing(hass, entry)
|
||||
|
||||
await _async_migrate_entries(hass, entry)
|
||||
|
||||
# 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],
|
||||
entry=entry,
|
||||
)
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
@ -109,9 +143,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
)
|
||||
)
|
||||
|
||||
# Add event listener for option flow changes
|
||||
entry.async_on_unload(entry.add_update_listener(async_update_options))
|
||||
|
||||
return True
|
||||
|
||||
|
||||
@ -127,20 +158,16 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
return unload_ok
|
||||
|
||||
|
||||
async def async_update_options(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
|
||||
"""Handle options update."""
|
||||
await hass.config_entries.async_reload(config_entry.entry_id)
|
||||
|
||||
|
||||
class BMWConnectedDriveBaseEntity(CoordinatorEntity[BMWDataUpdateCoordinator], Entity):
|
||||
class BMWBaseEntity(CoordinatorEntity[BMWDataUpdateCoordinator]):
|
||||
"""Common base for BMW entities."""
|
||||
|
||||
coordinator: BMWDataUpdateCoordinator
|
||||
_attr_attribution = ATTRIBUTION
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: BMWDataUpdateCoordinator,
|
||||
vehicle: ConnectedDriveVehicle,
|
||||
vehicle: MyBMWVehicle,
|
||||
) -> None:
|
||||
"""Initialize entity."""
|
||||
super().__init__(coordinator)
|
||||
|
@ -1,18 +1,15 @@
|
||||
"""Reads vehicle status from BMW connected drive portal."""
|
||||
"""Reads vehicle status from BMW MyBMW portal."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
from typing import Any, cast
|
||||
from typing import Any
|
||||
|
||||
from bimmer_connected.vehicle import ConnectedDriveVehicle
|
||||
from bimmer_connected.vehicle_status import (
|
||||
ChargingState,
|
||||
ConditionBasedServiceReport,
|
||||
LockState,
|
||||
VehicleStatus,
|
||||
)
|
||||
from bimmer_connected.vehicle import MyBMWVehicle
|
||||
from bimmer_connected.vehicle.doors_windows import LockState
|
||||
from bimmer_connected.vehicle.fuel_and_battery import ChargingState
|
||||
from bimmer_connected.vehicle.reports import ConditionBasedService
|
||||
|
||||
from homeassistant.components.binary_sensor import (
|
||||
BinarySensorDeviceClass,
|
||||
@ -24,7 +21,7 @@ from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.util.unit_system import UnitSystem
|
||||
|
||||
from . import BMWConnectedDriveBaseEntity
|
||||
from . import BMWBaseEntity
|
||||
from .const import DOMAIN, UNIT_MAP
|
||||
from .coordinator import BMWDataUpdateCoordinator
|
||||
|
||||
@ -32,20 +29,20 @@ _LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _condition_based_services(
|
||||
vehicle_state: VehicleStatus, unit_system: UnitSystem
|
||||
vehicle: MyBMWVehicle, unit_system: UnitSystem
|
||||
) -> dict[str, Any]:
|
||||
extra_attributes = {}
|
||||
for report in vehicle_state.condition_based_services:
|
||||
for report in vehicle.condition_based_services.messages:
|
||||
extra_attributes.update(_format_cbs_report(report, unit_system))
|
||||
return extra_attributes
|
||||
|
||||
|
||||
def _check_control_messages(vehicle_state: VehicleStatus) -> dict[str, Any]:
|
||||
def _check_control_messages(vehicle: MyBMWVehicle) -> dict[str, Any]:
|
||||
extra_attributes: dict[str, Any] = {}
|
||||
if vehicle_state.has_check_control_messages:
|
||||
if vehicle.check_control_messages.has_check_control_messages:
|
||||
cbs_list = [
|
||||
message.description_short
|
||||
for message in vehicle_state.check_control_messages
|
||||
for message in vehicle.check_control_messages.messages
|
||||
]
|
||||
extra_attributes["check_control_messages"] = cbs_list
|
||||
else:
|
||||
@ -54,18 +51,18 @@ def _check_control_messages(vehicle_state: VehicleStatus) -> dict[str, Any]:
|
||||
|
||||
|
||||
def _format_cbs_report(
|
||||
report: ConditionBasedServiceReport, unit_system: UnitSystem
|
||||
report: ConditionBasedService, unit_system: UnitSystem
|
||||
) -> dict[str, Any]:
|
||||
result: dict[str, Any] = {}
|
||||
service_type = report.service_type.lower().replace("_", " ")
|
||||
result[f"{service_type} status"] = report.state.value
|
||||
if report.due_date is not None:
|
||||
result[f"{service_type} date"] = report.due_date.strftime("%Y-%m-%d")
|
||||
if report.due_distance is not None:
|
||||
if report.due_distance.value and report.due_distance.unit:
|
||||
distance = round(
|
||||
unit_system.length(
|
||||
report.due_distance[0],
|
||||
UNIT_MAP.get(report.due_distance[1], report.due_distance[1]),
|
||||
report.due_distance.value,
|
||||
UNIT_MAP.get(report.due_distance.unit, report.due_distance.unit),
|
||||
)
|
||||
)
|
||||
result[f"{service_type} distance"] = f"{distance} {unit_system.length_unit}"
|
||||
@ -76,7 +73,7 @@ def _format_cbs_report(
|
||||
class BMWRequiredKeysMixin:
|
||||
"""Mixin for required keys."""
|
||||
|
||||
value_fn: Callable[[VehicleStatus], bool]
|
||||
value_fn: Callable[[MyBMWVehicle], bool]
|
||||
|
||||
|
||||
@dataclass
|
||||
@ -85,7 +82,7 @@ class BMWBinarySensorEntityDescription(
|
||||
):
|
||||
"""Describes BMW binary_sensor entity."""
|
||||
|
||||
attr_fn: Callable[[VehicleStatus, UnitSystem], dict[str, Any]] | None = None
|
||||
attr_fn: Callable[[MyBMWVehicle, UnitSystem], dict[str, Any]] | None = None
|
||||
|
||||
|
||||
SENSOR_TYPES: tuple[BMWBinarySensorEntityDescription, ...] = (
|
||||
@ -95,8 +92,10 @@ SENSOR_TYPES: tuple[BMWBinarySensorEntityDescription, ...] = (
|
||||
device_class=BinarySensorDeviceClass.OPENING,
|
||||
icon="mdi:car-door-lock",
|
||||
# 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},
|
||||
value_fn=lambda v: not v.doors_and_windows.all_lids_closed,
|
||||
attr_fn=lambda v, u: {
|
||||
lid.name: lid.state.value for lid in v.doors_and_windows.lids
|
||||
},
|
||||
),
|
||||
BMWBinarySensorEntityDescription(
|
||||
key="windows",
|
||||
@ -104,8 +103,10 @@ SENSOR_TYPES: tuple[BMWBinarySensorEntityDescription, ...] = (
|
||||
device_class=BinarySensorDeviceClass.OPENING,
|
||||
icon="mdi:car-door",
|
||||
# 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},
|
||||
value_fn=lambda v: not v.doors_and_windows.all_windows_closed,
|
||||
attr_fn=lambda v, u: {
|
||||
window.name: window.state.value for window in v.doors_and_windows.windows
|
||||
},
|
||||
),
|
||||
BMWBinarySensorEntityDescription(
|
||||
key="door_lock_state",
|
||||
@ -114,29 +115,19 @@ SENSOR_TYPES: tuple[BMWBinarySensorEntityDescription, ...] = (
|
||||
icon="mdi:car-key",
|
||||
# device class lock: On means unlocked, Off means locked
|
||||
# Possible values: LOCKED, SECURED, SELECTIVE_LOCKED, UNLOCKED
|
||||
value_fn=lambda s: s.door_lock_state
|
||||
value_fn=lambda v: v.doors_and_windows.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,
|
||||
attr_fn=lambda v, u: {
|
||||
"door_lock_state": v.doors_and_windows.door_lock_state.value
|
||||
},
|
||||
),
|
||||
BMWBinarySensorEntityDescription(
|
||||
key="lights_parking",
|
||||
name="Parking lights",
|
||||
device_class=BinarySensorDeviceClass.LIGHT,
|
||||
icon="mdi:car-parking-lights",
|
||||
# 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",
|
||||
# device class problem: On means problem detected, Off means no problem
|
||||
value_fn=lambda s: not s.are_all_cbs_ok,
|
||||
value_fn=lambda v: v.condition_based_services.is_service_required,
|
||||
attr_fn=_condition_based_services,
|
||||
),
|
||||
BMWBinarySensorEntityDescription(
|
||||
@ -145,8 +136,8 @@ SENSOR_TYPES: tuple[BMWBinarySensorEntityDescription, ...] = (
|
||||
device_class=BinarySensorDeviceClass.PROBLEM,
|
||||
icon="mdi:car-tire-alert",
|
||||
# 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),
|
||||
value_fn=lambda v: v.check_control_messages.has_check_control_messages,
|
||||
attr_fn=lambda v, u: _check_control_messages(v),
|
||||
),
|
||||
# electric
|
||||
BMWBinarySensorEntityDescription(
|
||||
@ -155,10 +146,9 @@ SENSOR_TYPES: tuple[BMWBinarySensorEntityDescription, ...] = (
|
||||
device_class=BinarySensorDeviceClass.BATTERY_CHARGING,
|
||||
icon="mdi:ev-station",
|
||||
# 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,
|
||||
value_fn=lambda v: v.fuel_and_battery.charging_status == ChargingState.CHARGING,
|
||||
attr_fn=lambda v, u: {
|
||||
"charging_status": str(v.fuel_and_battery.charging_status),
|
||||
},
|
||||
),
|
||||
BMWBinarySensorEntityDescription(
|
||||
@ -166,8 +156,7 @@ SENSOR_TYPES: tuple[BMWBinarySensorEntityDescription, ...] = (
|
||||
name="Connection status",
|
||||
device_class=BinarySensorDeviceClass.PLUG,
|
||||
icon="mdi:car-electric",
|
||||
value_fn=lambda s: cast(str, s.connection_status) == "CONNECTED",
|
||||
attr_fn=lambda s, u: {"connection_status": s.connection_status},
|
||||
value_fn=lambda v: v.fuel_and_battery.is_charger_connected,
|
||||
),
|
||||
)
|
||||
|
||||
@ -177,11 +166,11 @@ async def async_setup_entry(
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the BMW ConnectedDrive binary sensors from config entry."""
|
||||
"""Set up the BMW binary sensors from config entry."""
|
||||
coordinator: BMWDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
|
||||
|
||||
entities = [
|
||||
BMWConnectedDriveSensor(coordinator, vehicle, description, hass.config.units)
|
||||
BMWBinarySensor(coordinator, vehicle, description, hass.config.units)
|
||||
for vehicle in coordinator.account.vehicles
|
||||
for description in SENSOR_TYPES
|
||||
if description.key in vehicle.available_attributes
|
||||
@ -189,7 +178,7 @@ async def async_setup_entry(
|
||||
async_add_entities(entities)
|
||||
|
||||
|
||||
class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, BinarySensorEntity):
|
||||
class BMWBinarySensor(BMWBaseEntity, BinarySensorEntity):
|
||||
"""Representation of a BMW vehicle binary sensor."""
|
||||
|
||||
entity_description: BMWBinarySensorEntityDescription
|
||||
@ -197,7 +186,7 @@ class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, BinarySensorEntity):
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: BMWDataUpdateCoordinator,
|
||||
vehicle: ConnectedDriveVehicle,
|
||||
vehicle: MyBMWVehicle,
|
||||
description: BMWBinarySensorEntityDescription,
|
||||
unit_system: UnitSystem,
|
||||
) -> None:
|
||||
@ -217,14 +206,12 @@ class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, BinarySensorEntity):
|
||||
self.entity_description.key,
|
||||
self.vehicle.name,
|
||||
)
|
||||
vehicle_state = self.vehicle.status
|
||||
|
||||
self._attr_is_on = self.entity_description.value_fn(vehicle_state)
|
||||
self._attr_is_on = self.entity_description.value_fn(self.vehicle)
|
||||
|
||||
if self.entity_description.attr_fn:
|
||||
self._attr_extra_state_attributes = dict(
|
||||
self._attrs,
|
||||
**self.entity_description.attr_fn(vehicle_state, self._unit_system),
|
||||
**self.entity_description.attr_fn(self.vehicle, self._unit_system),
|
||||
)
|
||||
|
||||
super()._handle_coordinator_update()
|
||||
|
@ -1,19 +1,20 @@
|
||||
"""Support for BMW connected drive button entities."""
|
||||
"""Support for MyBMW button entities."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable, Coroutine
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from bimmer_connected.vehicle import ConnectedDriveVehicle
|
||||
from bimmer_connected.vehicle import MyBMWVehicle
|
||||
from bimmer_connected.vehicle.remote_services import RemoteServiceStatus
|
||||
|
||||
from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from . import BMWConnectedDriveBaseEntity
|
||||
from . import BMWBaseEntity
|
||||
from .const import DOMAIN
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@ -27,7 +28,9 @@ class BMWButtonEntityDescription(ButtonEntityDescription):
|
||||
"""Class describing BMW button entities."""
|
||||
|
||||
enabled_when_read_only: bool = False
|
||||
remote_function: str | None = None
|
||||
remote_function: Callable[
|
||||
[MyBMWVehicle], Coroutine[Any, Any, RemoteServiceStatus]
|
||||
] | None = None
|
||||
account_function: Callable[[BMWDataUpdateCoordinator], Coroutine] | None = None
|
||||
|
||||
|
||||
@ -36,31 +39,31 @@ BUTTON_TYPES: tuple[BMWButtonEntityDescription, ...] = (
|
||||
key="light_flash",
|
||||
icon="mdi:car-light-alert",
|
||||
name="Flash Lights",
|
||||
remote_function="trigger_remote_light_flash",
|
||||
remote_function=lambda vehicle: vehicle.remote_services.trigger_remote_light_flash(),
|
||||
),
|
||||
BMWButtonEntityDescription(
|
||||
key="sound_horn",
|
||||
icon="mdi:bullhorn",
|
||||
name="Sound Horn",
|
||||
remote_function="trigger_remote_horn",
|
||||
remote_function=lambda vehicle: vehicle.remote_services.trigger_remote_horn(),
|
||||
),
|
||||
BMWButtonEntityDescription(
|
||||
key="activate_air_conditioning",
|
||||
icon="mdi:hvac",
|
||||
name="Activate Air Conditioning",
|
||||
remote_function="trigger_remote_air_conditioning",
|
||||
remote_function=lambda vehicle: vehicle.remote_services.trigger_remote_air_conditioning(),
|
||||
),
|
||||
BMWButtonEntityDescription(
|
||||
key="deactivate_air_conditioning",
|
||||
icon="mdi:hvac-off",
|
||||
name="Deactivate Air Conditioning",
|
||||
remote_function="trigger_remote_air_conditioning_stop",
|
||||
remote_function=lambda vehicle: vehicle.remote_services.trigger_remote_air_conditioning_stop(),
|
||||
),
|
||||
BMWButtonEntityDescription(
|
||||
key="find_vehicle",
|
||||
icon="mdi:crosshairs-question",
|
||||
name="Find Vehicle",
|
||||
remote_function="trigger_remote_vehicle_finder",
|
||||
remote_function=lambda vehicle: vehicle.remote_services.trigger_remote_vehicle_finder(),
|
||||
),
|
||||
BMWButtonEntityDescription(
|
||||
key="refresh",
|
||||
@ -77,7 +80,7 @@ async def async_setup_entry(
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the BMW ConnectedDrive buttons from config entry."""
|
||||
"""Set up the BMW buttons from config entry."""
|
||||
coordinator: BMWDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
|
||||
|
||||
entities: list[BMWButton] = []
|
||||
@ -95,15 +98,15 @@ async def async_setup_entry(
|
||||
async_add_entities(entities)
|
||||
|
||||
|
||||
class BMWButton(BMWConnectedDriveBaseEntity, ButtonEntity):
|
||||
"""Representation of a BMW Connected Drive button."""
|
||||
class BMWButton(BMWBaseEntity, ButtonEntity):
|
||||
"""Representation of a MyBMW button."""
|
||||
|
||||
entity_description: BMWButtonEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: BMWDataUpdateCoordinator,
|
||||
vehicle: ConnectedDriveVehicle,
|
||||
vehicle: MyBMWVehicle,
|
||||
description: BMWButtonEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize BMW vehicle sensor."""
|
||||
@ -116,12 +119,7 @@ class BMWButton(BMWConnectedDriveBaseEntity, ButtonEntity):
|
||||
async def async_press(self) -> None:
|
||||
"""Press the button."""
|
||||
if self.entity_description.remote_function:
|
||||
await self.hass.async_add_executor_job(
|
||||
getattr(
|
||||
self.vehicle.remote_services,
|
||||
self.entity_description.remote_function,
|
||||
)
|
||||
)
|
||||
await self.entity_description.remote_function(self.vehicle)
|
||||
elif self.entity_description.account_function:
|
||||
_LOGGER.warning(
|
||||
"The 'Refresh from cloud' button is deprecated. Use the 'homeassistant.update_entity' "
|
||||
|
@ -3,8 +3,9 @@ from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from bimmer_connected.account import ConnectedDriveAccount
|
||||
from bimmer_connected.country_selector import get_region_from_name
|
||||
from bimmer_connected.account import MyBMWAccount
|
||||
from bimmer_connected.api.regions import get_region_from_name
|
||||
from httpx import HTTPError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries, core, exceptions
|
||||
@ -31,22 +32,23 @@ async def validate_input(
|
||||
|
||||
Data has the keys from DATA_SCHEMA with values provided by the user.
|
||||
"""
|
||||
account = MyBMWAccount(
|
||||
data[CONF_USERNAME],
|
||||
data[CONF_PASSWORD],
|
||||
get_region_from_name(data[CONF_REGION]),
|
||||
)
|
||||
|
||||
try:
|
||||
await hass.async_add_executor_job(
|
||||
ConnectedDriveAccount,
|
||||
data[CONF_USERNAME],
|
||||
data[CONF_PASSWORD],
|
||||
get_region_from_name(data[CONF_REGION]),
|
||||
)
|
||||
except OSError as ex:
|
||||
await account.get_vehicles()
|
||||
except HTTPError as ex:
|
||||
raise CannotConnect from ex
|
||||
|
||||
# Return info that you want to store in the config entry.
|
||||
return {"title": f"{data[CONF_USERNAME]}{data.get(CONF_SOURCE, '')}"}
|
||||
|
||||
|
||||
class BMWConnectedDriveConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for BMW ConnectedDrive."""
|
||||
class BMWConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for MyBMW."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
@ -78,16 +80,16 @@ class BMWConnectedDriveConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
@callback
|
||||
def async_get_options_flow(
|
||||
config_entry: config_entries.ConfigEntry,
|
||||
) -> BMWConnectedDriveOptionsFlow:
|
||||
"""Return a BWM ConnectedDrive option flow."""
|
||||
return BMWConnectedDriveOptionsFlow(config_entry)
|
||||
) -> BMWOptionsFlow:
|
||||
"""Return a MyBMW option flow."""
|
||||
return BMWOptionsFlow(config_entry)
|
||||
|
||||
|
||||
class BMWConnectedDriveOptionsFlow(config_entries.OptionsFlow):
|
||||
"""Handle a option flow for BMW ConnectedDrive."""
|
||||
class BMWOptionsFlow(config_entries.OptionsFlow):
|
||||
"""Handle a option flow for MyBMW."""
|
||||
|
||||
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
|
||||
"""Initialize BMW ConnectedDrive option flow."""
|
||||
"""Initialize MyBMW option flow."""
|
||||
self.config_entry = config_entry
|
||||
self.options = dict(config_entry.options)
|
||||
|
||||
@ -102,6 +104,16 @@ class BMWConnectedDriveOptionsFlow(config_entries.OptionsFlow):
|
||||
) -> FlowResult:
|
||||
"""Handle the initial step."""
|
||||
if user_input is not None:
|
||||
# Manually update & reload the config entry after options change.
|
||||
# Required as each successful login will store the latest refresh_token
|
||||
# using async_update_entry, which would otherwise trigger a full reload
|
||||
# if the options would be refreshed using a listener.
|
||||
changed = self.hass.config_entries.async_update_entry(
|
||||
self.config_entry,
|
||||
options=user_input,
|
||||
)
|
||||
if changed:
|
||||
await self.hass.config_entries.async_reload(self.config_entry.entry_id)
|
||||
return self.async_create_entry(title="", data=user_input)
|
||||
return self.async_show_form(
|
||||
step_id="account_options",
|
||||
|
@ -1,4 +1,4 @@
|
||||
"""Const file for the BMW Connected Drive integration."""
|
||||
"""Const file for the MyBMW integration."""
|
||||
from homeassistant.const import (
|
||||
LENGTH_KILOMETERS,
|
||||
LENGTH_MILES,
|
||||
@ -7,7 +7,7 @@ from homeassistant.const import (
|
||||
)
|
||||
|
||||
DOMAIN = "bmw_connected_drive"
|
||||
ATTRIBUTION = "Data provided by BMW Connected Drive"
|
||||
ATTRIBUTION = "Data provided by MyBMW"
|
||||
|
||||
ATTR_DIRECTION = "direction"
|
||||
ATTR_VIN = "vin"
|
||||
@ -15,6 +15,7 @@ ATTR_VIN = "vin"
|
||||
CONF_ALLOWED_REGIONS = ["china", "north_america", "rest_of_world"]
|
||||
CONF_READ_ONLY = "read_only"
|
||||
CONF_ACCOUNT = "account"
|
||||
CONF_REFRESH_TOKEN = "refresh_token"
|
||||
|
||||
DATA_HASS_CONFIG = "hass_config"
|
||||
|
||||
|
@ -4,14 +4,17 @@ 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 bimmer_connected.account import MyBMWAccount
|
||||
from bimmer_connected.api.regions import get_region_from_name
|
||||
from bimmer_connected.vehicle.models import GPSPosition
|
||||
from httpx import HTTPError, TimeoutException
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_REGION, CONF_USERNAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import DOMAIN
|
||||
from .const import CONF_READ_ONLY, CONF_REFRESH_TOKEN, DOMAIN
|
||||
|
||||
SCAN_INTERVAL = timedelta(seconds=300)
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
@ -20,53 +23,56 @@ _LOGGER = logging.getLogger(__name__)
|
||||
class BMWDataUpdateCoordinator(DataUpdateCoordinator):
|
||||
"""Class to manage fetching BMW data."""
|
||||
|
||||
account: ConnectedDriveAccount
|
||||
account: MyBMWAccount
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
*,
|
||||
username: str,
|
||||
password: str,
|
||||
region: str,
|
||||
read_only: bool = False,
|
||||
) -> None:
|
||||
def __init__(self, hass: HomeAssistant, *, entry: ConfigEntry) -> 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 = MyBMWAccount(
|
||||
entry.data[CONF_USERNAME],
|
||||
entry.data[CONF_PASSWORD],
|
||||
get_region_from_name(entry.data[CONF_REGION]),
|
||||
observer_position=GPSPosition(hass.config.latitude, hass.config.longitude),
|
||||
)
|
||||
self.read_only = entry.options[CONF_READ_ONLY]
|
||||
self._entry = entry
|
||||
|
||||
self.account = None
|
||||
self.read_only = read_only
|
||||
if CONF_REFRESH_TOKEN in entry.data:
|
||||
self.account.set_refresh_token(entry.data[CONF_REFRESH_TOKEN])
|
||||
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=f"{DOMAIN}-{username}",
|
||||
name=f"{DOMAIN}-{entry.data['username']}",
|
||||
update_interval=SCAN_INTERVAL,
|
||||
)
|
||||
|
||||
async def _async_update_data(self) -> None:
|
||||
"""Fetch data from BMW."""
|
||||
old_refresh_token = self.account.refresh_token
|
||||
|
||||
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
|
||||
await self.account.get_vehicles()
|
||||
except (HTTPError, TimeoutException) as err:
|
||||
self._update_config_entry_refresh_token(None)
|
||||
raise UpdateFailed(f"Error communicating with BMW API: {err}") from err
|
||||
|
||||
if self.account.refresh_token != old_refresh_token:
|
||||
self._update_config_entry_refresh_token(self.account.refresh_token)
|
||||
_LOGGER.debug(
|
||||
"bimmer_connected: refresh token %s > %s",
|
||||
old_refresh_token,
|
||||
self.account.refresh_token,
|
||||
)
|
||||
|
||||
def _update_config_entry_refresh_token(self, refresh_token: str | None) -> None:
|
||||
"""Update or delete the refresh_token in the Config Entry."""
|
||||
data = {
|
||||
**self._entry.data,
|
||||
CONF_REFRESH_TOKEN: refresh_token,
|
||||
}
|
||||
if not refresh_token:
|
||||
data.pop(CONF_REFRESH_TOKEN)
|
||||
self.hass.config_entries.async_update_entry(self._entry, data=data)
|
||||
|
||||
def notify_listeners(self) -> None:
|
||||
"""Notify all listeners to refresh HA state machine."""
|
||||
|
@ -1,10 +1,10 @@
|
||||
"""Device tracker for BMW Connected Drive vehicles."""
|
||||
"""Device tracker for MyBMW vehicles."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Literal
|
||||
|
||||
from bimmer_connected.vehicle import ConnectedDriveVehicle
|
||||
from bimmer_connected.vehicle import MyBMWVehicle
|
||||
|
||||
from homeassistant.components.device_tracker import SOURCE_TYPE_GPS
|
||||
from homeassistant.components.device_tracker.config_entry import TrackerEntity
|
||||
@ -12,7 +12,7 @@ from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from . import BMWConnectedDriveBaseEntity
|
||||
from . import BMWBaseEntity
|
||||
from .const import ATTR_DIRECTION, DOMAIN
|
||||
from .coordinator import BMWDataUpdateCoordinator
|
||||
|
||||
@ -24,7 +24,7 @@ async def async_setup_entry(
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the BMW ConnectedDrive tracker from config entry."""
|
||||
"""Set up the MyBMW tracker from config entry."""
|
||||
coordinator: BMWDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
|
||||
entities: list[BMWDeviceTracker] = []
|
||||
|
||||
@ -39,8 +39,8 @@ async def async_setup_entry(
|
||||
async_add_entities(entities)
|
||||
|
||||
|
||||
class BMWDeviceTracker(BMWConnectedDriveBaseEntity, TrackerEntity):
|
||||
"""BMW Connected Drive device tracker."""
|
||||
class BMWDeviceTracker(BMWBaseEntity, TrackerEntity):
|
||||
"""MyBMW device tracker."""
|
||||
|
||||
_attr_force_update = False
|
||||
_attr_icon = "mdi:car"
|
||||
@ -48,7 +48,7 @@ class BMWDeviceTracker(BMWConnectedDriveBaseEntity, TrackerEntity):
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: BMWDataUpdateCoordinator,
|
||||
vehicle: ConnectedDriveVehicle,
|
||||
vehicle: MyBMWVehicle,
|
||||
) -> None:
|
||||
"""Initialize the Tracker."""
|
||||
super().__init__(coordinator, vehicle)
|
||||
@ -59,13 +59,13 @@ class BMWDeviceTracker(BMWConnectedDriveBaseEntity, TrackerEntity):
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict:
|
||||
"""Return entity specific state attributes."""
|
||||
return dict(self._attrs, **{ATTR_DIRECTION: self.vehicle.status.gps_heading})
|
||||
return {**self._attrs, ATTR_DIRECTION: self.vehicle.vehicle_location.heading}
|
||||
|
||||
@property
|
||||
def latitude(self) -> float | None:
|
||||
"""Return latitude value of the device."""
|
||||
return (
|
||||
self.vehicle.status.gps_position[0]
|
||||
self.vehicle.vehicle_location.location[0]
|
||||
if self.vehicle.is_vehicle_tracking_enabled
|
||||
else None
|
||||
)
|
||||
@ -74,7 +74,7 @@ class BMWDeviceTracker(BMWConnectedDriveBaseEntity, TrackerEntity):
|
||||
def longitude(self) -> float | None:
|
||||
"""Return longitude value of the device."""
|
||||
return (
|
||||
self.vehicle.status.gps_position[1]
|
||||
self.vehicle.vehicle_location.location[1]
|
||||
if self.vehicle.is_vehicle_tracking_enabled
|
||||
else None
|
||||
)
|
||||
|
@ -4,15 +4,15 @@ from __future__ import annotations
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from bimmer_connected.vehicle import ConnectedDriveVehicle
|
||||
from bimmer_connected.vehicle_status import LockState
|
||||
from bimmer_connected.vehicle import MyBMWVehicle
|
||||
from bimmer_connected.vehicle.doors_windows import LockState
|
||||
|
||||
from homeassistant.components.lock import LockEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from . import BMWConnectedDriveBaseEntity
|
||||
from . import BMWBaseEntity
|
||||
from .const import DOMAIN
|
||||
from .coordinator import BMWDataUpdateCoordinator
|
||||
|
||||
@ -25,7 +25,7 @@ async def async_setup_entry(
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the BMW ConnectedDrive binary sensors from config entry."""
|
||||
"""Set up the MyBMW lock from config entry."""
|
||||
coordinator: BMWDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
|
||||
|
||||
entities: list[BMWLock] = []
|
||||
@ -36,13 +36,13 @@ async def async_setup_entry(
|
||||
async_add_entities(entities)
|
||||
|
||||
|
||||
class BMWLock(BMWConnectedDriveBaseEntity, LockEntity):
|
||||
"""Representation of a BMW vehicle lock."""
|
||||
class BMWLock(BMWBaseEntity, LockEntity):
|
||||
"""Representation of a MyBMW vehicle lock."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: BMWDataUpdateCoordinator,
|
||||
vehicle: ConnectedDriveVehicle,
|
||||
vehicle: MyBMWVehicle,
|
||||
attribute: str,
|
||||
sensor_name: str,
|
||||
) -> None:
|
||||
@ -55,7 +55,7 @@ class BMWLock(BMWConnectedDriveBaseEntity, LockEntity):
|
||||
self._sensor_name = sensor_name
|
||||
self.door_lock_state_available = DOOR_LOCK_STATE in vehicle.available_attributes
|
||||
|
||||
def lock(self, **kwargs: Any) -> None:
|
||||
async def async_lock(self, **kwargs: Any) -> None:
|
||||
"""Lock the car."""
|
||||
_LOGGER.debug("%s: locking doors", self.vehicle.name)
|
||||
# Only update the HA state machine if the vehicle reliably reports its lock state
|
||||
@ -63,10 +63,10 @@ class BMWLock(BMWConnectedDriveBaseEntity, LockEntity):
|
||||
# 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()
|
||||
self.async_write_ha_state()
|
||||
await self.vehicle.remote_services.trigger_remote_door_lock()
|
||||
|
||||
def unlock(self, **kwargs: Any) -> None:
|
||||
async def async_unlock(self, **kwargs: Any) -> None:
|
||||
"""Unlock the car."""
|
||||
_LOGGER.debug("%s: unlocking doors", self.vehicle.name)
|
||||
# Only update the HA state machine if the vehicle reliably reports its lock state
|
||||
@ -74,8 +74,8 @@ class BMWLock(BMWConnectedDriveBaseEntity, LockEntity):
|
||||
# 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()
|
||||
self.async_write_ha_state()
|
||||
await self.vehicle.remote_services.trigger_remote_door_unlock()
|
||||
|
||||
@callback
|
||||
def _handle_coordinator_update(self) -> None:
|
||||
@ -83,16 +83,14 @@ class BMWLock(BMWConnectedDriveBaseEntity, LockEntity):
|
||||
_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 {
|
||||
self._attr_is_locked = self.vehicle.doors_and_windows.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,
|
||||
"door_lock_state": self.vehicle.doors_and_windows.door_lock_state.value,
|
||||
},
|
||||
)
|
||||
|
||||
|
@ -2,7 +2,7 @@
|
||||
"domain": "bmw_connected_drive",
|
||||
"name": "BMW Connected Drive",
|
||||
"documentation": "https://www.home-assistant.io/integrations/bmw_connected_drive",
|
||||
"requirements": ["bimmer_connected==0.8.12"],
|
||||
"requirements": ["bimmer_connected==0.9.0"],
|
||||
"codeowners": ["@gerard33", "@rikroe"],
|
||||
"config_flow": true,
|
||||
"iot_class": "cloud_polling",
|
||||
|
@ -4,7 +4,7 @@ from __future__ import annotations
|
||||
import logging
|
||||
from typing import Any, cast
|
||||
|
||||
from bimmer_connected.vehicle import ConnectedDriveVehicle
|
||||
from bimmer_connected.vehicle import MyBMWVehicle
|
||||
|
||||
from homeassistant.components.notify import (
|
||||
ATTR_DATA,
|
||||
@ -52,14 +52,14 @@ def get_service(
|
||||
class BMWNotificationService(BaseNotificationService):
|
||||
"""Send Notifications to BMW."""
|
||||
|
||||
def __init__(self, targets: dict[str, ConnectedDriveVehicle]) -> None:
|
||||
def __init__(self, targets: dict[str, MyBMWVehicle]) -> None:
|
||||
"""Set up the notification service."""
|
||||
self.targets: dict[str, ConnectedDriveVehicle] = targets
|
||||
self.targets: dict[str, MyBMWVehicle] = targets
|
||||
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message or POI to the car."""
|
||||
for vehicle in kwargs[ATTR_TARGET]:
|
||||
vehicle = cast(ConnectedDriveVehicle, vehicle)
|
||||
vehicle = cast(MyBMWVehicle, vehicle)
|
||||
_LOGGER.debug("Sending message to %s", vehicle.name)
|
||||
|
||||
# Extract params from data dict
|
||||
|
@ -1,4 +1,4 @@
|
||||
"""Support for reading vehicle status from BMW connected drive portal."""
|
||||
"""Support for reading vehicle status from MyBMW portal."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
@ -6,7 +6,8 @@ from dataclasses import dataclass
|
||||
import logging
|
||||
from typing import cast
|
||||
|
||||
from bimmer_connected.vehicle import ConnectedDriveVehicle
|
||||
from bimmer_connected.vehicle import MyBMWVehicle
|
||||
from bimmer_connected.vehicle.models import ValueWithUnit
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
SensorDeviceClass,
|
||||
@ -27,7 +28,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import StateType
|
||||
from homeassistant.util.unit_system import UnitSystem
|
||||
|
||||
from . import BMWConnectedDriveBaseEntity
|
||||
from . import BMWBaseEntity
|
||||
from .const import DOMAIN, UNIT_MAP
|
||||
from .coordinator import BMWDataUpdateCoordinator
|
||||
|
||||
@ -38,44 +39,54 @@ _LOGGER = logging.getLogger(__name__)
|
||||
class BMWSensorEntityDescription(SensorEntityDescription):
|
||||
"""Describes BMW sensor entity."""
|
||||
|
||||
key_class: str | None = None
|
||||
unit_metric: str | None = None
|
||||
unit_imperial: str | None = None
|
||||
value: Callable = lambda x, y: x
|
||||
|
||||
|
||||
def convert_and_round(
|
||||
state: tuple,
|
||||
state: ValueWithUnit,
|
||||
converter: Callable[[float | None, str], float],
|
||||
precision: int,
|
||||
) -> float | None:
|
||||
"""Safely convert and round a value from a Tuple[value, unit]."""
|
||||
if state[0] is None:
|
||||
return None
|
||||
return round(converter(state[0], UNIT_MAP.get(state[1], state[1])), precision)
|
||||
"""Safely convert and round a value from ValueWithUnit."""
|
||||
if state.value and state.unit:
|
||||
return round(
|
||||
converter(state.value, UNIT_MAP.get(state.unit, state.unit)), precision
|
||||
)
|
||||
if state.value:
|
||||
return state.value
|
||||
return None
|
||||
|
||||
|
||||
SENSOR_TYPES: dict[str, BMWSensorEntityDescription] = {
|
||||
# --- Generic ---
|
||||
"charging_start_time": BMWSensorEntityDescription(
|
||||
key="charging_start_time",
|
||||
key_class="fuel_and_battery",
|
||||
device_class=SensorDeviceClass.TIMESTAMP,
|
||||
entity_registry_enabled_default=False,
|
||||
),
|
||||
"charging_end_time": BMWSensorEntityDescription(
|
||||
key="charging_end_time",
|
||||
key_class="fuel_and_battery",
|
||||
device_class=SensorDeviceClass.TIMESTAMP,
|
||||
),
|
||||
"charging_time_label": BMWSensorEntityDescription(
|
||||
key="charging_time_label",
|
||||
key_class="fuel_and_battery",
|
||||
entity_registry_enabled_default=False,
|
||||
),
|
||||
"charging_status": BMWSensorEntityDescription(
|
||||
key="charging_status",
|
||||
key_class="fuel_and_battery",
|
||||
icon="mdi:ev-station",
|
||||
value=lambda x, y: x.value,
|
||||
),
|
||||
"charging_level_hv": BMWSensorEntityDescription(
|
||||
key="charging_level_hv",
|
||||
"remaining_battery_percent": BMWSensorEntityDescription(
|
||||
key="remaining_battery_percent",
|
||||
key_class="fuel_and_battery",
|
||||
unit_metric=PERCENTAGE,
|
||||
unit_imperial=PERCENTAGE,
|
||||
device_class=SensorDeviceClass.BATTERY,
|
||||
@ -90,6 +101,7 @@ SENSOR_TYPES: dict[str, BMWSensorEntityDescription] = {
|
||||
),
|
||||
"remaining_range_total": BMWSensorEntityDescription(
|
||||
key="remaining_range_total",
|
||||
key_class="fuel_and_battery",
|
||||
icon="mdi:map-marker-distance",
|
||||
unit_metric=LENGTH_KILOMETERS,
|
||||
unit_imperial=LENGTH_MILES,
|
||||
@ -97,6 +109,7 @@ SENSOR_TYPES: dict[str, BMWSensorEntityDescription] = {
|
||||
),
|
||||
"remaining_range_electric": BMWSensorEntityDescription(
|
||||
key="remaining_range_electric",
|
||||
key_class="fuel_and_battery",
|
||||
icon="mdi:map-marker-distance",
|
||||
unit_metric=LENGTH_KILOMETERS,
|
||||
unit_imperial=LENGTH_MILES,
|
||||
@ -104,6 +117,7 @@ SENSOR_TYPES: dict[str, BMWSensorEntityDescription] = {
|
||||
),
|
||||
"remaining_range_fuel": BMWSensorEntityDescription(
|
||||
key="remaining_range_fuel",
|
||||
key_class="fuel_and_battery",
|
||||
icon="mdi:map-marker-distance",
|
||||
unit_metric=LENGTH_KILOMETERS,
|
||||
unit_imperial=LENGTH_MILES,
|
||||
@ -111,13 +125,15 @@ SENSOR_TYPES: dict[str, BMWSensorEntityDescription] = {
|
||||
),
|
||||
"remaining_fuel": BMWSensorEntityDescription(
|
||||
key="remaining_fuel",
|
||||
key_class="fuel_and_battery",
|
||||
icon="mdi:gas-station",
|
||||
unit_metric=VOLUME_LITERS,
|
||||
unit_imperial=VOLUME_GALLONS,
|
||||
value=lambda x, hass: convert_and_round(x, hass.config.units.volume, 2),
|
||||
),
|
||||
"fuel_percent": BMWSensorEntityDescription(
|
||||
key="fuel_percent",
|
||||
"remaining_fuel_percent": BMWSensorEntityDescription(
|
||||
key="remaining_fuel_percent",
|
||||
key_class="fuel_and_battery",
|
||||
icon="mdi:gas-station",
|
||||
unit_metric=PERCENTAGE,
|
||||
unit_imperial=PERCENTAGE,
|
||||
@ -130,16 +146,16 @@ async def async_setup_entry(
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the BMW ConnectedDrive sensors from config entry."""
|
||||
"""Set up the MyBMW sensors from config entry."""
|
||||
unit_system = hass.config.units
|
||||
coordinator: BMWDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
|
||||
|
||||
entities: list[BMWConnectedDriveSensor] = []
|
||||
entities: list[BMWSensor] = []
|
||||
|
||||
for vehicle in coordinator.account.vehicles:
|
||||
entities.extend(
|
||||
[
|
||||
BMWConnectedDriveSensor(coordinator, vehicle, description, unit_system)
|
||||
BMWSensor(coordinator, vehicle, description, unit_system)
|
||||
for attribute_name in vehicle.available_attributes
|
||||
if (description := SENSOR_TYPES.get(attribute_name))
|
||||
]
|
||||
@ -148,7 +164,7 @@ async def async_setup_entry(
|
||||
async_add_entities(entities)
|
||||
|
||||
|
||||
class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, SensorEntity):
|
||||
class BMWSensor(BMWBaseEntity, SensorEntity):
|
||||
"""Representation of a BMW vehicle sensor."""
|
||||
|
||||
entity_description: BMWSensorEntityDescription
|
||||
@ -156,7 +172,7 @@ class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, SensorEntity):
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: BMWDataUpdateCoordinator,
|
||||
vehicle: ConnectedDriveVehicle,
|
||||
vehicle: MyBMWVehicle,
|
||||
description: BMWSensorEntityDescription,
|
||||
unit_system: UnitSystem,
|
||||
) -> None:
|
||||
@ -178,7 +194,13 @@ class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, SensorEntity):
|
||||
_LOGGER.debug(
|
||||
"Updating sensor '%s' of %s", self.entity_description.key, self.vehicle.name
|
||||
)
|
||||
state = getattr(self.vehicle.status, self.entity_description.key)
|
||||
if self.entity_description.key_class is None:
|
||||
state = getattr(self.vehicle, self.entity_description.key)
|
||||
else:
|
||||
state = getattr(
|
||||
getattr(self.vehicle, self.entity_description.key_class),
|
||||
self.entity_description.key,
|
||||
)
|
||||
self._attr_native_value = cast(
|
||||
StateType, self.entity_description.value(state, self.hass)
|
||||
)
|
||||
|
@ -397,7 +397,7 @@ beautifulsoup4==4.11.1
|
||||
bellows==0.30.0
|
||||
|
||||
# homeassistant.components.bmw_connected_drive
|
||||
bimmer_connected==0.8.12
|
||||
bimmer_connected==0.9.0
|
||||
|
||||
# homeassistant.components.bizkaibus
|
||||
bizkaibus==0.1.1
|
||||
|
@ -312,7 +312,7 @@ beautifulsoup4==4.11.1
|
||||
bellows==0.30.0
|
||||
|
||||
# homeassistant.components.bmw_connected_drive
|
||||
bimmer_connected==0.8.12
|
||||
bimmer_connected==0.9.0
|
||||
|
||||
# homeassistant.components.blebox
|
||||
blebox_uniapi==1.3.3
|
||||
|
@ -1 +1,28 @@
|
||||
"""Tests for the for the BMW Connected Drive integration."""
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.bmw_connected_drive.const import (
|
||||
CONF_READ_ONLY,
|
||||
DOMAIN as BMW_DOMAIN,
|
||||
)
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_REGION, CONF_USERNAME
|
||||
|
||||
FIXTURE_USER_INPUT = {
|
||||
CONF_USERNAME: "user@domain.com",
|
||||
CONF_PASSWORD: "p4ssw0rd",
|
||||
CONF_REGION: "rest_of_world",
|
||||
}
|
||||
|
||||
FIXTURE_CONFIG_ENTRY = {
|
||||
"entry_id": "1",
|
||||
"domain": BMW_DOMAIN,
|
||||
"title": FIXTURE_USER_INPUT[CONF_USERNAME],
|
||||
"data": {
|
||||
CONF_USERNAME: FIXTURE_USER_INPUT[CONF_USERNAME],
|
||||
CONF_PASSWORD: FIXTURE_USER_INPUT[CONF_PASSWORD],
|
||||
CONF_REGION: FIXTURE_USER_INPUT[CONF_REGION],
|
||||
},
|
||||
"options": {CONF_READ_ONLY: False},
|
||||
"source": config_entries.SOURCE_USER,
|
||||
"unique_id": f"{FIXTURE_USER_INPUT[CONF_REGION]}-{FIXTURE_USER_INPUT[CONF_REGION]}",
|
||||
}
|
||||
|
@ -1,35 +1,20 @@
|
||||
"""Test the for the BMW Connected Drive config flow."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from httpx import HTTPError
|
||||
|
||||
from homeassistant import config_entries, data_entry_flow
|
||||
from homeassistant.components.bmw_connected_drive.config_flow import DOMAIN
|
||||
from homeassistant.components.bmw_connected_drive.const import CONF_READ_ONLY
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_REGION, CONF_USERNAME
|
||||
from homeassistant.const import CONF_USERNAME
|
||||
|
||||
from . import FIXTURE_CONFIG_ENTRY, FIXTURE_USER_INPUT
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
FIXTURE_USER_INPUT = {
|
||||
CONF_USERNAME: "user@domain.com",
|
||||
CONF_PASSWORD: "p4ssw0rd",
|
||||
CONF_REGION: "rest_of_world",
|
||||
}
|
||||
FIXTURE_COMPLETE_ENTRY = FIXTURE_USER_INPUT.copy()
|
||||
FIXTURE_IMPORT_ENTRY = FIXTURE_USER_INPUT.copy()
|
||||
|
||||
FIXTURE_CONFIG_ENTRY = {
|
||||
"entry_id": "1",
|
||||
"domain": DOMAIN,
|
||||
"title": FIXTURE_USER_INPUT[CONF_USERNAME],
|
||||
"data": {
|
||||
CONF_USERNAME: FIXTURE_USER_INPUT[CONF_USERNAME],
|
||||
CONF_PASSWORD: FIXTURE_USER_INPUT[CONF_PASSWORD],
|
||||
CONF_REGION: FIXTURE_USER_INPUT[CONF_REGION],
|
||||
},
|
||||
"options": {CONF_READ_ONLY: False},
|
||||
"source": config_entries.SOURCE_USER,
|
||||
"unique_id": f"{FIXTURE_USER_INPUT[CONF_REGION]}-{FIXTURE_USER_INPUT[CONF_REGION]}",
|
||||
}
|
||||
|
||||
|
||||
async def test_show_form(hass):
|
||||
"""Test that the form is served with no input."""
|
||||
@ -48,8 +33,8 @@ async def test_connection_error(hass):
|
||||
pass
|
||||
|
||||
with patch(
|
||||
"bimmer_connected.account.ConnectedDriveAccount._get_oauth_token",
|
||||
side_effect=OSError,
|
||||
"bimmer_connected.api.authentication.MyBMWAuthentication.login",
|
||||
side_effect=HTTPError("login failure"),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
@ -65,7 +50,7 @@ async def test_connection_error(hass):
|
||||
async def test_full_user_flow_implementation(hass):
|
||||
"""Test registering an integration and finishing flow works."""
|
||||
with patch(
|
||||
"bimmer_connected.account.ConnectedDriveAccount._get_vehicles",
|
||||
"bimmer_connected.account.MyBMWAccount.get_vehicles",
|
||||
return_value=[],
|
||||
), patch(
|
||||
"homeassistant.components.bmw_connected_drive.async_setup_entry",
|
||||
@ -86,7 +71,7 @@ async def test_full_user_flow_implementation(hass):
|
||||
async def test_options_flow_implementation(hass):
|
||||
"""Test config flow options."""
|
||||
with patch(
|
||||
"bimmer_connected.account.ConnectedDriveAccount._get_vehicles",
|
||||
"bimmer_connected.account.MyBMWAccount.get_vehicles",
|
||||
return_value=[],
|
||||
), patch(
|
||||
"homeassistant.components.bmw_connected_drive.async_setup_entry",
|
||||
@ -104,13 +89,13 @@ async def test_options_flow_implementation(hass):
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_READ_ONLY: False},
|
||||
user_input={CONF_READ_ONLY: True},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"] == {
|
||||
CONF_READ_ONLY: False,
|
||||
CONF_READ_ONLY: True,
|
||||
}
|
||||
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
136
tests/components/bmw_connected_drive/test_init.py
Normal file
136
tests/components/bmw_connected_drive/test_init.py
Normal file
@ -0,0 +1,136 @@
|
||||
"""Test Axis component setup process."""
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.bmw_connected_drive.const import DOMAIN as BMW_DOMAIN
|
||||
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from . import FIXTURE_CONFIG_ENTRY
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
VIN = "WBYYYYYYYYYYYYYYY"
|
||||
VEHICLE_NAME = "i3 (+ REX)"
|
||||
VEHICLE_NAME_SLUG = "i3_rex"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"entitydata,old_unique_id,new_unique_id",
|
||||
[
|
||||
(
|
||||
{
|
||||
"domain": SENSOR_DOMAIN,
|
||||
"platform": BMW_DOMAIN,
|
||||
"unique_id": f"{VIN}-charging_level_hv",
|
||||
"suggested_object_id": f"{VEHICLE_NAME} charging_level_hv",
|
||||
"disabled_by": None,
|
||||
},
|
||||
f"{VIN}-charging_level_hv",
|
||||
f"{VIN}-remaining_battery_percent",
|
||||
),
|
||||
(
|
||||
{
|
||||
"domain": SENSOR_DOMAIN,
|
||||
"platform": BMW_DOMAIN,
|
||||
"unique_id": f"{VIN}-remaining_range_total",
|
||||
"suggested_object_id": f"{VEHICLE_NAME} remaining_range_total",
|
||||
"disabled_by": None,
|
||||
},
|
||||
f"{VIN}-remaining_range_total",
|
||||
f"{VIN}-remaining_range_total",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_migrate_unique_ids(
|
||||
hass: HomeAssistant,
|
||||
entitydata: dict,
|
||||
old_unique_id: str,
|
||||
new_unique_id: str,
|
||||
) -> None:
|
||||
"""Test successful migration of entity unique_ids."""
|
||||
mock_config_entry = MockConfigEntry(**FIXTURE_CONFIG_ENTRY)
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
entity_registry = er.async_get(hass)
|
||||
entity: er.RegistryEntry = entity_registry.async_get_or_create(
|
||||
**entitydata,
|
||||
config_entry=mock_config_entry,
|
||||
)
|
||||
|
||||
assert entity.unique_id == old_unique_id
|
||||
|
||||
with patch(
|
||||
"bimmer_connected.account.MyBMWAccount.get_vehicles",
|
||||
return_value=[],
|
||||
):
|
||||
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entity_migrated = entity_registry.async_get(entity.entity_id)
|
||||
assert entity_migrated
|
||||
assert entity_migrated.unique_id == new_unique_id
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"entitydata,old_unique_id,new_unique_id",
|
||||
[
|
||||
(
|
||||
{
|
||||
"domain": SENSOR_DOMAIN,
|
||||
"platform": BMW_DOMAIN,
|
||||
"unique_id": f"{VIN}-charging_level_hv",
|
||||
"suggested_object_id": f"{VEHICLE_NAME} charging_level_hv",
|
||||
"disabled_by": None,
|
||||
},
|
||||
f"{VIN}-charging_level_hv",
|
||||
f"{VIN}-remaining_battery_percent",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_dont_migrate_unique_ids(
|
||||
hass: HomeAssistant,
|
||||
entitydata: dict,
|
||||
old_unique_id: str,
|
||||
new_unique_id: str,
|
||||
) -> None:
|
||||
"""Test successful migration of entity unique_ids."""
|
||||
mock_config_entry = MockConfigEntry(**FIXTURE_CONFIG_ENTRY)
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
entity_registry = er.async_get(hass)
|
||||
|
||||
# create existing entry with new_unique_id
|
||||
existing_entity = entity_registry.async_get_or_create(
|
||||
SENSOR_DOMAIN,
|
||||
BMW_DOMAIN,
|
||||
unique_id=f"{VIN}-remaining_battery_percent",
|
||||
suggested_object_id=f"{VEHICLE_NAME} remaining_battery_percent",
|
||||
config_entry=mock_config_entry,
|
||||
)
|
||||
|
||||
entity: er.RegistryEntry = entity_registry.async_get_or_create(
|
||||
**entitydata,
|
||||
config_entry=mock_config_entry,
|
||||
)
|
||||
|
||||
assert entity.unique_id == old_unique_id
|
||||
|
||||
with patch(
|
||||
"bimmer_connected.account.MyBMWAccount.get_vehicles",
|
||||
return_value=[],
|
||||
):
|
||||
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entity_migrated = entity_registry.async_get(entity.entity_id)
|
||||
assert entity_migrated
|
||||
assert entity_migrated.unique_id == old_unique_id
|
||||
|
||||
entity_not_changed = entity_registry.async_get(existing_entity.entity_id)
|
||||
assert entity_not_changed
|
||||
assert entity_not_changed.unique_id == new_unique_id
|
||||
|
||||
assert entity_migrated != entity_not_changed
|
Loading…
x
Reference in New Issue
Block a user