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

Improve BMWDataUpdateCoordinator typing (#132087)

Browse Source 
Co-authored-by: rikroe <rikroe@users.noreply.github.com>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Richard Kroegel 2024-12-04 00:33:45 +01:00 committed by GitHub
parent 4deaeaeda0
commit 535b47789f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 37 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2,7 +2,6 @@
from __future__ import annotations
from dataclasses import dataclass
import logging
import voluptuous as vol
@ -18,7 +17,7 @@ from homeassistant.helpers import (
import homeassistant.helpers.config_validation as cv
from .const import ATTR_VIN, CONF_READ_ONLY, DOMAIN
from .coordinator import BMWDataUpdateCoordinator
from .coordinator import BMWConfigEntry, BMWDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
@ -49,16 +48,6 @@ PLATFORMS = [
SERVICE_UPDATE_STATE = "update_state"
type BMWConfigEntry = ConfigEntry[BMWData]
@dataclass
class BMWData:
"""Class to store BMW runtime data."""
coordinator: BMWDataUpdateCoordinator
@callback
def _async_migrate_options_from_data_if_missing(
hass: HomeAssistant, entry: ConfigEntry
@ -137,11 +126,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# Set up one data coordinator per account/config entry
coordinator = BMWDataUpdateCoordinator(
hass,
entry=entry,
config_entry=entry,
)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = BMWData(coordinator)
entry.runtime_data = coordinator
# Set up all platforms except notify
await hass.config_entries.async_forward_entry_setups(

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

@ -203,7 +203,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the BMW binary sensors from config entry."""
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
entities = [
BMWBinarySensor(coordinator, vehicle, description, hass.config.units)

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

@ -73,7 +73,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the BMW buttons from config entry."""
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
entities: list[BMWButton] = []

32
homeassistant/components/bmw_connected_drive/coordinator.py
View File

@ -27,34 +27,40 @@ from .const import CONF_GCID, CONF_READ_ONLY, CONF_REFRESH_TOKEN, DOMAIN, SCAN_I
_LOGGER = logging.getLogger(__name__)
type BMWConfigEntry = ConfigEntry[BMWDataUpdateCoordinator]
class BMWDataUpdateCoordinator(DataUpdateCoordinator[None]):
"""Class to manage fetching BMW data."""
account: MyBMWAccount
config_entry: BMWConfigEntry
def __init__(self, hass: HomeAssistant, *, entry: ConfigEntry) -> None:
def __init__(self, hass: HomeAssistant, *, config_entry: ConfigEntry) -> None:
"""Initialize account-wide BMW data updater."""
self.account = MyBMWAccount(
entry.data[CONF_USERNAME],
entry.data[CONF_PASSWORD],
get_region_from_name(entry.data[CONF_REGION]),
config_entry.data[CONF_USERNAME],
config_entry.data[CONF_PASSWORD],
get_region_from_name(config_entry.data[CONF_REGION]),
observer_position=GPSPosition(hass.config.latitude, hass.config.longitude),
verify=get_default_context(),
)
self.read_only = entry.options[CONF_READ_ONLY]
self._entry = entry
self.read_only: bool = config_entry.options[CONF_READ_ONLY]
if CONF_REFRESH_TOKEN in entry.data:
if CONF_REFRESH_TOKEN in config_entry.data:
self.account.set_refresh_token(
refresh_token=entry.data[CONF_REFRESH_TOKEN],
gcid=entry.data.get(CONF_GCID),
refresh_token=config_entry.data[CONF_REFRESH_TOKEN],
gcid=config_entry.data.get(CONF_GCID),
)
super().__init__(
hass,
_LOGGER,
name=f"{DOMAIN}-{entry.data['username']}",
update_interval=timedelta(seconds=SCAN_INTERVALS[entry.data[CONF_REGION]]),
config_entry=config_entry,
name=f"{DOMAIN}-{config_entry.data[CONF_USERNAME]}",
update_interval=timedelta(
seconds=SCAN_INTERVALS[config_entry.data[CONF_REGION]]
),
)
# Default to false on init so _async_update_data logic works
@ -88,9 +94,9 @@ class BMWDataUpdateCoordinator(DataUpdateCoordinator[None]):
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,
**self.config_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)
self.hass.config_entries.async_update_entry(self.config_entry, data=data)

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

@ -27,7 +27,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the MyBMW tracker from config entry."""
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
entities: list[BMWDeviceTracker] = []
for vehicle in coordinator.account.vehicles:

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

@ -51,7 +51,7 @@ async def async_get_config_entry_diagnostics(
hass: HomeAssistant, config_entry: BMWConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
coordinator.account.config.log_responses = True
await coordinator.account.get_vehicles(force_init=True)
@ -77,7 +77,7 @@ async def async_get_device_diagnostics(
hass: HomeAssistant, config_entry: BMWConfigEntry, device: DeviceEntry
) -> dict[str, Any]:
"""Return diagnostics for a device."""
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
coordinator.account.config.log_responses = True
await coordinator.account.get_vehicles(force_init=True)

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

@ -31,7 +31,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the MyBMW lock from config entry."""
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
if not coordinator.read_only:
async_add_entities(

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

@ -53,7 +53,7 @@ def get_service(
targets = {}
if (
config_entry
and (coordinator := config_entry.runtime_data.coordinator)
and (coordinator := config_entry.runtime_data)
and not coordinator.read_only
):
targets.update({v.name: v for v in coordinator.account.vehicles})

2
homeassistant/components/bmw_connected_drive/number.py
View File

@ -61,7 +61,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the MyBMW number from config entry."""
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
entities: list[BMWNumber] = []

2
homeassistant/components/bmw_connected_drive/select.py
View File

@ -68,7 +68,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the MyBMW lock from config entry."""
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
entities: list[BMWSelect] = []

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

@ -193,7 +193,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the MyBMW sensors from config entry."""
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
entities = [
BMWSensor(coordinator, vehicle, description)

2
homeassistant/components/bmw_connected_drive/switch.py
View File

@ -69,7 +69,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the MyBMW switch from config entry."""
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
entities: list[BMWSwitch] = []

8
tests/components/bmw_connected_drive/test_coordinator.py
View File

@ -33,7 +33,7 @@ async def test_update_success(hass: HomeAssistant) -> None:
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.runtime_data.coordinator.last_update_success is True
assert config_entry.runtime_data.last_update_success is True
@pytest.mark.usefixtures("bmw_fixture")
@ -48,7 +48,7 @@ async def test_update_failed(
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
assert coordinator.last_update_success is True
@ -77,7 +77,7 @@ async def test_update_reauth(
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
assert coordinator.last_update_success is True
@ -146,7 +146,7 @@ async def test_captcha_reauth(
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
coordinator = config_entry.runtime_data.coordinator
coordinator = config_entry.runtime_data
assert coordinator.last_update_success is True