Improve ffmpeg and freebox typing (#108026)

homeassistant/components/ffmpeg/camera.py

@@ -1,6 +1,8 @@
 """Support for Cameras with FFmpeg as decoder."""
 from __future__ import annotations
 
+from typing import Any
+
 from aiohttp import web
 from haffmpeg.camera import CameraMjpeg
 from haffmpeg.tools import IMAGE_JPEG
@@ -14,7 +16,13 @@ import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
-from . import CONF_EXTRA_ARGUMENTS, CONF_INPUT, DATA_FFMPEG, async_get_image
+from . import (
+    CONF_EXTRA_ARGUMENTS,
+    CONF_INPUT,
+    DATA_FFMPEG,
+    FFmpegManager,
+    async_get_image,
+)
 
 DEFAULT_NAME = "FFmpeg"
 DEFAULT_ARGUMENTS = "-pred 1"
@@ -43,14 +51,14 @@ class FFmpegCamera(Camera):
 
     _attr_supported_features = CameraEntityFeature.STREAM
 
-    def __init__(self, hass, config):
+    def __init__(self, hass: HomeAssistant, config: dict[str, Any]) -> None:
         """Initialize a FFmpeg camera."""
         super().__init__()
 
-        self._manager = hass.data[DATA_FFMPEG]
-        self._name = config.get(CONF_NAME)
-        self._input = config.get(CONF_INPUT)
-        self._extra_arguments = config.get(CONF_EXTRA_ARGUMENTS)
+        self._manager: FFmpegManager = hass.data[DATA_FFMPEG]
+        self._name: str = config[CONF_NAME]
+        self._input: str = config[CONF_INPUT]
+        self._extra_arguments: str = config[CONF_EXTRA_ARGUMENTS]
 
     async def stream_source(self):
         """Return the stream source."""

homeassistant/components/freebox/binary_sensor.py

@@ -83,7 +83,7 @@ class FreeboxHomeBinarySensor(FreeboxHomeEntity, BinarySensorEntity):
         )
         self._attr_is_on = self._edit_state(self.get_value("signal", self._sensor_name))
 
-    async def async_update_signal(self):
+    async def async_update_signal(self) -> None:
         """Update name & state."""
         self._attr_is_on = self._edit_state(
             await self.get_home_endpoint_value(self._command_id)
@@ -167,7 +167,7 @@ class FreeboxRaidDegradedSensor(BinarySensorEntity):
         return self._raid["degraded"]
 
     @callback
-    def async_on_demand_update(self):
+    def async_on_demand_update(self) -> None:
         """Update state."""
         self.async_update_state()
         self.async_write_ha_state()

homeassistant/components/freebox/camera.py

@@ -29,11 +29,11 @@ async def async_setup_entry(
     hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
 ) -> None:
     """Set up cameras."""
-    router = hass.data[DOMAIN][entry.unique_id]
-    tracked: set = set()
+    router: FreeboxRouter = hass.data[DOMAIN][entry.unique_id]
+    tracked: set[str] = set()
 
     @callback
-    def update_callback():
+    def update_callback() -> None:
         add_entities(hass, router, async_add_entities, tracked)
 
     router.listeners.append(
@@ -45,9 +45,14 @@ async def async_setup_entry(
 
 
 @callback
-def add_entities(hass: HomeAssistant, router, async_add_entities, tracked):
+def add_entities(
+    hass: HomeAssistant,
+    router: FreeboxRouter,
+    async_add_entities: AddEntitiesCallback,
+    tracked: set[str],
+) -> None:
     """Add new cameras from the router."""
-    new_tracked = []
+    new_tracked: list[FreeboxCamera] = []
 
     for nodeid, node in router.home_devices.items():
         if (node["category"] != FreeboxHomeCategory.CAMERA) or (nodeid in tracked):

homeassistant/components/freebox/device_tracker.py

@@ -103,7 +103,7 @@ class FreeboxDevice(ScannerEntity):
         return SourceType.ROUTER
 
     @callback
-    def async_on_demand_update(self):
+    def async_on_demand_update(self) -> None:
         """Update state."""
         self.async_update_state()
         self.async_write_ha_state()
@@ -120,6 +120,6 @@ class FreeboxDevice(ScannerEntity):
         )
 
 
-def icon_for_freebox_device(device) -> str:
+def icon_for_freebox_device(device: dict[str, Any]) -> str:
     """Return a device icon from its type."""
     return DEVICE_ICONS.get(device["host_type"], "mdi:help-network")

homeassistant/components/freebox/home_base.py

@@ -1,6 +1,7 @@
 """Support for Freebox base features."""
 from __future__ import annotations
 
+from collections.abc import Callable
 import logging
 from typing import Any
 
@@ -42,7 +43,7 @@ class FreeboxHomeEntity(Entity):
         self._available = True
         self._firmware = node["props"].get("FwVersion")
         self._manufacturer = "Freebox SAS"
-        self._remove_signal_update: Any
+        self._remove_signal_update: Callable[[], None] | None = None
 
         self._model = CATEGORY_TO_MODEL.get(node["category"])
         if self._model is None:
@@ -65,7 +66,7 @@ class FreeboxHomeEntity(Entity):
             ),
         )
 
-    async def async_update_signal(self):
+    async def async_update_signal(self) -> None:
         """Update signal."""
         self._node = self._router.home_devices[self._id]
         # Update name
@@ -77,7 +78,9 @@ class FreeboxHomeEntity(Entity):
             )
         self.async_write_ha_state()
 
-    async def set_home_endpoint_value(self, command_id: Any, value=None) -> bool:
+    async def set_home_endpoint_value(
+        self, command_id: int | None, value: bool | None = None
+    ) -> bool:
         """Set Home endpoint value."""
         if command_id is None:
             _LOGGER.error("Unable to SET a value through the API. Command is None")
@@ -97,7 +100,7 @@ class FreeboxHomeEntity(Entity):
         node = await self._router.home.get_home_endpoint_value(self._id, command_id)
         return node.get("value")
 
-    def get_command_id(self, nodes, ep_type, name) -> int | None:
+    def get_command_id(self, nodes, ep_type: str, name: str) -> int | None:
         """Get the command id."""
         node = next(
             filter(lambda x: (x["name"] == name and x["ep_type"] == ep_type), nodes),
@@ -110,7 +113,7 @@ class FreeboxHomeEntity(Entity):
             return None
         return node["id"]
 
-    async def async_added_to_hass(self):
+    async def async_added_to_hass(self) -> None:
         """Register state update callback."""
         self.remove_signal_update(
             async_dispatcher_connect(
@@ -120,11 +123,12 @@ class FreeboxHomeEntity(Entity):
             )
         )
 
-    async def async_will_remove_from_hass(self):
+    async def async_will_remove_from_hass(self) -> None:
         """When entity will be removed from hass."""
-        self._remove_signal_update()
+        if self._remove_signal_update is not None:
+            self._remove_signal_update()
 
-    def remove_signal_update(self, dispacher: Any):
+    def remove_signal_update(self, dispacher: Callable[[], None]) -> None:
         """Register state update callback."""
         self._remove_signal_update = dispacher
 

homeassistant/components/freebox/router.py

@@ -1,7 +1,7 @@
 """Represent the Freebox router and its devices and sensors."""
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from contextlib import suppress
 from datetime import datetime
 import json
@@ -38,7 +38,7 @@ from .const import (
 _LOGGER = logging.getLogger(__name__)
 
 
-def is_json(json_str):
+def is_json(json_str: str) -> bool:
     """Validate if a String is a JSON value or not."""
     try:
         json.loads(json_str)
@@ -95,7 +95,7 @@ class FreeboxRouter:
         self.call_list: list[dict[str, Any]] = []
         self.home_granted = True
         self.home_devices: dict[str, Any] = {}
-        self.listeners: list[dict[str, Any]] = []
+        self.listeners: list[Callable[[], None]] = []
 
     async def update_all(self, now: datetime | None = None) -> None:
         """Update all Freebox platforms."""

homeassistant/components/freebox/switch.py

@@ -51,7 +51,7 @@ class FreeboxSwitch(SwitchEntity):
         self._attr_device_info = router.device_info
         self._attr_unique_id = f"{router.mac} {entity_description.name}"
 
-    async def _async_set_state(self, enabled: bool):
+    async def _async_set_state(self, enabled: bool) -> None:
         """Turn the switch on or off."""
         try:
             await self._router.wifi.set_global_config({"enabled": enabled})