Improve ffmpeg* typing (#108092)

homeassistant/components/ffmpeg/__init__.py

@@ -3,7 +3,9 @@ from __future__ import annotations
 
 import asyncio
 import re
+from typing import Generic, TypeVar
 
+from haffmpeg.core import HAFFmpeg
 from haffmpeg.tools import IMAGE_JPEG, FFVersion, ImageFrame
 import voluptuous as vol
 
@@ -13,9 +15,10 @@ from homeassistant.const import (
     EVENT_HOMEASSISTANT_START,
     EVENT_HOMEASSISTANT_STOP,
 )
-from homeassistant.core import HomeAssistant, ServiceCall, callback
+from homeassistant.core import Event, HomeAssistant, ServiceCall, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.dispatcher import (
+    SignalType,
     async_dispatcher_connect,
     async_dispatcher_send,
 )
@@ -23,15 +26,17 @@ from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.typing import ConfigType
 from homeassistant.loader import bind_hass
 
+_HAFFmpegT = TypeVar("_HAFFmpegT", bound=HAFFmpeg)
+
 DOMAIN = "ffmpeg"
 
 SERVICE_START = "start"
 SERVICE_STOP = "stop"
 SERVICE_RESTART = "restart"
 
-SIGNAL_FFMPEG_START = "ffmpeg.start"
+SIGNAL_FFMPEG_START = SignalType[list[str] | None]("ffmpeg.start")
-SIGNAL_FFMPEG_STOP = "ffmpeg.stop"
+SIGNAL_FFMPEG_STOP = SignalType[list[str] | None]("ffmpeg.stop")
-SIGNAL_FFMPEG_RESTART = "ffmpeg.restart"
+SIGNAL_FFMPEG_RESTART = SignalType[list[str] | None]("ffmpeg.restart")
 
 DATA_FFMPEG = "ffmpeg"
 
@@ -66,7 +71,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     # Register service
     async def async_service_handle(service: ServiceCall) -> None:
         """Handle service ffmpeg process."""
-        entity_ids = service.data.get(ATTR_ENTITY_ID)
+        entity_ids: list[str] | None = service.data.get(ATTR_ENTITY_ID)
 
         if service.service == SERVICE_START:
             async_dispatcher_send(hass, SIGNAL_FFMPEG_START, entity_ids)
@@ -128,20 +133,20 @@ async def async_get_image(
 class FFmpegManager:
     """Helper for ha-ffmpeg."""
 
-    def __init__(self, hass, ffmpeg_bin):
+    def __init__(self, hass: HomeAssistant, ffmpeg_bin: str) -> None:
         """Initialize helper."""
         self.hass = hass
-        self._cache = {}
+        self._cache = {}  # type: ignore[var-annotated]
         self._bin = ffmpeg_bin
-        self._version = None
+        self._version: str | None = None
-        self._major_version = None
+        self._major_version: int | None = None
 
     @property
-    def binary(self):
+    def binary(self) -> str:
         """Return ffmpeg binary from config."""
         return self._bin
 
-    async def async_get_version(self):
+    async def async_get_version(self) -> tuple[str | None, int | None]:
         """Return ffmpeg version."""
 
         ffversion = FFVersion(self._bin)
@@ -156,7 +161,7 @@ class FFmpegManager:
         return self._version, self._major_version
 
     @property
-    def ffmpeg_stream_content_type(self):
+    def ffmpeg_stream_content_type(self) -> str:
         """Return HTTP content type for ffmpeg stream."""
         if self._major_version is not None and self._major_version > 3:
             return CONTENT_TYPE_MULTIPART.format("ffmpeg")
@@ -164,17 +169,17 @@ class FFmpegManager:
         return CONTENT_TYPE_MULTIPART.format("ffserver")
 
 
-class FFmpegBase(Entity):
+class FFmpegBase(Entity, Generic[_HAFFmpegT]):
     """Interface object for FFmpeg."""
 
     _attr_should_poll = False
 
-    def __init__(self, initial_state=True):
+    def __init__(self, ffmpeg: _HAFFmpegT, initial_state: bool = True) -> None:
         """Initialize ffmpeg base object."""
-        self.ffmpeg = None
+        self.ffmpeg = ffmpeg
         self.initial_state = initial_state
 
-    async def async_added_to_hass(self):
+    async def async_added_to_hass(self) -> None:
         """Register dispatcher & events.
 
         This method is a coroutine.
@@ -199,18 +204,18 @@ class FFmpegBase(Entity):
         self._async_register_events()
 
     @property
-    def available(self):
+    def available(self) -> bool:
         """Return True if entity is available."""
         return self.ffmpeg.is_running
 
-    async def _async_start_ffmpeg(self, entity_ids):
+    async def _async_start_ffmpeg(self, entity_ids: list[str] | None) -> None:
         """Start a FFmpeg process.
 
         This method is a coroutine.
         """
         raise NotImplementedError()
 
-    async def _async_stop_ffmpeg(self, entity_ids):
+    async def _async_stop_ffmpeg(self, entity_ids: list[str] | None) -> None:
         """Stop a FFmpeg process.
 
         This method is a coroutine.
@@ -218,7 +223,7 @@ class FFmpegBase(Entity):
         if entity_ids is None or self.entity_id in entity_ids:
             await self.ffmpeg.close()
 
-    async def _async_restart_ffmpeg(self, entity_ids):
+    async def _async_restart_ffmpeg(self, entity_ids: list[str] | None) -> None:
         """Stop a FFmpeg process.
 
         This method is a coroutine.
@@ -228,10 +233,10 @@ class FFmpegBase(Entity):
             await self._async_start_ffmpeg(None)
 
     @callback
-    def _async_register_events(self):
+    def _async_register_events(self) -> None:
         """Register a FFmpeg process/device."""
 
-        async def async_shutdown_handle(event):
+        async def async_shutdown_handle(event: Event) -> None:
             """Stop FFmpeg process."""
             await self._async_stop_ffmpeg(None)
 
@@ -241,7 +246,7 @@ class FFmpegBase(Entity):
         if not self.initial_state:
             return
 
-        async def async_start_handle(event):
+        async def async_start_handle(event: Event) -> None:
             """Start FFmpeg process."""
             await self._async_start_ffmpeg(None)
             self.async_write_ha_state()

homeassistant/components/ffmpeg/camera.py

@@ -60,7 +60,7 @@ class FFmpegCamera(Camera):
         self._input: str = config[CONF_INPUT]
         self._extra_arguments: str = config[CONF_EXTRA_ARGUMENTS]
 
-    async def stream_source(self):
+    async def stream_source(self) -> str:
         """Return the stream source."""
         return self._input.split(" ")[-1]
 
@@ -95,6 +95,6 @@ class FFmpegCamera(Camera):
             await stream.close()
 
     @property
-    def name(self):
+    def name(self) -> str:
         """Return the name of this camera."""
         return self._name

homeassistant/components/ffmpeg_motion/binary_sensor.py

@@ -1,6 +1,9 @@
 """Provides a binary sensor which is a collection of ffmpeg tools."""
 from __future__ import annotations
 
+from typing import Any, TypeVar
+
+from haffmpeg.core import HAFFmpeg
 import haffmpeg.sensor as ffmpeg_sensor
 import voluptuous as vol
 
@@ -14,6 +17,7 @@ from homeassistant.components.ffmpeg import (
     CONF_INITIAL_STATE,
     CONF_INPUT,
     FFmpegBase,
+    FFmpegManager,
     get_ffmpeg_manager,
 )
 from homeassistant.const import CONF_NAME, CONF_REPEAT
@@ -22,6 +26,8 @@ import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
+_HAFFmpegT = TypeVar("_HAFFmpegT", bound=HAFFmpeg)
+
 CONF_RESET = "reset"
 CONF_CHANGES = "changes"
 CONF_REPEAT_TIME = "repeat_time"
@@ -63,43 +69,45 @@ async def async_setup_platform(
     async_add_entities([entity])
 
 
-class FFmpegBinarySensor(FFmpegBase, BinarySensorEntity):
+class FFmpegBinarySensor(FFmpegBase[_HAFFmpegT], BinarySensorEntity):
     """A binary sensor which use FFmpeg for noise detection."""
 
-    def __init__(self, config):
+    def __init__(self, ffmpeg: _HAFFmpegT, config: dict[str, Any]) -> None:
         """Init for the binary sensor noise detection."""
-        super().__init__(config.get(CONF_INITIAL_STATE))
+        super().__init__(ffmpeg, config[CONF_INITIAL_STATE])
 
-        self._state = False
+        self._state: bool | None = False
         self._config = config
-        self._name = config.get(CONF_NAME)
+        self._name: str = config[CONF_NAME]
 
     @callback
-    def _async_callback(self, state):
+    def _async_callback(self, state: bool | None) -> None:
         """HA-FFmpeg callback for noise detection."""
         self._state = state
         self.async_write_ha_state()
 
     @property
-    def is_on(self):
+    def is_on(self) -> bool | None:
         """Return true if the binary sensor is on."""
         return self._state
 
     @property
-    def name(self):
+    def name(self) -> str:
         """Return the name of the entity."""
         return self._name
 
 
-class FFmpegMotion(FFmpegBinarySensor):
+class FFmpegMotion(FFmpegBinarySensor[ffmpeg_sensor.SensorMotion]):
     """A binary sensor which use FFmpeg for noise detection."""
 
-    def __init__(self, hass, manager, config):
+    def __init__(
+        self, hass: HomeAssistant, manager: FFmpegManager, config: dict[str, Any]
+    ) -> None:
         """Initialize FFmpeg motion binary sensor."""
-        super().__init__(config)
+        ffmpeg = ffmpeg_sensor.SensorMotion(manager.binary, self._async_callback)
-        self.ffmpeg = ffmpeg_sensor.SensorMotion(manager.binary, self._async_callback)
+        super().__init__(ffmpeg, config)
 
-    async def _async_start_ffmpeg(self, entity_ids):
+    async def _async_start_ffmpeg(self, entity_ids: list[str] | None) -> None:
         """Start a FFmpeg instance.
 
         This method is a coroutine.
@@ -109,19 +117,19 @@ class FFmpegMotion(FFmpegBinarySensor):
 
         # init config
         self.ffmpeg.set_options(
-            time_reset=self._config.get(CONF_RESET),
+            time_reset=self._config[CONF_RESET],
             time_repeat=self._config.get(CONF_REPEAT_TIME, 0),
             repeat=self._config.get(CONF_REPEAT, 0),
-            changes=self._config.get(CONF_CHANGES),
+            changes=self._config[CONF_CHANGES],
         )
 
         # run
         await self.ffmpeg.open_sensor(
-            input_source=self._config.get(CONF_INPUT),
+            input_source=self._config[CONF_INPUT],
             extra_cmd=self._config.get(CONF_EXTRA_ARGUMENTS),
         )
 
     @property
-    def device_class(self):
+    def device_class(self) -> BinarySensorDeviceClass:
         """Return the class of this sensor, from DEVICE_CLASSES."""
         return BinarySensorDeviceClass.MOTION

homeassistant/components/ffmpeg_noise/binary_sensor.py

@@ -1,6 +1,8 @@
 """Provides a binary sensor which is a collection of ffmpeg tools."""
 from __future__ import annotations
 
+from typing import Any
+
 import haffmpeg.sensor as ffmpeg_sensor
 import voluptuous as vol
 
@@ -13,6 +15,7 @@ from homeassistant.components.ffmpeg import (
     CONF_INITIAL_STATE,
     CONF_INPUT,
     CONF_OUTPUT,
+    FFmpegManager,
     get_ffmpeg_manager,
 )
 from homeassistant.components.ffmpeg_motion.binary_sensor import FFmpegBinarySensor
@@ -59,16 +62,18 @@ async def async_setup_platform(
     async_add_entities([entity])
 
 
-class FFmpegNoise(FFmpegBinarySensor):
+class FFmpegNoise(FFmpegBinarySensor[ffmpeg_sensor.SensorNoise]):
     """A binary sensor which use FFmpeg for noise detection."""
 
-    def __init__(self, hass, manager, config):
+    def __init__(
+        self, hass: HomeAssistant, manager: FFmpegManager, config: dict[str, Any]
+    ) -> None:
         """Initialize FFmpeg noise binary sensor."""
 
-        super().__init__(config)
+        ffmpeg = ffmpeg_sensor.SensorNoise(manager.binary, self._async_callback)
-        self.ffmpeg = ffmpeg_sensor.SensorNoise(manager.binary, self._async_callback)
+        super().__init__(ffmpeg, config)
 
-    async def _async_start_ffmpeg(self, entity_ids):
+    async def _async_start_ffmpeg(self, entity_ids: list[str] | None) -> None:
         """Start a FFmpeg instance.
 
         This method is a coroutine.
@@ -77,18 +82,18 @@ class FFmpegNoise(FFmpegBinarySensor):
             return
 
         self.ffmpeg.set_options(
-            time_duration=self._config.get(CONF_DURATION),
+            time_duration=self._config[CONF_DURATION],
-            time_reset=self._config.get(CONF_RESET),
+            time_reset=self._config[CONF_RESET],
-            peak=self._config.get(CONF_PEAK),
+            peak=self._config[CONF_PEAK],
         )
 
         await self.ffmpeg.open_sensor(
-            input_source=self._config.get(CONF_INPUT),
+            input_source=self._config[CONF_INPUT],
             output_dest=self._config.get(CONF_OUTPUT),
             extra_cmd=self._config.get(CONF_EXTRA_ARGUMENTS),
         )
 
     @property
-    def device_class(self):
+    def device_class(self) -> BinarySensorDeviceClass:
         """Return the class of this sensor, from DEVICE_CLASSES."""
         return BinarySensorDeviceClass.SOUND

tests/components/ffmpeg/test_init.py

@@ -54,7 +54,7 @@ class MockFFmpegDev(ffmpeg.FFmpegBase):
 
     def __init__(self, hass, initial_state=True, entity_id="test.ffmpeg_device"):
         """Initialize mock."""
-        super().__init__(initial_state)
+        super().__init__(None, initial_state)
 
         self.hass = hass
         self.entity_id = entity_id