From bc2acb3c0e6562d4515e4593b5f2343b25d97d4d Mon Sep 17 00:00:00 2001 From: Marc Mueller <30130371+cdce8p@users.noreply.github.com> Date: Fri, 19 Jan 2024 08:46:34 +0100 Subject: [PATCH] Improve ffmpeg* typing (#108092) --- homeassistant/components/ffmpeg/__init__.py | 51 ++++++++++--------- homeassistant/components/ffmpeg/camera.py | 4 +- .../components/ffmpeg_motion/binary_sensor.py | 42 ++++++++------- .../components/ffmpeg_noise/binary_sensor.py | 25 +++++---- tests/components/ffmpeg/test_init.py | 2 +- 5 files changed, 71 insertions(+), 53 deletions(-) diff --git a/homeassistant/components/ffmpeg/__init__.py b/homeassistant/components/ffmpeg/__init__.py index a98766c78c6..4ab4ee32a09 100644 --- a/homeassistant/components/ffmpeg/__init__.py +++ b/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_STOP = "ffmpeg.stop" -SIGNAL_FFMPEG_RESTART = "ffmpeg.restart" +SIGNAL_FFMPEG_START = SignalType[list[str] | None]("ffmpeg.start") +SIGNAL_FFMPEG_STOP = SignalType[list[str] | None]("ffmpeg.stop") +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._major_version = None + self._version: str | None = 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() diff --git a/homeassistant/components/ffmpeg/camera.py b/homeassistant/components/ffmpeg/camera.py index b3e9e3f909f..884629c8ae6 100644 --- a/homeassistant/components/ffmpeg/camera.py +++ b/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 diff --git a/homeassistant/components/ffmpeg_motion/binary_sensor.py b/homeassistant/components/ffmpeg_motion/binary_sensor.py index c3603f74a5a..b982d944c6a 100644 --- a/homeassistant/components/ffmpeg_motion/binary_sensor.py +++ b/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) - self.ffmpeg = ffmpeg_sensor.SensorMotion(manager.binary, self._async_callback) + 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 diff --git a/homeassistant/components/ffmpeg_noise/binary_sensor.py b/homeassistant/components/ffmpeg_noise/binary_sensor.py index a7493930a48..a802868334d 100644 --- a/homeassistant/components/ffmpeg_noise/binary_sensor.py +++ b/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) - self.ffmpeg = ffmpeg_sensor.SensorNoise(manager.binary, self._async_callback) + 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_reset=self._config.get(CONF_RESET), - peak=self._config.get(CONF_PEAK), + time_duration=self._config[CONF_DURATION], + time_reset=self._config[CONF_RESET], + 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 diff --git a/tests/components/ffmpeg/test_init.py b/tests/components/ffmpeg/test_init.py index 0c6ce300d01..9a88ef242e8 100644 --- a/tests/components/ffmpeg/test_init.py +++ b/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