Move medcom_ble coordinator to separate module (home-assistant#148009)

Author: epenet

homeassistant/components/medcom_ble/__init__.py

@@ -2,34 +2,23 @@
 
 from __future__ import annotations
 
-from datetime import timedelta
-import logging
-
-from bleak import BleakError
-from medcom_ble import MedcomBleDeviceData
-
 from homeassistant.components import bluetooth
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
-from homeassistant.util.unit_system import METRIC_SYSTEM
 
-from .const import DEFAULT_SCAN_INTERVAL, DOMAIN
+from .const import DOMAIN
+from .coordinator import MedcomBleUpdateCoordinator
 
 # Supported platforms
 PLATFORMS: list[Platform] = [Platform.SENSOR]
 
-_LOGGER = logging.getLogger(__name__)
-
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up Medcom BLE radiation monitor from a config entry."""
 
     address = entry.unique_id
-    elevation = hass.config.elevation
-    is_metric = hass.config.units is METRIC_SYSTEM
     assert address is not None
 
     ble_device = bluetooth.async_ble_device_from_address(hass, address)
@@ -38,26 +27,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             f"Could not find Medcom BLE device with address {address}"
         )
 
-    async def _async_update_method():
-        """Get data from Medcom BLE radiation monitor."""
-        ble_device = bluetooth.async_ble_device_from_address(hass, address)
-        inspector = MedcomBleDeviceData(_LOGGER, elevation, is_metric)
-
-        try:
-            data = await inspector.update_device(ble_device)
-        except BleakError as err:
-            raise UpdateFailed(f"Unable to fetch data: {err}") from err
-
-        return data
-
-    coordinator = DataUpdateCoordinator(
-        hass,
-        _LOGGER,
-        config_entry=entry,
-        name=DOMAIN,
-        update_method=_async_update_method,
-        update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
-    )
+    coordinator = MedcomBleUpdateCoordinator(hass, entry, address)
 
     await coordinator.async_config_entry_first_refresh()
 

homeassistant/components/medcom_ble/coordinator.py

@@ -0,0 +1,50 @@
+"""The Medcom BLE integration."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+
+from bleak import BleakError
+from medcom_ble import MedcomBleDevice, MedcomBleDeviceData
+
+from homeassistant.components import bluetooth
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+from homeassistant.util.unit_system import METRIC_SYSTEM
+
+from .const import DEFAULT_SCAN_INTERVAL, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class MedcomBleUpdateCoordinator(DataUpdateCoordinator[MedcomBleDevice]):
+    """Coordinator for Medcom BLE radiation monitor data."""
+
+    config_entry: ConfigEntry
+
+    def __init__(self, hass: HomeAssistant, entry: ConfigEntry, address: str) -> None:
+        """Initialize the coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            config_entry=entry,
+            name=DOMAIN,
+            update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
+        )
+        self._address = address
+        self._elevation = hass.config.elevation
+        self._is_metric = hass.config.units is METRIC_SYSTEM
+
+    async def _async_update_data(self) -> MedcomBleDevice:
+        """Get data from Medcom BLE radiation monitor."""
+        ble_device = bluetooth.async_ble_device_from_address(self.hass, self._address)
+        inspector = MedcomBleDeviceData(_LOGGER, self._elevation, self._is_metric)
+
+        try:
+            data = await inspector.update_device(ble_device)
+        except BleakError as err:
+            raise UpdateFailed(f"Unable to fetch data: {err}") from err
+
+        return data

homeassistant/components/medcom_ble/sensor.py

@@ -4,8 +4,6 @@
 
 import logging
 
-from medcom_ble import MedcomBleDevice
-
 from homeassistant import config_entries
 from homeassistant.components.sensor import (
     SensorEntity,
@@ -15,12 +13,10 @@
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
-from homeassistant.helpers.update_coordinator import (
-    CoordinatorEntity,
-    DataUpdateCoordinator,
-)
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import DOMAIN, UNIT_CPM
+from .coordinator import MedcomBleUpdateCoordinator
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -41,9 +37,7 @@ async def async_setup_entry(
 ) -> None:
     """Set up Medcom BLE radiation monitor sensors."""
 
-    coordinator: DataUpdateCoordinator[MedcomBleDevice] = hass.data[DOMAIN][
-        entry.entry_id
-    ]
+    coordinator: MedcomBleUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
 
     entities = []
     _LOGGER.debug("got sensors: %s", coordinator.data.sensors)
@@ -62,16 +56,14 @@ async def async_setup_entry(
     async_add_entities(entities)
 
 
-class MedcomSensor(
-    CoordinatorEntity[DataUpdateCoordinator[MedcomBleDevice]], SensorEntity
-):
+class MedcomSensor(CoordinatorEntity[MedcomBleUpdateCoordinator], SensorEntity):
     """Medcom BLE radiation monitor sensors for the device."""
 
     _attr_has_entity_name = True
 
     def __init__(
         self,
-        coordinator: DataUpdateCoordinator[MedcomBleDevice],
+        coordinator: MedcomBleUpdateCoordinator,
         entity_description: SensorEntityDescription,
     ) -> None:
         """Populate the medcom entity with relevant data."""