Move DataUpdateCoordinator to separate module in subaru (#164918)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

homeassistant/components/subaru/__init__.py

@@ -1,8 +1,6 @@
 """The Subaru integration."""
 
-from datetime import timedelta
 import logging
-import time
 
 from subarulink import Controller as SubaruAPI, InvalidCredentials, SubaruException
 
@@ -18,11 +16,8 @@ from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import aiohttp_client
 from homeassistant.helpers.device_registry import DeviceInfo
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 from .const import (
-    CONF_UPDATE_ENABLED,
-    COORDINATOR_NAME,
     DOMAIN,
     ENTRY_CONTROLLER,
     ENTRY_COORDINATOR,
@@ -42,6 +37,7 @@ from .const import (
     VEHICLE_NAME,
     VEHICLE_VIN,
 )
+from .coordinator import SubaruDataUpdateCoordinator
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -75,20 +71,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         if controller.get_subscription_status(vin):
             vehicle_info[vin] = get_vehicle_info(controller, vin)
 
-    async def async_update_data():
-        """Fetch data from API endpoint."""
-        try:
-            return await refresh_subaru_data(entry, vehicle_info, controller)
-        except SubaruException as err:
-            raise UpdateFailed(err.message) from err
-
-    coordinator = DataUpdateCoordinator(
-        hass,
-        _LOGGER,
-        config_entry=entry,
-        name=COORDINATOR_NAME,
-        update_method=async_update_data,
-        update_interval=timedelta(seconds=FETCH_INTERVAL),
+    coordinator = SubaruDataUpdateCoordinator(
+        hass, entry, controller=controller, vehicle_info=vehicle_info
     )
 
     await coordinator.async_refresh()
@@ -113,41 +97,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     return unload_ok
 
 
-async def refresh_subaru_data(config_entry, vehicle_info, controller):
-    """Refresh local data with data fetched via Subaru API.
-
-    Subaru API calls assume a server side vehicle context
-    Data fetch/update must be done for each vehicle
-    """
-    data = {}
-
-    for vehicle in vehicle_info.values():
-        vin = vehicle[VEHICLE_VIN]
-
-        # Optionally send an "update" remote command to vehicle (throttled with update_interval)
-        if config_entry.options.get(CONF_UPDATE_ENABLED, False):
-            await update_subaru(vehicle, controller)
-
-        # Fetch data from Subaru servers
-        await controller.fetch(vin, force=True)
-
-        # Update our local data that will go to entity states
-        if received_data := await controller.get_data(vin):
-            data[vin] = received_data
-
-    return data
-
-
-async def update_subaru(vehicle, controller):
-    """Commands remote vehicle update (polls the vehicle to update subaru API cache)."""
-    cur_time = time.time()
-    last_update = vehicle[VEHICLE_LAST_UPDATE]
-
-    if cur_time - last_update > controller.get_update_interval():
-        await controller.update(vehicle[VEHICLE_VIN], force=True)
-        vehicle[VEHICLE_LAST_UPDATE] = cur_time
-
-
 def get_vehicle_info(controller, vin):
     """Obtain vehicle identifiers and capabilities."""
     return {

homeassistant/components/subaru/coordinator.py

@@ -0,0 +1,97 @@
+"""Data update coordinator for Subaru."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+import time
+from typing import Any
+
+from subarulink import Controller as SubaruAPI, SubaruException
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import (
+    CONF_UPDATE_ENABLED,
+    COORDINATOR_NAME,
+    FETCH_INTERVAL,
+    VEHICLE_LAST_UPDATE,
+    VEHICLE_VIN,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class SubaruDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
+    """Class to manage fetching Subaru data."""
+
+    config_entry: ConfigEntry
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: ConfigEntry,
+        *,
+        controller: SubaruAPI,
+        vehicle_info: dict[str, dict[str, Any]],
+    ) -> None:
+        """Initialize the coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            config_entry=config_entry,
+            name=COORDINATOR_NAME,
+            update_interval=timedelta(seconds=FETCH_INTERVAL),
+        )
+        self._controller = controller
+        self._vehicle_info = vehicle_info
+
+    async def _async_update_data(self) -> dict[str, Any]:
+        """Fetch data from Subaru API."""
+        try:
+            return await _refresh_subaru_data(
+                self.config_entry, self._vehicle_info, self._controller
+            )
+        except SubaruException as err:
+            raise UpdateFailed(err.message) from err
+
+
+async def _refresh_subaru_data(
+    config_entry: ConfigEntry,
+    vehicle_info: dict[str, dict[str, Any]],
+    controller: SubaruAPI,
+) -> dict[str, Any]:
+    """Refresh local data with data fetched via Subaru API.
+
+    Subaru API calls assume a server side vehicle context
+    Data fetch/update must be done for each vehicle
+    """
+    data: dict[str, Any] = {}
+
+    for vehicle in vehicle_info.values():
+        vin = vehicle[VEHICLE_VIN]
+
+        # Optionally send an "update" remote command to vehicle (throttled with update_interval)
+        if config_entry.options.get(CONF_UPDATE_ENABLED, False):
+            await _update_subaru(vehicle, controller)
+
+        # Fetch data from Subaru servers
+        await controller.fetch(vin, force=True)
+
+        # Update our local data that will go to entity states
+        if received_data := await controller.get_data(vin):
+            data[vin] = received_data
+
+    return data
+
+
+async def _update_subaru(vehicle: dict[str, Any], controller: SubaruAPI) -> None:
+    """Commands remote vehicle update (polls the vehicle to update subaru API cache)."""
+    cur_time = time.time()
+    last_update = vehicle[VEHICLE_LAST_UPDATE]
+
+    if cur_time - last_update > controller.get_update_interval():
+        await controller.update(vehicle[VEHICLE_VIN], force=True)
+        vehicle[VEHICLE_LAST_UPDATE] = cur_time

homeassistant/components/subaru/device_tracker.py

@@ -10,10 +10,7 @@ from homeassistant.components.device_tracker import TrackerEntity
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
-from homeassistant.helpers.update_coordinator import (
-    CoordinatorEntity,
-    DataUpdateCoordinator,
-)
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from . import get_device_info
 from .const import (
@@ -24,6 +21,7 @@ from .const import (
     VEHICLE_STATUS,
     VEHICLE_VIN,
 )
+from .coordinator import SubaruDataUpdateCoordinator
 
 
 async def async_setup_entry(
@@ -33,7 +31,7 @@ async def async_setup_entry(
 ) -> None:
     """Set up the Subaru device tracker by config_entry."""
     entry: dict = hass.data[DOMAIN][config_entry.entry_id]
-    coordinator: DataUpdateCoordinator = entry[ENTRY_COORDINATOR]
+    coordinator: SubaruDataUpdateCoordinator = entry[ENTRY_COORDINATOR]
     vehicle_info: dict = entry[ENTRY_VEHICLES]
     async_add_entities(
         SubaruDeviceTracker(vehicle, coordinator)
@@ -43,7 +41,7 @@ async def async_setup_entry(
 
 
 class SubaruDeviceTracker(
-    CoordinatorEntity[DataUpdateCoordinator[dict[str, Any]]], TrackerEntity
+    CoordinatorEntity[SubaruDataUpdateCoordinator], TrackerEntity
 ):
     """Class for Subaru device tracker."""
 
@@ -51,7 +49,9 @@ class SubaruDeviceTracker(
     _attr_has_entity_name = True
     _attr_name = None
 
-    def __init__(self, vehicle_info: dict, coordinator: DataUpdateCoordinator) -> None:
+    def __init__(
+        self, vehicle_info: dict, coordinator: SubaruDataUpdateCoordinator
+    ) -> None:
         """Initialize the device tracker."""
         super().__init__(coordinator)
         self.vin = vehicle_info[VEHICLE_VIN]

homeassistant/components/subaru/sensor.py

@@ -18,10 +18,7 @@ from homeassistant.const import PERCENTAGE, UnitOfLength, UnitOfPressure, UnitOf
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
-from homeassistant.helpers.update_coordinator import (
-    CoordinatorEntity,
-    DataUpdateCoordinator,
-)
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
 from homeassistant.util.unit_conversion import DistanceConverter, VolumeConverter
 from homeassistant.util.unit_system import METRIC_SYSTEM
 
@@ -37,6 +34,7 @@ from .const import (
     VEHICLE_STATUS,
     VEHICLE_VIN,
 )
+from .coordinator import SubaruDataUpdateCoordinator
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -155,7 +153,7 @@ async def async_setup_entry(
 
 
 def create_vehicle_sensors(
-    vehicle_info, coordinator: DataUpdateCoordinator
+    vehicle_info, coordinator: SubaruDataUpdateCoordinator
 ) -> list[SubaruSensor]:
     """Instantiate all available sensors for the vehicle."""
     sensor_descriptions_to_add = []
@@ -180,9 +178,7 @@ def create_vehicle_sensors(
     ]
 
 
-class SubaruSensor(
-    CoordinatorEntity[DataUpdateCoordinator[dict[str, Any]]], SensorEntity
-):
+class SubaruSensor(CoordinatorEntity[SubaruDataUpdateCoordinator], SensorEntity):
     """Class for Subaru sensors."""
 
     _attr_has_entity_name = True
@@ -190,7 +186,7 @@ class SubaruSensor(
     def __init__(
         self,
         vehicle_info: dict,
-        coordinator: DataUpdateCoordinator,
+        coordinator: SubaruDataUpdateCoordinator,
         description: SensorEntityDescription,
     ) -> None:
         """Initialize the sensor."""