From 00221f1d66519b88df9d54ec760b2885bf5a7159 Mon Sep 17 00:00:00 2001 From: Franck Nijhof Date: Tue, 15 Feb 2022 12:32:14 +0100 Subject: [PATCH] Cleanup and strict typing for MJPEG integration (#66526) --- .strict-typing | 1 + homeassistant/components/mjpeg/camera.py | 63 ++++++++++++++++-------- mypy.ini | 11 +++++ 3 files changed, 55 insertions(+), 20 deletions(-) diff --git a/.strict-typing b/.strict-typing index 364a3d0db6a..afd8c0fb93e 100644 --- a/.strict-typing +++ b/.strict-typing @@ -119,6 +119,7 @@ homeassistant.components.lookin.* homeassistant.components.luftdaten.* homeassistant.components.mailbox.* homeassistant.components.media_player.* +homeassistant.components.mjpeg.* homeassistant.components.modbus.* homeassistant.components.modem_callerid.* homeassistant.components.media_source.* diff --git a/homeassistant/components/mjpeg/camera.py b/homeassistant/components/mjpeg/camera.py index 1d60206f2d8..0de303455b9 100644 --- a/homeassistant/components/mjpeg/camera.py +++ b/homeassistant/components/mjpeg/camera.py @@ -2,10 +2,12 @@ from __future__ import annotations import asyncio +from collections.abc import Iterable from contextlib import closing import logging import aiohttp +from aiohttp import web import async_timeout import requests from requests.auth import HTTPBasicAuth, HTTPDigestAuth @@ -65,17 +67,30 @@ async def async_setup_platform( if discovery_info: config = PLATFORM_SCHEMA(discovery_info) - async_add_entities([MjpegCamera(config)]) + + async_add_entities( + [ + MjpegCamera( + config[CONF_NAME], + config[CONF_AUTHENTICATION], + config.get(CONF_USERNAME), + config.get(CONF_PASSWORD), + config[CONF_MJPEG_URL], + config.get(CONF_STILL_IMAGE_URL), + config[CONF_VERIFY_SSL], + ) + ] + ) -def filter_urllib3_logging(): +def filter_urllib3_logging() -> None: """Filter header errors from urllib3 due to a urllib3 bug.""" urllib3_logger = logging.getLogger("urllib3.connectionpool") if not any(isinstance(x, NoHeaderErrorFilter) for x in urllib3_logger.filters): urllib3_logger.addFilter(NoHeaderErrorFilter()) -def extract_image_from_mjpeg(stream): +def extract_image_from_mjpeg(stream: Iterable[bytes]) -> bytes | None: """Take in a MJPEG stream object, return the jpg from it.""" data = b"" @@ -93,19 +108,30 @@ def extract_image_from_mjpeg(stream): return data[jpg_start : jpg_end + 2] + return None + class MjpegCamera(Camera): """An implementation of an IP camera that is reachable over a URL.""" - def __init__(self, device_info): + def __init__( + self, + name: str, + authentication: str, + username: str | None, + password: str | None, + mjpeg_url: str, + still_image_url: str | None, + verify_ssl: bool, + ) -> None: """Initialize a MJPEG camera.""" super().__init__() - self._name = device_info.get(CONF_NAME) - self._authentication = device_info.get(CONF_AUTHENTICATION) - self._username = device_info.get(CONF_USERNAME) - self._password = device_info.get(CONF_PASSWORD) - self._mjpeg_url = device_info[CONF_MJPEG_URL] - self._still_image_url = device_info.get(CONF_STILL_IMAGE_URL) + self._attr_name = name + self._authentication = authentication + self._username = username + self._password = password + self._mjpeg_url = mjpeg_url + self._still_image_url = still_image_url self._auth = None if ( @@ -114,7 +140,7 @@ class MjpegCamera(Camera): and self._authentication == HTTP_BASIC_AUTHENTICATION ): self._auth = aiohttp.BasicAuth(self._username, password=self._password) - self._verify_ssl = device_info.get(CONF_VERIFY_SSL) + self._verify_ssl = verify_ssl async def async_camera_image( self, width: int | None = None, height: int | None = None @@ -137,10 +163,10 @@ class MjpegCamera(Camera): return image except asyncio.TimeoutError: - _LOGGER.error("Timeout getting camera image from %s", self._name) + _LOGGER.error("Timeout getting camera image from %s", self.name) except aiohttp.ClientError as err: - _LOGGER.error("Error getting new camera image from %s: %s", self._name, err) + _LOGGER.error("Error getting new camera image from %s: %s", self.name, err) return None @@ -168,7 +194,9 @@ class MjpegCamera(Camera): with closing(req) as response: return extract_image_from_mjpeg(response.iter_content(102400)) - async def handle_async_mjpeg_stream(self, request): + async def handle_async_mjpeg_stream( + self, request: web.Request + ) -> web.StreamResponse | None: """Generate an HTTP MJPEG stream from the camera.""" # aiohttp don't support DigestAuth -> Fallback if self._authentication == HTTP_DIGEST_AUTHENTICATION: @@ -180,15 +208,10 @@ class MjpegCamera(Camera): return await async_aiohttp_proxy_web(self.hass, request, stream_coro) - @property - def name(self): - """Return the name of this camera.""" - return self._name - class NoHeaderErrorFilter(logging.Filter): """Filter out urllib3 Header Parsing Errors due to a urllib3 bug.""" - def filter(self, record): + def filter(self, record: logging.LogRecord) -> bool: """Filter out Header Parsing Errors.""" return "Failed to parse headers" not in record.getMessage() diff --git a/mypy.ini b/mypy.ini index 8372a98d332..18d951cbe30 100644 --- a/mypy.ini +++ b/mypy.ini @@ -1118,6 +1118,17 @@ no_implicit_optional = true warn_return_any = true warn_unreachable = true +[mypy-homeassistant.components.mjpeg.*] +check_untyped_defs = true +disallow_incomplete_defs = true +disallow_subclassing_any = true +disallow_untyped_calls = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +no_implicit_optional = true +warn_return_any = true +warn_unreachable = true + [mypy-homeassistant.components.modbus.*] check_untyped_defs = true disallow_incomplete_defs = true