Enable strict typing - bmw_connected_drive (#58506)
This commit is contained in:
parent
b85217c62e
commit
ced89d8f29
9 changed files with 190 additions and 93 deletions
|
@ -20,6 +20,7 @@ homeassistant.components.ampio.*
|
|||
homeassistant.components.automation.*
|
||||
homeassistant.components.binary_sensor.*
|
||||
homeassistant.components.bluetooth_tracker.*
|
||||
homeassistant.components.bmw_connected_drive.*
|
||||
homeassistant.components.bond.*
|
||||
homeassistant.components.braviatv.*
|
||||
homeassistant.components.brother.*
|
||||
|
|
|
@ -3,22 +3,23 @@ from __future__ import annotations
|
|||
|
||||
from collections.abc import Callable
|
||||
import logging
|
||||
from typing import Any, cast
|
||||
|
||||
from bimmer_connected.account import ConnectedDriveAccount
|
||||
from bimmer_connected.country_selector import get_region_from_name
|
||||
from bimmer_connected.vehicle import ConnectedDriveVehicle
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||
from homeassistant.const import (
|
||||
ATTR_ATTRIBUTION,
|
||||
CONF_DEVICE_ID,
|
||||
CONF_NAME,
|
||||
CONF_PASSWORD,
|
||||
CONF_REGION,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.core import HomeAssistant, ServiceCall, callback
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers import device_registry, discovery
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
@ -99,7 +100,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
|||
|
||||
|
||||
@callback
|
||||
def _async_migrate_options_from_data_if_missing(hass, entry):
|
||||
def _async_migrate_options_from_data_if_missing(
|
||||
hass: HomeAssistant, entry: ConfigEntry
|
||||
) -> None:
|
||||
data = dict(entry.data)
|
||||
options = dict(entry.options)
|
||||
|
||||
|
@ -124,7 +127,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||
except OSError as ex:
|
||||
raise ConfigEntryNotReady from ex
|
||||
|
||||
async def _async_update_all(service_call=None):
|
||||
async def _async_update_all(service_call: ServiceCall | None = None) -> None:
|
||||
"""Update all BMW accounts."""
|
||||
await hass.async_add_executor_job(_update_all)
|
||||
|
||||
|
@ -192,18 +195,20 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||
return unload_ok
|
||||
|
||||
|
||||
async def update_listener(hass, config_entry):
|
||||
async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
|
||||
"""Handle options update."""
|
||||
await hass.config_entries.async_reload(config_entry.entry_id)
|
||||
|
||||
|
||||
def setup_account(entry: ConfigEntry, hass, name: str) -> BMWConnectedDriveAccount:
|
||||
def setup_account(
|
||||
entry: ConfigEntry, hass: HomeAssistant, name: str
|
||||
) -> BMWConnectedDriveAccount:
|
||||
"""Set up a new BMWConnectedDriveAccount based on the config."""
|
||||
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]
|
||||
username: str = entry.data[CONF_USERNAME]
|
||||
password: str = entry.data[CONF_PASSWORD]
|
||||
region: str = entry.data[CONF_REGION]
|
||||
read_only: bool = entry.options[CONF_READ_ONLY]
|
||||
use_location: bool = entry.options[CONF_USE_LOCATION]
|
||||
|
||||
_LOGGER.debug("Adding new account %s", name)
|
||||
|
||||
|
@ -214,16 +219,21 @@ def setup_account(entry: ConfigEntry, hass, name: str) -> BMWConnectedDriveAccou
|
|||
username, password, region, name, read_only, *pos
|
||||
)
|
||||
|
||||
def execute_service(call):
|
||||
def execute_service(call: ServiceCall) -> None:
|
||||
"""Execute a service for a vehicle."""
|
||||
vin = call.data.get(ATTR_VIN)
|
||||
device_id = call.data.get(CONF_DEVICE_ID)
|
||||
vin: str | None = call.data.get(ATTR_VIN)
|
||||
device_id: str | None = call.data.get(CONF_DEVICE_ID)
|
||||
|
||||
vehicle = None
|
||||
vehicle: ConnectedDriveVehicle | None = None
|
||||
|
||||
if not vin and device_id:
|
||||
device = device_registry.async_get(hass).async_get(device_id)
|
||||
# If vin is None, device_id must be set (given by SERVICE_SCHEMA)
|
||||
if not (device := device_registry.async_get(hass).async_get(device_id)):
|
||||
_LOGGER.error("Could not find a device for id: %s", device_id)
|
||||
return
|
||||
vin = next(iter(device.identifiers))[1]
|
||||
else:
|
||||
vin = cast(str, vin)
|
||||
|
||||
# Double check for read_only accounts as another account could create the services
|
||||
for entry_data in [
|
||||
|
@ -231,8 +241,8 @@ def setup_account(entry: ConfigEntry, hass, name: str) -> BMWConnectedDriveAccou
|
|||
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:
|
||||
account: ConnectedDriveAccount = entry_data[CONF_ACCOUNT].account
|
||||
if vehicle := account.get_vehicle(vin):
|
||||
break
|
||||
if not vehicle:
|
||||
_LOGGER.error("Could not find a vehicle for VIN %s", vin)
|
||||
|
@ -274,8 +284,8 @@ class BMWConnectedDriveAccount:
|
|||
region_str: str,
|
||||
name: str,
|
||||
read_only: bool,
|
||||
lat=None,
|
||||
lon=None,
|
||||
lat: float | None = None,
|
||||
lon: float | None = None,
|
||||
) -> None:
|
||||
"""Initialize account."""
|
||||
region = get_region_from_name(region_str)
|
||||
|
@ -291,7 +301,7 @@ class BMWConnectedDriveAccount:
|
|||
self.account.set_observer_position(lat, lon)
|
||||
self.account.update_vehicle_states()
|
||||
|
||||
def update(self, *_):
|
||||
def update(self, *_: Any) -> None:
|
||||
"""Update the state of all vehicles.
|
||||
|
||||
Notify all listeners about the update.
|
||||
|
@ -321,15 +331,19 @@ class BMWConnectedDriveBaseEntity(Entity):
|
|||
"""Common base for BMW entities."""
|
||||
|
||||
_attr_should_poll = False
|
||||
_attr_attribution = ATTRIBUTION
|
||||
|
||||
def __init__(self, account, vehicle):
|
||||
def __init__(
|
||||
self,
|
||||
account: BMWConnectedDriveAccount,
|
||||
vehicle: ConnectedDriveVehicle,
|
||||
) -> None:
|
||||
"""Initialize sensor."""
|
||||
self._account = account
|
||||
self._vehicle = vehicle
|
||||
self._attrs = {
|
||||
self._attrs: dict[str, Any] = {
|
||||
"car": self._vehicle.name,
|
||||
"vin": self._vehicle.vin,
|
||||
ATTR_ATTRIBUTION: ATTRIBUTION,
|
||||
}
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, vehicle.vin)},
|
||||
|
@ -338,11 +352,11 @@ class BMWConnectedDriveBaseEntity(Entity):
|
|||
name=f'{vehicle.attributes.get("brand")} {vehicle.name}',
|
||||
)
|
||||
|
||||
def update_callback(self):
|
||||
def update_callback(self) -> None:
|
||||
"""Schedule a state update."""
|
||||
self.schedule_update_ha_state(True)
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Add callback after being added to hass.
|
||||
|
||||
Show latest data after startup.
|
||||
|
|
|
@ -4,11 +4,11 @@ from __future__ import annotations
|
|||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
from typing import Any
|
||||
from typing import Any, cast
|
||||
|
||||
from bimmer_connected.state import ChargingState, LockState
|
||||
from bimmer_connected.state import ChargingState, LockState, VehicleState
|
||||
from bimmer_connected.vehicle import ConnectedDriveVehicle
|
||||
from bimmer_connected.vehicle_status import ConditionBasedServiceReport, VehicleStatus
|
||||
from bimmer_connected.vehicle_status import ConditionBasedServiceReport
|
||||
|
||||
from homeassistant.components.binary_sensor import (
|
||||
DEVICE_CLASS_OPENING,
|
||||
|
@ -34,7 +34,7 @@ _LOGGER = logging.getLogger(__name__)
|
|||
|
||||
|
||||
def _are_doors_closed(
|
||||
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
|
||||
vehicle_state: VehicleState, extra_attributes: dict[str, Any], *args: Any
|
||||
) -> bool:
|
||||
# device class opening: On means open, Off means closed
|
||||
_LOGGER.debug("Status of lid: %s", vehicle_state.all_lids_closed)
|
||||
|
@ -44,7 +44,7 @@ def _are_doors_closed(
|
|||
|
||||
|
||||
def _are_windows_closed(
|
||||
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
|
||||
vehicle_state: VehicleState, extra_attributes: dict[str, Any], *args: Any
|
||||
) -> bool:
|
||||
# device class opening: On means open, Off means closed
|
||||
for window in vehicle_state.windows:
|
||||
|
@ -53,7 +53,7 @@ def _are_windows_closed(
|
|||
|
||||
|
||||
def _are_doors_locked(
|
||||
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
|
||||
vehicle_state: VehicleState, extra_attributes: dict[str, Any], *args: Any
|
||||
) -> bool:
|
||||
# device class lock: On means unlocked, Off means locked
|
||||
# Possible values: LOCKED, SECURED, SELECTIVE_LOCKED, UNLOCKED
|
||||
|
@ -63,15 +63,15 @@ def _are_doors_locked(
|
|||
|
||||
|
||||
def _are_parking_lights_on(
|
||||
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
|
||||
vehicle_state: VehicleState, extra_attributes: dict[str, Any], *args: Any
|
||||
) -> bool:
|
||||
# device class light: On means light detected, Off means no light
|
||||
extra_attributes["lights_parking"] = vehicle_state.parking_lights.value
|
||||
return vehicle_state.are_parking_lights_on
|
||||
return cast(bool, vehicle_state.are_parking_lights_on)
|
||||
|
||||
|
||||
def _are_problems_detected(
|
||||
vehicle_state: VehicleStatus,
|
||||
vehicle_state: VehicleState,
|
||||
extra_attributes: dict[str, Any],
|
||||
unit_system: UnitSystem,
|
||||
) -> bool:
|
||||
|
@ -82,7 +82,7 @@ def _are_problems_detected(
|
|||
|
||||
|
||||
def _check_control_messages(
|
||||
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
|
||||
vehicle_state: VehicleState, extra_attributes: dict[str, Any], *args: Any
|
||||
) -> bool:
|
||||
# device class problem: On means problem detected, Off means no problem
|
||||
check_control_messages = vehicle_state.check_control_messages
|
||||
|
@ -92,27 +92,27 @@ def _check_control_messages(
|
|||
extra_attributes["check_control_messages"] = cbs_list
|
||||
else:
|
||||
extra_attributes["check_control_messages"] = "OK"
|
||||
return vehicle_state.has_check_control_messages
|
||||
return cast(bool, vehicle_state.has_check_control_messages)
|
||||
|
||||
|
||||
def _is_vehicle_charging(
|
||||
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
|
||||
vehicle_state: VehicleState, extra_attributes: dict[str, Any], *args: Any
|
||||
) -> bool:
|
||||
# device class power: On means power detected, Off means no power
|
||||
extra_attributes["charging_status"] = vehicle_state.charging_status.value
|
||||
extra_attributes[
|
||||
"last_charging_end_result"
|
||||
] = vehicle_state.last_charging_end_result
|
||||
return vehicle_state.charging_status == ChargingState.CHARGING
|
||||
return cast(bool, vehicle_state.charging_status == ChargingState.CHARGING)
|
||||
|
||||
|
||||
def _is_vehicle_plugged_in(
|
||||
vehicle_state: VehicleStatus, extra_attributes: dict[str, Any], *args: Any
|
||||
vehicle_state: VehicleState, extra_attributes: dict[str, Any], *args: Any
|
||||
) -> bool:
|
||||
# device class plug: On means device is plugged in,
|
||||
# Off means device is unplugged
|
||||
extra_attributes["connection_status"] = vehicle_state.connection_status
|
||||
return vehicle_state.connection_status == "CONNECTED"
|
||||
return cast(str, vehicle_state.connection_status) == "CONNECTED"
|
||||
|
||||
|
||||
def _format_cbs_report(
|
||||
|
@ -133,7 +133,7 @@ def _format_cbs_report(
|
|||
class BMWRequiredKeysMixin:
|
||||
"""Mixin for required keys."""
|
||||
|
||||
value_fn: Callable[[VehicleStatus, dict[str, Any], UnitSystem], bool]
|
||||
value_fn: Callable[[VehicleState, dict[str, Any], UnitSystem], bool]
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
|
@ -1,4 +1,8 @@
|
|||
"""Config flow for BMW ConnectedDrive integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from bimmer_connected.account import ConnectedDriveAccount
|
||||
from bimmer_connected.country_selector import get_region_from_name
|
||||
import voluptuous as vol
|
||||
|
@ -6,6 +10,7 @@ import voluptuous as vol
|
|||
from homeassistant import config_entries, core, exceptions
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_REGION, CONF_SOURCE, CONF_USERNAME
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
|
||||
from . import DOMAIN
|
||||
from .const import CONF_ALLOWED_REGIONS, CONF_READ_ONLY, CONF_USE_LOCATION
|
||||
|
@ -19,7 +24,9 @@ DATA_SCHEMA = vol.Schema(
|
|||
)
|
||||
|
||||
|
||||
async def validate_input(hass: core.HomeAssistant, data):
|
||||
async def validate_input(
|
||||
hass: core.HomeAssistant, data: dict[str, Any]
|
||||
) -> dict[str, str]:
|
||||
"""Validate the user input allows us to connect.
|
||||
|
||||
Data has the keys from DATA_SCHEMA with values provided by the user.
|
||||
|
@ -43,9 +50,11 @@ class BMWConnectedDriveConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
|
||||
VERSION = 1
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the initial step."""
|
||||
errors = {}
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
unique_id = f"{user_input[CONF_REGION]}-{user_input[CONF_USERNAME]}"
|
||||
|
||||
|
@ -65,13 +74,15 @@ class BMWConnectedDriveConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
step_id="user", data_schema=DATA_SCHEMA, errors=errors
|
||||
)
|
||||
|
||||
async def async_step_import(self, user_input):
|
||||
async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
|
||||
"""Handle import."""
|
||||
return await self.async_step_user(user_input)
|
||||
|
||||
@staticmethod
|
||||
@callback
|
||||
def async_get_options_flow(config_entry):
|
||||
def async_get_options_flow(
|
||||
config_entry: config_entries.ConfigEntry,
|
||||
) -> BMWConnectedDriveOptionsFlow:
|
||||
"""Return a BWM ConnectedDrive option flow."""
|
||||
return BMWConnectedDriveOptionsFlow(config_entry)
|
||||
|
||||
|
@ -79,16 +90,20 @@ class BMWConnectedDriveConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
class BMWConnectedDriveOptionsFlow(config_entries.OptionsFlow):
|
||||
"""Handle a option flow for BMW ConnectedDrive."""
|
||||
|
||||
def __init__(self, config_entry):
|
||||
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
|
||||
"""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):
|
||||
async def async_step_init(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Manage the options."""
|
||||
return await self.async_step_account_options()
|
||||
|
||||
async def async_step_account_options(self, user_input=None):
|
||||
async def async_step_account_options(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the initial step."""
|
||||
if user_input is not None:
|
||||
return self.async_create_entry(title="", data=user_input)
|
||||
|
|
|
@ -1,19 +1,37 @@
|
|||
"""Device tracker for BMW Connected Drive vehicles."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Literal
|
||||
|
||||
from bimmer_connected.vehicle import ConnectedDriveVehicle
|
||||
|
||||
from homeassistant.components.device_tracker import SOURCE_TYPE_GPS
|
||||
from homeassistant.components.device_tracker.config_entry import TrackerEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from . import DOMAIN as BMW_DOMAIN, BMWConnectedDriveBaseEntity
|
||||
from . import (
|
||||
DOMAIN as BMW_DOMAIN,
|
||||
BMWConnectedDriveAccount,
|
||||
BMWConnectedDriveBaseEntity,
|
||||
)
|
||||
from .const import CONF_ACCOUNT, DATA_ENTRIES
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the BMW ConnectedDrive tracker from config entry."""
|
||||
account = hass.data[BMW_DOMAIN][DATA_ENTRIES][config_entry.entry_id][CONF_ACCOUNT]
|
||||
entities = []
|
||||
account: BMWConnectedDriveAccount = hass.data[BMW_DOMAIN][DATA_ENTRIES][
|
||||
config_entry.entry_id
|
||||
][CONF_ACCOUNT]
|
||||
entities: list[BMWDeviceTracker] = []
|
||||
|
||||
for vehicle in account.account.vehicles:
|
||||
entities.append(BMWDeviceTracker(account, vehicle))
|
||||
|
@ -32,36 +50,38 @@ class BMWDeviceTracker(BMWConnectedDriveBaseEntity, TrackerEntity):
|
|||
_attr_force_update = False
|
||||
_attr_icon = "mdi:car"
|
||||
|
||||
def __init__(self, account, vehicle):
|
||||
def __init__(
|
||||
self,
|
||||
account: BMWConnectedDriveAccount,
|
||||
vehicle: ConnectedDriveVehicle,
|
||||
) -> None:
|
||||
"""Initialize the Tracker."""
|
||||
super().__init__(account, vehicle)
|
||||
|
||||
self._attr_unique_id = vehicle.vin
|
||||
self._location = (
|
||||
vehicle.state.gps_position if vehicle.state.gps_position else (None, None)
|
||||
)
|
||||
self._location = pos if (pos := vehicle.state.gps_position) else None
|
||||
self._attr_name = vehicle.name
|
||||
|
||||
@property
|
||||
def latitude(self):
|
||||
def latitude(self) -> float | None:
|
||||
"""Return latitude value of the device."""
|
||||
return self._location[0] if self._location else None
|
||||
|
||||
@property
|
||||
def longitude(self):
|
||||
def longitude(self) -> float | None:
|
||||
"""Return longitude value of the device."""
|
||||
return self._location[1] if self._location else None
|
||||
|
||||
@property
|
||||
def source_type(self):
|
||||
def source_type(self) -> Literal["gps"]:
|
||||
"""Return the source type, eg gps or router, of the device."""
|
||||
return SOURCE_TYPE_GPS
|
||||
|
||||
def update(self):
|
||||
def update(self) -> None:
|
||||
"""Update state of the decvice tracker."""
|
||||
self._attr_extra_state_attributes = self._attrs
|
||||
self._location = (
|
||||
self._vehicle.state.gps_position
|
||||
if self._vehicle.state.is_vehicle_tracking_enabled
|
||||
else (None, None)
|
||||
else None
|
||||
)
|
||||
|
|
|
@ -1,33 +1,54 @@
|
|||
"""Support for BMW car locks with BMW ConnectedDrive."""
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from bimmer_connected.state import LockState
|
||||
from bimmer_connected.vehicle import ConnectedDriveVehicle
|
||||
|
||||
from homeassistant.components.lock import LockEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from . import DOMAIN as BMW_DOMAIN, BMWConnectedDriveBaseEntity
|
||||
from . import (
|
||||
DOMAIN as BMW_DOMAIN,
|
||||
BMWConnectedDriveAccount,
|
||||
BMWConnectedDriveBaseEntity,
|
||||
)
|
||||
from .const import CONF_ACCOUNT, DATA_ENTRIES
|
||||
|
||||
DOOR_LOCK_STATE = "door_lock_state"
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the BMW ConnectedDrive binary sensors from config entry."""
|
||||
account = hass.data[BMW_DOMAIN][DATA_ENTRIES][config_entry.entry_id][CONF_ACCOUNT]
|
||||
entities = []
|
||||
account: BMWConnectedDriveAccount = hass.data[BMW_DOMAIN][DATA_ENTRIES][
|
||||
config_entry.entry_id
|
||||
][CONF_ACCOUNT]
|
||||
|
||||
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)
|
||||
entities = [
|
||||
BMWLock(account, vehicle, "lock", "BMW lock")
|
||||
for vehicle in account.account.vehicles
|
||||
]
|
||||
async_add_entities(entities, True)
|
||||
|
||||
|
||||
class BMWLock(BMWConnectedDriveBaseEntity, LockEntity):
|
||||
"""Representation of a BMW vehicle lock."""
|
||||
|
||||
def __init__(self, account, vehicle, attribute: str, sensor_name):
|
||||
def __init__(
|
||||
self,
|
||||
account: BMWConnectedDriveAccount,
|
||||
vehicle: ConnectedDriveVehicle,
|
||||
attribute: str,
|
||||
sensor_name: str,
|
||||
) -> None:
|
||||
"""Initialize the lock."""
|
||||
super().__init__(account, vehicle)
|
||||
|
||||
|
@ -37,7 +58,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):
|
||||
def lock(self, **kwargs: Any) -> None:
|
||||
"""Lock the car."""
|
||||
_LOGGER.debug("%s: locking doors", self._vehicle.name)
|
||||
# Optimistic state set here because it takes some time before the
|
||||
|
@ -46,7 +67,7 @@ class BMWLock(BMWConnectedDriveBaseEntity, LockEntity):
|
|||
self.schedule_update_ha_state()
|
||||
self._vehicle.remote_services.trigger_remote_door_lock()
|
||||
|
||||
def unlock(self, **kwargs):
|
||||
def unlock(self, **kwargs: Any) -> None:
|
||||
"""Unlock the car."""
|
||||
_LOGGER.debug("%s: unlocking doors", self._vehicle.name)
|
||||
# Optimistic state set here because it takes some time before the
|
||||
|
@ -55,17 +76,18 @@ class BMWLock(BMWConnectedDriveBaseEntity, LockEntity):
|
|||
self.schedule_update_ha_state()
|
||||
self._vehicle.remote_services.trigger_remote_door_unlock()
|
||||
|
||||
def update(self):
|
||||
def update(self) -> None:
|
||||
"""Update state of the lock."""
|
||||
_LOGGER.debug("%s: updating data for %s", self._vehicle.name, self._attribute)
|
||||
if self._vehicle.state.door_lock_state in [LockState.LOCKED, LockState.SECURED]:
|
||||
self._attr_is_locked = True
|
||||
else:
|
||||
self._attr_is_locked = False
|
||||
vehicle_state = self._vehicle.state
|
||||
if not self.door_lock_state_available:
|
||||
self._attr_is_locked = None
|
||||
else:
|
||||
self._attr_is_locked = vehicle_state.door_lock_state in {
|
||||
LockState.LOCKED,
|
||||
LockState.SECURED,
|
||||
}
|
||||
|
||||
vehicle_state = self._vehicle.state
|
||||
result = self._attrs.copy()
|
||||
if self.door_lock_state_available:
|
||||
result["door_lock_state"] = vehicle_state.door_lock_state.value
|
||||
|
|
|
@ -1,5 +1,10 @@
|
|||
"""Support for BMW notifications."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any, cast
|
||||
|
||||
from bimmer_connected.vehicle import ConnectedDriveVehicle
|
||||
|
||||
from homeassistant.components.notify import (
|
||||
ATTR_DATA,
|
||||
|
@ -9,8 +14,10 @@ from homeassistant.components.notify import (
|
|||
BaseNotificationService,
|
||||
)
|
||||
from homeassistant.const import ATTR_LATITUDE, ATTR_LOCATION, ATTR_LONGITUDE, ATTR_NAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
from . import DOMAIN as BMW_DOMAIN
|
||||
from . import DOMAIN as BMW_DOMAIN, BMWConnectedDriveAccount
|
||||
from .const import CONF_ACCOUNT, DATA_ENTRIES
|
||||
|
||||
ATTR_LAT = "lat"
|
||||
|
@ -22,9 +29,15 @@ ATTR_TEXT = "text"
|
|||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def get_service(hass, config, discovery_info=None):
|
||||
def get_service(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> BMWNotificationService:
|
||||
"""Get the BMW notification service."""
|
||||
accounts = [e[CONF_ACCOUNT] for e in hass.data[BMW_DOMAIN][DATA_ENTRIES].values()]
|
||||
accounts: list[BMWConnectedDriveAccount] = [
|
||||
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)
|
||||
|
@ -34,22 +47,23 @@ def get_service(hass, config, discovery_info=None):
|
|||
class BMWNotificationService(BaseNotificationService):
|
||||
"""Send Notifications to BMW."""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
"""Set up the notification service."""
|
||||
self.targets = {}
|
||||
self.targets: dict[str, ConnectedDriveVehicle] = {}
|
||||
|
||||
def setup(self, accounts):
|
||||
def setup(self, accounts: list[BMWConnectedDriveAccount]) -> None:
|
||||
"""Get the BMW vehicle(s) for the account(s)."""
|
||||
for account in accounts:
|
||||
self.targets.update({v.name: v for v in account.account.vehicles})
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message or POI to the car."""
|
||||
for _vehicle in kwargs[ATTR_TARGET]:
|
||||
_LOGGER.debug("Sending message to %s", _vehicle.name)
|
||||
for vehicle in kwargs[ATTR_TARGET]:
|
||||
vehicle = cast(ConnectedDriveVehicle, vehicle)
|
||||
_LOGGER.debug("Sending message to %s", vehicle.name)
|
||||
|
||||
# Extract params from data dict
|
||||
title = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
|
||||
title: str = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
|
||||
data = kwargs.get(ATTR_DATA)
|
||||
|
||||
# Check if message is a POI
|
||||
|
@ -68,8 +82,8 @@ class BMWNotificationService(BaseNotificationService):
|
|||
}
|
||||
)
|
||||
|
||||
_vehicle.remote_services.trigger_send_poi(location_dict)
|
||||
vehicle.remote_services.trigger_send_poi(location_dict)
|
||||
else:
|
||||
_vehicle.remote_services.trigger_send_message(
|
||||
vehicle.remote_services.trigger_send_message(
|
||||
{ATTR_TEXT: message, ATTR_SUBJECT: title}
|
||||
)
|
||||
|
|
|
@ -578,7 +578,7 @@ class BMWConnectedDriveSensor(BMWConnectedDriveBaseEntity, SensorEntity):
|
|||
self._attr_native_value = getattr(vehicle_all_trips, sensor_key)
|
||||
|
||||
vehicle_state = self._vehicle.state
|
||||
charging_state = vehicle_state.charging_status in [ChargingState.CHARGING]
|
||||
charging_state = vehicle_state.charging_status in {ChargingState.CHARGING}
|
||||
|
||||
if sensor_key == "charging_level_hv":
|
||||
self._attr_icon = icon_for_battery_level(
|
||||
|
|
11
mypy.ini
11
mypy.ini
|
@ -231,6 +231,17 @@ no_implicit_optional = true
|
|||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.bmw_connected_drive.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
disallow_untyped_decorators = true
|
||||
disallow_untyped_defs = true
|
||||
no_implicit_optional = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.bond.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
|
Loading…
Add table
Reference in a new issue