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

Add Config Flow to bmw_connected_drive (#39585)

Browse Source 
* Add Config Flow to bmw_connected_drive

* Fix checks for bmw_connected_drive

* Adjust code as requested

* Clean .coveragerc after merge

* Use references for config flow

* Fix execute_service check against allowed accounts

* Adjust translation as username can be email or phone no

* Add BMWConnectedDriveBaseEntity mixin, remove unnecessary type casts

* Use BaseEntity correctly, fix pylint error

* Bump bimmer_connected to 0.7.13

* Adjustments for review

* Fix pylint

* Fix loading notify, move vin to entity attrs

* Remove vin from device registry

* Remove commented-out code

* Show tracker warning only if vehicle (currently) doesn't support location

* Remove unnecessary return values & other small adjustments

* Move original hass_config to own domain in hass.data

* Move entries to separate dict in hass.data

* Remove invalid_auth exception handling & test as it cannot happen

Co-authored-by: rikroe <rikroe@users.noreply.github.com>
This commit is contained in:
rikroe 2020-12-29 11:06:12 +01:00 committed by GitHub
parent 5164a18d53
commit e5f31665b1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 736 additions and 217 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.coveragerc
View File

@ -100,7 +100,12 @@ omit =
homeassistant/components/bme280/sensor.py
homeassistant/components/bme680/sensor.py
homeassistant/components/bmp280/sensor.py
homeassistant/components/bmw_connected_drive/*
homeassistant/components/bmw_connected_drive/__init__.py
homeassistant/components/bmw_connected_drive/binary_sensor.py
homeassistant/components/bmw_connected_drive/device_tracker.py
homeassistant/components/bmw_connected_drive/lock.py
homeassistant/components/bmw_connected_drive/notify.py
homeassistant/components/bmw_connected_drive/sensor.py
homeassistant/components/braviatv/__init__.py
homeassistant/components/braviatv/const.py
homeassistant/components/braviatv/media_player.py

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

@ -1,29 +1,50 @@
"""Reads vehicle status from BMW connected drive portal."""
import asyncio
import logging
from bimmer_connected.account import ConnectedDriveAccount
from bimmer_connected.country_selector import get_region_from_name
import voluptuous as vol
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
ATTR_ATTRIBUTION,
CONF_NAME,
CONF_PASSWORD,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.event import track_utc_time_change
from homeassistant.util import slugify
import homeassistant.util.dt as dt_util
from .const import (
ATTRIBUTION,
CONF_ACCOUNT,
CONF_ALLOWED_REGIONS,
CONF_READ_ONLY,
CONF_REGION,
CONF_USE_LOCATION,
DATA_ENTRIES,
DATA_HASS_CONFIG,
)
_LOGGER = logging.getLogger(__name__)
DOMAIN = "bmw_connected_drive"
CONF_REGION = "region"
CONF_READ_ONLY = "read_only"
ATTR_VIN = "vin"
ACCOUNT_SCHEMA = vol.Schema(
{
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(CONF_REGION): vol.Any("north_america", "china", "rest_of_world"),
vol.Optional(CONF_READ_ONLY, default=False): cv.boolean,
vol.Required(CONF_REGION): vol.In(CONF_ALLOWED_REGIONS),
vol.Optional(CONF_READ_ONLY): cv.boolean,
}
)
@ -31,8 +52,12 @@ CONFIG_SCHEMA = vol.Schema({DOMAIN: {cv.string: ACCOUNT_SCHEMA}}, extra=vol.ALLO
SERVICE_SCHEMA = vol.Schema({vol.Required(ATTR_VIN): cv.string})
DEFAULT_OPTIONS = {
CONF_READ_ONLY: False,
CONF_USE_LOCATION: False,
}
BMW_COMPONENTS = ["binary_sensor", "device_tracker", "lock", "notify", "sensor"]
BMW_PLATFORMS = ["binary_sensor", "device_tracker", "lock", "notify", "sensor"]
UPDATE_INTERVAL = 5 # in minutes
SERVICE_UPDATE_STATE = "update_state"
@ -44,49 +69,162 @@ _SERVICE_MAP = {
"find_vehicle": "trigger_remote_vehicle_finder",
}
UNDO_UPDATE_LISTENER = "undo_update_listener"
def setup(hass, config: dict):
"""Set up the BMW connected drive components."""
accounts = []
for name, account_config in config[DOMAIN].items():
accounts.append(setup_account(account_config, hass, name))
hass.data[DOMAIN] = accounts
async def async_setup(hass: HomeAssistant, config: dict):
"""Set up the BMW Connected Drive component from configuration.yaml."""
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][DATA_HASS_CONFIG] = config
def _update_all(call) -> None:
"""Update all BMW accounts."""
for cd_account in hass.data[DOMAIN]:
cd_account.update()
# Service to manually trigger updates for all accounts.
hass.services.register(DOMAIN, SERVICE_UPDATE_STATE, _update_all)
_update_all(None)
for component in BMW_COMPONENTS:
discovery.load_platform(hass, component, DOMAIN, {}, config)
if DOMAIN in config:
for entry_config in config[DOMAIN].values():
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=entry_config
)
)
return True
def setup_account(account_config: dict, hass, name: str) -> "BMWConnectedDriveAccount":
@callback
def _async_migrate_options_from_data_if_missing(hass, entry):
data = dict(entry.data)
options = dict(entry.options)
if CONF_READ_ONLY in data or list(options) != list(DEFAULT_OPTIONS):
options = dict(DEFAULT_OPTIONS, **options)
options[CONF_READ_ONLY] = data.pop(CONF_READ_ONLY, False)
hass.config_entries.async_update_entry(entry, data=data, options=options)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""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
async def _async_update_all(service_call=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].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,
}
# Service to manually trigger updates for all accounts.
hass.services.async_register(DOMAIN, SERVICE_UPDATE_STATE, _async_update_all)
await _async_update_all()
for platform in BMW_PLATFORMS:
if platform != NOTIFY_DOMAIN:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, platform)
)
# set up notify platform, no entry support for notify component yet,
# have to use discovery to load platform.
hass.async_create_task(
discovery.async_load_platform(
hass,
NOTIFY_DOMAIN,
DOMAIN,
{CONF_NAME: DOMAIN},
hass.data[DOMAIN][DATA_HASS_CONFIG],
)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in BMW_PLATFORMS
if component != NOTIFY_DOMAIN
]
)
)
# Only remove services if it is the last account and not read only
if (
len(hass.data[DOMAIN][DATA_ENTRIES]) == 1
and not hass.data[DOMAIN][DATA_ENTRIES][entry.entry_id][CONF_ACCOUNT].read_only
):
services = list(_SERVICE_MAP) + [SERVICE_UPDATE_STATE]
for service in services:
hass.services.async_remove(DOMAIN, service)
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)
return unload_ok
async def update_listener(hass, config_entry):
"""Handle options update."""
await hass.config_entries.async_reload(config_entry.entry_id)
def setup_account(entry: ConfigEntry, hass, name: str) -> "BMWConnectedDriveAccount":
"""Set up a new BMWConnectedDriveAccount based on the config."""
username = account_config[CONF_USERNAME]
password = account_config[CONF_PASSWORD]
region = account_config[CONF_REGION]
read_only = account_config[CONF_READ_ONLY]
username = entry.data[CONF_USERNAME]
password = entry.data[CONF_PASSWORD]
region = entry.data[CONF_REGION]
read_only = entry.options[CONF_READ_ONLY]
use_location = entry.options[CONF_USE_LOCATION]
_LOGGER.debug("Adding new account %s", name)
cd_account = BMWConnectedDriveAccount(username, password, region, name, read_only)
pos = (
(hass.config.latitude, hass.config.longitude) if use_location else (None, None)
)
cd_account = BMWConnectedDriveAccount(
username, password, region, name, read_only, *pos
)
def execute_service(call):
"""Execute a service for a vehicle.
This must be a member function as we need access to the cd_account
object here.
"""
"""Execute a service for a vehicle."""
vin = call.data[ATTR_VIN]
vehicle = cd_account.account.get_vehicle(vin)
vehicle = None
# Double check for read_only accounts as another account could create the services
for entry_data in [
e
for e in hass.data[DOMAIN][DATA_ENTRIES].values()
if not e[CONF_ACCOUNT].read_only
]:
vehicle = entry_data[CONF_ACCOUNT].account.get_vehicle(vin)
if vehicle:
break
if not vehicle:
_LOGGER.error("Could not find a vehicle for VIN %s", vin)
return
@ -111,6 +249,9 @@ def setup_account(account_config: dict, hass, name: str) -> "BMWConnectedDriveAc
second=now.second,
)
# Initialize
cd_account.update()
return cd_account
@ -118,7 +259,14 @@ class BMWConnectedDriveAccount:
"""Representation of a BMW vehicle."""
def __init__(
self, username: str, password: str, region_str: str, name: str, read_only
self,
username: str,
password: str,
region_str: str,
name: str,
read_only: bool,
lat=None,
lon=None,
) -> None:
"""Initialize account."""
region = get_region_from_name(region_str)
@ -128,6 +276,12 @@ class BMWConnectedDriveAccount:
self.name = name
self._update_listeners = []
# 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, *_):
"""Update the state of all vehicles.
@ -152,3 +306,51 @@ class BMWConnectedDriveAccount:
def add_update_listener(self, listener):
"""Add a listener for update notifications."""
self._update_listeners.append(listener)
class BMWConnectedDriveBaseEntity(Entity):
"""Common base for BMW entities."""
def __init__(self, account, vehicle):
"""Initialize sensor."""
self._account = account
self._vehicle = vehicle
self._attrs = {
"car": self._vehicle.name,
"vin": self._vehicle.vin,
ATTR_ATTRIBUTION: ATTRIBUTION,
}
@property
def device_info(self) -> dict:
"""Return info for device registry."""
return {
"identifiers": {(DOMAIN, self._vehicle.vin)},
"name": f'{self._vehicle.attributes.get("brand")} {self._vehicle.name}',
"model": self._vehicle.name,
"manufacturer": self._vehicle.attributes.get("brand"),
}
@property
def device_state_attributes(self):
"""Return the state attributes of the sensor."""
return self._attrs
@property
def should_poll(self):
"""Do not poll this class.
Updates are triggered from BMWConnectedDriveAccount.
"""
return False
def update_callback(self):
"""Schedule a state update."""
self.schedule_update_ha_state(True)
async def async_added_to_hass(self):
"""Add callback after being added to hass.
Show latest data after startup.
"""
self._account.add_update_listener(self.update_callback)

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

@ -9,10 +9,10 @@ from homeassistant.components.binary_sensor import (
DEVICE_CLASS_PROBLEM,
BinarySensorEntity,
)
from homeassistant.const import ATTR_ATTRIBUTION, LENGTH_KILOMETERS
from homeassistant.const import LENGTH_KILOMETERS
from . import DOMAIN as BMW_DOMAIN
from .const import ATTRIBUTION
from . import DOMAIN as BMW_DOMAIN, BMWConnectedDriveBaseEntity
from .const import CONF_ACCOUNT, DATA_ENTRIES
_LOGGER = logging.getLogger(__name__)
@ -41,41 +41,40 @@ SENSOR_TYPES_ELEC = {
SENSOR_TYPES_ELEC.update(SENSOR_TYPES)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the BMW sensors."""
accounts = hass.data[BMW_DOMAIN]
_LOGGER.debug("Found BMW accounts: %s", ", ".join([a.name for a in accounts]))
devices = []
for account in accounts:
for vehicle in account.account.vehicles:
if vehicle.has_hv_battery:
_LOGGER.debug("BMW with a high voltage battery")
for key, value in sorted(SENSOR_TYPES_ELEC.items()):
if key in vehicle.available_attributes:
device = BMWConnectedDriveSensor(
account, vehicle, key, value[0], value[1], value[2]
)
devices.append(device)
elif vehicle.has_internal_combustion_engine:
_LOGGER.debug("BMW with an internal combustion engine")
for key, value in sorted(SENSOR_TYPES.items()):
if key in vehicle.available_attributes:
device = BMWConnectedDriveSensor(
account, vehicle, key, value[0], value[1], value[2]
)
devices.append(device)
add_entities(devices, True)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the BMW ConnectedDrive binary sensors from config entry."""
account = hass.data[BMW_DOMAIN][DATA_ENTRIES][config_entry.entry_id][CONF_ACCOUNT]
entities = []
for vehicle in account.account.vehicles:
if vehicle.has_hv_battery:
_LOGGER.debug("BMW with a high voltage battery")
for key, value in sorted(SENSOR_TYPES_ELEC.items()):
if key in vehicle.available_attributes:
device = BMWConnectedDriveSensor(
account, vehicle, key, value[0], value[1], value[2]
)
entities.append(device)
elif vehicle.has_internal_combustion_engine:
_LOGGER.debug("BMW with an internal combustion engine")
for key, value in sorted(SENSOR_TYPES.items()):
if key in vehicle.available_attributes:
device = BMWConnectedDriveSensor(
account, vehicle, key, value[0], value[1], value[2]
)
entities.append(device)
async_add_entities(entities, True)
class BMWConnectedDriveSensor(BinarySensorEntity):
class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, BinarySensorEntity):
"""Representation of a BMW vehicle binary sensor."""
def __init__(
self, account, vehicle, attribute: str, sensor_name, device_class, icon
):
"""Initialize sensor."""
self._account = account
self._vehicle = vehicle
super().__init__(account, vehicle)
self._attribute = attribute
self._name = f"{self._vehicle.name} {self._attribute}"
self._unique_id = f"{self._vehicle.vin}-{self._attribute}"
@ -84,14 +83,6 @@ class BMWConnectedDriveSensor(BinarySensorEntity):
self._icon = icon
self._state = None
@property
def should_poll(self) -> bool:
"""Return False.
Data update is triggered from BMWConnectedDriveEntity.
"""
return False
@property
def unique_id(self):
"""Return the unique ID of the binary sensor."""
@ -121,10 +112,7 @@ class BMWConnectedDriveSensor(BinarySensorEntity):
def device_state_attributes(self):
"""Return the state attributes of the binary sensor."""
vehicle_state = self._vehicle.state
result = {
"car": self._vehicle.name,
ATTR_ATTRIBUTION: ATTRIBUTION,
}
result = self._attrs.copy()
if self._attribute == "lids":
for lid in vehicle_state.lids:
@ -205,14 +193,3 @@ class BMWConnectedDriveSensor(BinarySensorEntity):
f"{service_type} distance"
] = f"{distance} {self.hass.config.units.length_unit}"
return result
def update_callback(self):
"""Schedule a state update."""
self.schedule_update_ha_state(True)
async def async_added_to_hass(self):
"""Add callback after being added to hass.
Show latest data after startup.
"""
self._account.add_update_listener(self.update_callback)

119
homeassistant/components/bmw_connected_drive/config_flow.py Normal file
View File

@ -0,0 +1,119 @@
"""Config flow for BMW ConnectedDrive integration."""
import logging
from bimmer_connected.account import ConnectedDriveAccount
from bimmer_connected.country_selector import get_region_from_name
import voluptuous as vol
from homeassistant import config_entries, core, exceptions
from homeassistant.const import CONF_PASSWORD, CONF_SOURCE, CONF_USERNAME
from homeassistant.core import callback
from . import DOMAIN # pylint: disable=unused-import
from .const import CONF_ALLOWED_REGIONS, CONF_READ_ONLY, CONF_REGION, CONF_USE_LOCATION
_LOGGER = logging.getLogger(__name__)
DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
vol.Required(CONF_REGION): vol.In(CONF_ALLOWED_REGIONS),
}
)
async def validate_input(hass: core.HomeAssistant, data):
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
"""
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:
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."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
async def async_step_user(self, user_input=None):
"""Handle the initial step."""
errors = {}
if user_input is not None:
unique_id = f"{user_input[CONF_REGION]}-{user_input[CONF_USERNAME]}"
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured()
info = None
try:
info = await validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
if info:
return self.async_create_entry(title=info["title"], data=user_input)
return self.async_show_form(
step_id="user", data_schema=DATA_SCHEMA, errors=errors
)
async def async_step_import(self, user_input):
"""Handle import."""
return await self.async_step_user(user_input)
@staticmethod
@callback
def async_get_options_flow(config_entry):
"""Return a BWM ConnectedDrive option flow."""
return BMWConnectedDriveOptionsFlow(config_entry)
class BMWConnectedDriveOptionsFlow(config_entries.OptionsFlow):
"""Handle a option flow for BMW ConnectedDrive."""
def __init__(self, config_entry):
"""Initialize BMW ConnectedDrive option flow."""
self.config_entry = config_entry
self.options = dict(config_entry.options)
async def async_step_init(self, user_input=None):
"""Manage the options."""
return await self.async_step_account_options()
async def async_step_account_options(self, user_input=None):
"""Handle the initial step."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
return self.async_show_form(
step_id="account_options",
data_schema=vol.Schema(
{
vol.Optional(
CONF_READ_ONLY,
default=self.config_entry.options.get(CONF_READ_ONLY, False),
): bool,
vol.Optional(
CONF_USE_LOCATION,
default=self.config_entry.options.get(CONF_USE_LOCATION, False),
): bool,
}
),
)
class CannotConnect(exceptions.HomeAssistantError):
"""Error to indicate we cannot connect."""

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

@ -1,2 +1,12 @@
"""Const file for the BMW Connected Drive integration."""
ATTRIBUTION = "Data provided by BMW Connected Drive"
CONF_REGION = "region"
CONF_ALLOWED_REGIONS = ["china", "north_america", "rest_of_world"]
CONF_READ_ONLY = "read_only"
CONF_USE_LOCATION = "use_location"
CONF_ACCOUNT = "account"
DATA_HASS_CONFIG = "hass_config"
DATA_ENTRIES = "entries"

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

@ -1,51 +1,83 @@
"""Device tracker for BMW Connected Drive vehicles."""
import logging
from homeassistant.util import slugify
from homeassistant.components.device_tracker import SOURCE_TYPE_GPS
from homeassistant.components.device_tracker.config_entry import TrackerEntity
from . import DOMAIN as BMW_DOMAIN
from . import DOMAIN as BMW_DOMAIN, BMWConnectedDriveBaseEntity
from .const import CONF_ACCOUNT, DATA_ENTRIES
_LOGGER = logging.getLogger(__name__)
def setup_scanner(hass, config, see, discovery_info=None):
"""Set up the BMW tracker."""
accounts = hass.data[BMW_DOMAIN]
_LOGGER.debug("Found BMW accounts: %s", ", ".join([a.name for a in accounts]))
for account in accounts:
for vehicle in account.account.vehicles:
tracker = BMWDeviceTracker(see, vehicle)
account.add_update_listener(tracker.update)
tracker.update()
return True
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the BMW ConnectedDrive tracker from config entry."""
account = hass.data[BMW_DOMAIN][DATA_ENTRIES][config_entry.entry_id][CONF_ACCOUNT]
entities = []
for vehicle in account.account.vehicles:
entities.append(BMWDeviceTracker(account, vehicle))
if not vehicle.state.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)
class BMWDeviceTracker:
class BMWDeviceTracker(BMWConnectedDriveBaseEntity, TrackerEntity):
"""BMW Connected Drive device tracker."""
def __init__(self, see, vehicle):
def __init__(self, account, vehicle):
"""Initialize the Tracker."""
self._see = see
self.vehicle = vehicle
super().__init__(account, vehicle)
def update(self) -> None:
"""Update the device info.
Only update the state in Home Assistant if tracking in
the car is enabled.
"""
dev_id = slugify(self.vehicle.name)
if not self.vehicle.state.is_vehicle_tracking_enabled:
_LOGGER.debug("Tracking is disabled for vehicle %s", dev_id)
return
_LOGGER.debug("Updating %s", dev_id)
attrs = {"vin": self.vehicle.vin}
self._see(
dev_id=dev_id,
host_name=self.vehicle.name,
gps=self.vehicle.state.gps_position,
attributes=attrs,
icon="mdi:car",
self._unique_id = vehicle.vin
self._location = (
vehicle.state.gps_position if vehicle.state.gps_position else (None, None)
)
self._name = vehicle.name
@property
def latitude(self):
"""Return latitude value of the device."""
return self._location[0]
@property
def longitude(self):
"""Return longitude value of the device."""
return self._location[1]
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def unique_id(self):
"""Return the unique ID."""
return self._unique_id
@property
def source_type(self):
"""Return the source type, eg gps or router, of the device."""
return SOURCE_TYPE_GPS
@property
def icon(self):
"""Return the icon to use in the frontend, if any."""
return "mdi:car"
@property
def force_update(self):
"""All updates do not need to be written to the state machine."""
return False
def update(self):
"""Update state of the decvice tracker."""
self._location = (
self._vehicle.state.gps_position
if self._vehicle.state.is_vehicle_tracking_enabled
else (None, None)
)

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

@ -4,35 +4,34 @@ import logging
from bimmer_connected.state import LockState
from homeassistant.components.lock import LockEntity
from homeassistant.const import ATTR_ATTRIBUTION, STATE_LOCKED, STATE_UNLOCKED
from homeassistant.const import STATE_LOCKED, STATE_UNLOCKED
from . import DOMAIN as BMW_DOMAIN
from .const import ATTRIBUTION
from . import DOMAIN as BMW_DOMAIN, BMWConnectedDriveBaseEntity
from .const import CONF_ACCOUNT, DATA_ENTRIES
DOOR_LOCK_STATE = "door_lock_state"
_LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the BMW Connected Drive lock."""
accounts = hass.data[BMW_DOMAIN]
_LOGGER.debug("Found BMW accounts: %s", ", ".join([a.name for a in accounts]))
devices = []
for account in accounts:
if not account.read_only:
for vehicle in account.account.vehicles:
device = BMWLock(account, vehicle, "lock", "BMW lock")
devices.append(device)
add_entities(devices, True)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the BMW ConnectedDrive binary sensors from config entry."""
account = hass.data[BMW_DOMAIN][DATA_ENTRIES][config_entry.entry_id][CONF_ACCOUNT]
entities = []
if not account.read_only:
for vehicle in account.account.vehicles:
device = BMWLock(account, vehicle, "lock", "BMW lock")
entities.append(device)
async_add_entities(entities, True)
class BMWLock(LockEntity):
class BMWLock(BMWConnectedDriveBaseEntity, LockEntity):
"""Representation of a BMW vehicle lock."""
def __init__(self, account, vehicle, attribute: str, sensor_name):
"""Initialize the lock."""
self._account = account
self._vehicle = vehicle
super().__init__(account, vehicle)
self._attribute = attribute
self._name = f"{self._vehicle.name} {self._attribute}"
self._unique_id = f"{self._vehicle.vin}-{self._attribute}"
@ -42,14 +41,6 @@ class BMWLock(LockEntity):
DOOR_LOCK_STATE in self._vehicle.available_attributes
)
@property
def should_poll(self):
"""Do not poll this class.
Updates are triggered from BMWConnectedDriveAccount.
"""
return False
@property
def unique_id(self):
"""Return the unique ID of the lock."""
@ -64,10 +55,8 @@ class BMWLock(LockEntity):
def device_state_attributes(self):
"""Return the state attributes of the lock."""
vehicle_state = self._vehicle.state
result = {
"car": self._vehicle.name,
ATTR_ATTRIBUTION: ATTRIBUTION,
}
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
@ -76,7 +65,11 @@ class BMWLock(LockEntity):
@property
def is_locked(self):
"""Return true if lock is locked."""
return self._state == STATE_LOCKED
if self.door_lock_state_available:
result = self._state == STATE_LOCKED
else:
result = None
return result
def lock(self, **kwargs):
"""Lock the car."""
@ -107,14 +100,3 @@ class BMWLock(LockEntity):
if vehicle_state.door_lock_state in [LockState.LOCKED, LockState.SECURED]
else STATE_UNLOCKED
)
def update_callback(self):
"""Schedule a state update."""
self.schedule_update_ha_state(True)
async def async_added_to_hass(self):
"""Add callback after being added to hass.
Show latest data after startup.
"""
self._account.add_update_listener(self.update_callback)

3
homeassistant/components/bmw_connected_drive/manifest.json
View File

@ -3,5 +3,6 @@
"name": "BMW Connected Drive",
"documentation": "https://www.home-assistant.io/integrations/bmw_connected_drive",
"requirements": ["bimmer_connected==0.7.13"],
"codeowners": ["@gerard33", "@rikroe"]
"codeowners": ["@gerard33", "@rikroe"],
"config_flow": true
}

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

@ -11,6 +11,7 @@ from homeassistant.components.notify import (
from homeassistant.const import ATTR_LATITUDE, ATTR_LOCATION, ATTR_LONGITUDE, ATTR_NAME
from . import DOMAIN as BMW_DOMAIN
from .const import CONF_ACCOUNT, DATA_ENTRIES
ATTR_LAT = "lat"
ATTR_LOCATION_ATTRIBUTES = ["street", "city", "postal_code", "country"]
@ -23,7 +24,7 @@ _LOGGER = logging.getLogger(__name__)
def get_service(hass, config, discovery_info=None):
"""Get the BMW notification service."""
accounts = hass.data[BMW_DOMAIN]
accounts = [e[CONF_ACCOUNT] for e in hass.data[BMW_DOMAIN][DATA_ENTRIES].values()]
_LOGGER.debug("Found BMW accounts: %s", ", ".join([a.name for a in accounts]))
svc = BMWNotificationService()
svc.setup(accounts)

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

@ -4,7 +4,6 @@ import logging
from bimmer_connected.state import ChargingState
from homeassistant.const import (
ATTR_ATTRIBUTION,
CONF_UNIT_SYSTEM_IMPERIAL,
LENGTH_KILOMETERS,
LENGTH_MILES,
@ -16,8 +15,8 @@ from homeassistant.const import (
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.icon import icon_for_battery_level
from . import DOMAIN as BMW_DOMAIN
from .const import ATTRIBUTION
from . import DOMAIN as BMW_DOMAIN, BMWConnectedDriveBaseEntity
from .const import CONF_ACCOUNT, DATA_ENTRIES
_LOGGER = logging.getLogger(__name__)
@ -48,48 +47,39 @@ ATTR_TO_HA_IMPERIAL = {
}
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the BMW sensors."""
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the BMW ConnectedDrive sensors from config entry."""
if hass.config.units.name == CONF_UNIT_SYSTEM_IMPERIAL:
attribute_info = ATTR_TO_HA_IMPERIAL
else:
attribute_info = ATTR_TO_HA_METRIC
accounts = hass.data[BMW_DOMAIN]
_LOGGER.debug("Found BMW accounts: %s", ", ".join([a.name for a in accounts]))
devices = []
for account in accounts:
for vehicle in account.account.vehicles:
for attribute_name in vehicle.drive_train_attributes:
if attribute_name in vehicle.available_attributes:
device = BMWConnectedDriveSensor(
account, vehicle, attribute_name, attribute_info
)
devices.append(device)
add_entities(devices, True)
account = hass.data[BMW_DOMAIN][DATA_ENTRIES][config_entry.entry_id][CONF_ACCOUNT]
entities = []
for vehicle in account.account.vehicles:
for attribute_name in vehicle.drive_train_attributes:
if attribute_name in vehicle.available_attributes:
device = BMWConnectedDriveSensor(
account, vehicle, attribute_name, attribute_info
)
entities.append(device)
async_add_entities(entities, True)
class BMWConnectedDriveSensor(Entity):
class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, Entity):
"""Representation of a BMW vehicle sensor."""
def __init__(self, account, vehicle, attribute: str, attribute_info):
"""Initialize BMW vehicle sensor."""
self._vehicle = vehicle
self._account = account
super().__init__(account, vehicle)
self._attribute = attribute
self._state = None
self._name = f"{self._vehicle.name} {self._attribute}"
self._unique_id = f"{self._vehicle.vin}-{self._attribute}"
self._attribute_info = attribute_info
@property
def should_poll(self) -> bool:
"""Return False.
Data update is triggered from BMWConnectedDriveEntity.
"""
return False
@property
def unique_id(self):
"""Return the unique ID of the sensor."""
@ -128,14 +118,6 @@ class BMWConnectedDriveSensor(Entity):
unit = self._attribute_info.get(self._attribute, [None, None])[1]
return unit
@property
def device_state_attributes(self):
"""Return the state attributes of the sensor."""
return {
"car": self._vehicle.name,
ATTR_ATTRIBUTION: ATTRIBUTION,
}
def update(self) -> None:
"""Read new state data from the library."""
_LOGGER.debug("Updating %s", self._vehicle.name)
@ -152,14 +134,3 @@ class BMWConnectedDriveSensor(Entity):
self._state = round(value_converted)
else:
self._state = getattr(vehicle_state, self._attribute)
def update_callback(self):
"""Schedule a state update."""
self.schedule_update_ha_state(True)
async def async_added_to_hass(self):
"""Add callback after being added to hass.
Show latest data after startup.
"""
self._account.add_update_listener(self.update_callback)

30
homeassistant/components/bmw_connected_drive/strings.json Normal file
View File

@ -0,0 +1,30 @@
{
"config": {
"step": {
"user": {
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"region": "ConnectedDrive Region"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
}
},
"options": {
"step": {
"account_options": {
"data": {
"read_only": "Read-only (only sensors and notify, no execution of services, no lock)",
"use_location": "Use Home Assistant location for car location polls (required for non i3/i8 vehicles produced before 7/2014)"
}
}
}
}
}

31
homeassistant/components/bmw_connected_drive/translations/en.json Normal file
View File

@ -0,0 +1,31 @@
{
"config": {
"abort": {
"already_configured": "Account is already configured"
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication"
},
"step": {
"user": {
"data": {
"password": "Password",
"read_only": "Read-only",
"region": "ConnectedDrive Region",
"username": "Username"
}
}
}
},
"options": {
"step": {
"account_options": {
"data": {
"read_only": "Read-only (only sensors and notify, no execution of services, no lock)",
"use_location": "Use Home Assistant location for car location polls (required for non i3/i8 vehicles produced before 7/2014)"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -28,6 +28,7 @@ FLOWS = [
"azure_devops",
"blebox",
"blink",
"bmw_connected_drive",
"bond",
"braviatv",
"broadlink",

3
requirements_test_all.txt
View File

@ -188,6 +188,9 @@ base36==0.1.1
# homeassistant.components.zha
bellows==0.21.0
# homeassistant.components.bmw_connected_drive
bimmer_connected==0.7.13
# homeassistant.components.blebox
blebox_uniapi==1.3.2

1
tests/components/bmw_connected_drive/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the for the BMW Connected Drive integration."""

153
tests/components/bmw_connected_drive/test_config_flow.py Normal file
View File

@ -0,0 +1,153 @@
"""Test the for the BMW Connected Drive config flow."""
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,
CONF_REGION,
CONF_USE_LOCATION,
)
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from tests.async_mock import patch
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, CONF_USE_LOCATION: False},
"system_options": {"disable_new_entities": False},
"source": "user",
"connection_class": config_entries.CONN_CLASS_CLOUD_POLL,
"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."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
async def test_connection_error(hass):
"""Test we show user form on BMW connected drive connection error."""
def _mock_get_oauth_token(*args, **kwargs):
pass
with patch(
"bimmer_connected.account.ConnectedDriveAccount._get_oauth_token",
side_effect=OSError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=FIXTURE_USER_INPUT,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
async def test_full_user_flow_implementation(hass):
"""Test registering an integration and finishing flow works."""
with patch(
"bimmer_connected.account.ConnectedDriveAccount._get_vehicles",
return_value=[],
), patch(
"homeassistant.components.bmw_connected_drive.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.bmw_connected_drive.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=FIXTURE_USER_INPUT,
)
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == FIXTURE_COMPLETE_ENTRY[CONF_USERNAME]
assert result2["data"] == FIXTURE_COMPLETE_ENTRY
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_full_config_flow_implementation(hass):
"""Test registering an integration and finishing flow works."""
with patch(
"bimmer_connected.account.ConnectedDriveAccount._get_vehicles",
return_value=[],
), patch(
"homeassistant.components.bmw_connected_drive.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.bmw_connected_drive.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=FIXTURE_USER_INPUT,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == FIXTURE_IMPORT_ENTRY[CONF_USERNAME]
assert result["data"] == FIXTURE_IMPORT_ENTRY
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_options_flow_implementation(hass):
"""Test config flow options."""
with patch(
"bimmer_connected.account.ConnectedDriveAccount._get_vehicles",
return_value=[],
), patch(
"homeassistant.components.bmw_connected_drive.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.bmw_connected_drive.async_setup_entry",
return_value=True,
) as mock_setup_entry:
config_entry = MockConfigEntry(**FIXTURE_CONFIG_ENTRY)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "account_options"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_READ_ONLY: False, CONF_USE_LOCATION: False},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] == {
CONF_READ_ONLY: False,
CONF_USE_LOCATION: False,
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1