Improve DataUpdateCoordinator typing in integrations (3) (#84657)

homeassistant/components/nws/__init__.py

@@ -42,7 +42,7 @@ def base_unique_id(latitude, longitude):
     return f"{latitude}_{longitude}"
 
 
-class NwsDataUpdateCoordinator(DataUpdateCoordinator):
+class NwsDataUpdateCoordinator(DataUpdateCoordinator[None]):
     """
     NWS data update coordinator.
 
@@ -57,7 +57,7 @@ class NwsDataUpdateCoordinator(DataUpdateCoordinator):
         name: str,
         update_interval: datetime.timedelta,
         failed_update_interval: datetime.timedelta,
-        update_method: Callable[[], Awaitable] | None = None,
+        update_method: Callable[[], Awaitable[None]] | None = None,
         request_refresh_debouncer: debounce.Debouncer | None = None,
     ) -> None:
         """Initialize NWS coordinator."""

homeassistant/components/nws/sensor.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 
+from pynws import SimpleNWS
+
 from homeassistant.components.sensor import (
     SensorDeviceClass,
     SensorEntity,
@@ -32,7 +34,7 @@ from homeassistant.util.unit_conversion import (
 )
 from homeassistant.util.unit_system import US_CUSTOMARY_SYSTEM
 
-from . import base_unique_id, device_info
+from . import NwsDataUpdateCoordinator, base_unique_id, device_info
 from .const import (
     ATTRIBUTION,
     CONF_STATION,
@@ -163,7 +165,7 @@ async def async_setup_entry(
     )
 
 
-class NWSSensor(CoordinatorEntity, SensorEntity):
+class NWSSensor(CoordinatorEntity[NwsDataUpdateCoordinator], SensorEntity):
     """An NWS Sensor Entity."""
 
     entity_description: NWSSensorEntityDescription
@@ -179,7 +181,7 @@ class NWSSensor(CoordinatorEntity, SensorEntity):
     ):
         """Initialise the platform with a data instance."""
         super().__init__(hass_data[COORDINATOR_OBSERVATION])
-        self._nws = hass_data[NWS_DATA]
+        self._nws: SimpleNWS = hass_data[NWS_DATA]
         self._latitude = entry_data[CONF_LATITUDE]
         self._longitude = entry_data[CONF_LONGITUDE]
         self.entity_description = description

homeassistant/components/powerwall/models.py

@@ -44,7 +44,7 @@ class PowerwallData:
 class PowerwallRuntimeData(TypedDict):
     """Run time data for the powerwall."""
 
-    coordinator: DataUpdateCoordinator | None
+    coordinator: DataUpdateCoordinator[PowerwallData] | None
     base_info: PowerwallBaseInfo
     api_changed: bool
     http_session: Session

homeassistant/components/powerwall/sensor.py

@@ -26,7 +26,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
 from .const import DOMAIN, POWERWALL_COORDINATOR
 from .entity import PowerWallEntity
-from .models import PowerwallData, PowerwallRuntimeData
+from .models import PowerwallRuntimeData
 
 _METER_DIRECTION_EXPORT = "export"
 _METER_DIRECTION_IMPORT = "import"
@@ -114,7 +114,7 @@ async def async_setup_entry(
     powerwall_data: PowerwallRuntimeData = hass.data[DOMAIN][config_entry.entry_id]
     coordinator = powerwall_data[POWERWALL_COORDINATOR]
     assert coordinator is not None
-    data: PowerwallData = coordinator.data
+    data = coordinator.data
     entities: list[PowerWallEntity] = [
         PowerWallChargeSensor(powerwall_data),
     ]

homeassistant/components/rainmachine/__init__.py

@@ -507,7 +507,7 @@ async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
     await hass.config_entries.async_reload(entry.entry_id)
 
 
-class RainMachineEntity(CoordinatorEntity):
+class RainMachineEntity(CoordinatorEntity[RainMachineDataUpdateCoordinator]):
     """Define a generic RainMachine entity."""
 
     _attr_has_entity_name = True

homeassistant/components/sensibo/coordinator.py

@@ -20,11 +20,9 @@ from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER, TIMEOUT
 REQUEST_REFRESH_DELAY = 0.35
 
 
-class SensiboDataUpdateCoordinator(DataUpdateCoordinator):
+class SensiboDataUpdateCoordinator(DataUpdateCoordinator[SensiboData]):
     """A Sensibo Data Update Coordinator."""
 
-    data: SensiboData
-
     def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
         """Initialize the Sensibo coordinator."""
         self.client = SensiboClient(

homeassistant/components/sharkiq/update_coordinator.py

@@ -20,7 +20,7 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
 from .const import API_TIMEOUT, DOMAIN, LOGGER, UPDATE_INTERVAL
 
 
-class SharkIqUpdateCoordinator(DataUpdateCoordinator):
+class SharkIqUpdateCoordinator(DataUpdateCoordinator[bool]):
     """Define a wrapper class to update Shark IQ data."""
 
     def __init__(

homeassistant/components/switcher_kis/__init__.py

@@ -122,7 +122,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     return True
 
 
-class SwitcherDataUpdateCoordinator(update_coordinator.DataUpdateCoordinator):
+class SwitcherDataUpdateCoordinator(
+    update_coordinator.DataUpdateCoordinator[SwitcherBase]
+):
     """Switcher device data update coordinator."""
 
     def __init__(
@@ -138,7 +140,7 @@ class SwitcherDataUpdateCoordinator(update_coordinator.DataUpdateCoordinator):
         self.entry = entry
         self.data = device
 
-    async def _async_update_data(self) -> None:
+    async def _async_update_data(self) -> SwitcherBase:
         """Mark device offline if no data."""
         raise update_coordinator.UpdateFailed(
             f"Device {self.name} did not send update for"

homeassistant/components/venstar/__init__.py

@@ -63,7 +63,7 @@ async def async_unload_entry(hass: HomeAssistant, config: ConfigEntry) -> bool:
     return unload_ok
 
 
-class VenstarDataUpdateCoordinator(update_coordinator.DataUpdateCoordinator):
+class VenstarDataUpdateCoordinator(update_coordinator.DataUpdateCoordinator[None]):
     """Class to manage fetching Venstar data."""
 
     def __init__(

homeassistant/components/venstar/sensor.py

@@ -61,8 +61,8 @@ RUNTIME_ATTRIBUTES = {
 class VenstarSensorTypeMixin:
     """Mixin for sensor required keys."""
 
-    value_fn: Callable[[Any, Any], Any]
-    name_fn: Callable[[Any, Any], str]
+    value_fn: Callable[[VenstarDataUpdateCoordinator, str], Any]
+    name_fn: Callable[[VenstarDataUpdateCoordinator, str], str]
     uom_fn: Callable[[Any], str | None]
 
 
@@ -77,7 +77,7 @@ async def async_setup_entry(
     async_add_entities: AddEntitiesCallback,
 ) -> None:
     """Set up Vensar device binary_sensors based on a config entry."""
-    coordinator = hass.data[DOMAIN][config_entry.entry_id]
+    coordinator: VenstarDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
     entities: list[Entity] = []
 
     if not (sensors := coordinator.client.get_sensor_list()):

homeassistant/components/vizio/__init__.py

@@ -95,7 +95,7 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
     return unload_ok
 
 
-class VizioAppsDataUpdateCoordinator(DataUpdateCoordinator):
+class VizioAppsDataUpdateCoordinator(DataUpdateCoordinator[list[dict[str, Any]]]):
     """Define an object to hold Vizio app config data."""
 
     def __init__(self, hass: HomeAssistant) -> None:

homeassistant/components/vizio/media_player.py

@@ -32,8 +32,8 @@ from homeassistant.helpers.dispatcher import (
 )
 from homeassistant.helpers.entity import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
+from . import VizioAppsDataUpdateCoordinator
 from .const import (
     CONF_ADDITIONAL_CONFIGS,
     CONF_APPS,
@@ -136,7 +136,7 @@ class VizioDevice(MediaPlayerEntity):
         device: VizioAsync,
         name: str,
         device_class: MediaPlayerDeviceClass,
-        apps_coordinator: DataUpdateCoordinator,
+        apps_coordinator: VizioAppsDataUpdateCoordinator,
     ) -> None:
         """Initialize Vizio device."""
         self._config_entry = config_entry
@@ -264,7 +264,9 @@ class VizioDevice(MediaPlayerEntity):
 
         # Create list of available known apps from known app list after
         # filtering by CONF_INCLUDE/CONF_EXCLUDE
-        self._available_apps = self._apps_list([app["name"] for app in self._all_apps])
+        self._available_apps = self._apps_list(
+            [app["name"] for app in self._all_apps or ()]
+        )
 
         self._current_app_config = await self._device.get_current_app_config(
             log_api_exception=False
@@ -272,7 +274,7 @@ class VizioDevice(MediaPlayerEntity):
 
         self._attr_app_name = find_app_name(
             self._current_app_config,
-            [APP_HOME, *self._all_apps, *self._additional_app_configs],
+            [APP_HOME, *(self._all_apps or ()), *self._additional_app_configs],
         )
 
         if self._attr_app_name == NO_APP_RUNNING:

homeassistant/components/xbox/__init__.py

@@ -174,7 +174,7 @@ class XboxData:
     presence: dict[str, PresenceData]
 
 
-class XboxUpdateCoordinator(DataUpdateCoordinator):
+class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]):
     """Store Xbox Console Status."""
 
     def __init__(
@@ -190,7 +190,7 @@ class XboxUpdateCoordinator(DataUpdateCoordinator):
             name=DOMAIN,
             update_interval=timedelta(seconds=10),
         )
-        self.data: XboxData = XboxData({}, {})
+        self.data = XboxData({}, {})
         self.client: XboxLiveClient = client
         self.consoles: SmartglassConsoleList = consoles