Enable type checks for camera platform (#50179)

This commit is contained in:
Ruslan Sayfutdinov 2021-05-09 17:04:57 +01:00 committed by GitHub
parent e9709d4449
commit 9b058551f7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 22 additions and 20 deletions

View File

@ -1,4 +1,6 @@
"""Component to interface with cameras.""" """Component to interface with cameras."""
from __future__ import annotations
import asyncio import asyncio
import base64 import base64
import collections import collections
@ -8,7 +10,7 @@ import hashlib
import logging import logging
import os import os
from random import SystemRandom from random import SystemRandom
from typing import final from typing import cast, final
from aiohttp import web from aiohttp import web
import async_timeout import async_timeout
@ -347,7 +349,7 @@ class Camera(Entity):
"""Return the interval between frames of the mjpeg stream.""" """Return the interval between frames of the mjpeg stream."""
return 0.5 return 0.5
async def create_stream(self) -> Stream: async def create_stream(self) -> Stream | None:
"""Create a Stream for stream_source.""" """Create a Stream for stream_source."""
# There is at most one stream (a decode worker) per camera # There is at most one stream (a decode worker) per camera
if not self.stream: if not self.stream:
@ -471,6 +473,8 @@ class CameraView(HomeAssistantView):
if camera is None: if camera is None:
raise web.HTTPNotFound() raise web.HTTPNotFound()
camera = cast(Camera, camera)
authenticated = ( authenticated = (
request[KEY_AUTHENTICATED] request[KEY_AUTHENTICATED]
or request.query.get("token") in camera.access_tokens or request.query.get("token") in camera.access_tokens
@ -516,13 +520,13 @@ class CameraMjpegStream(CameraView):
async def handle(self, request: web.Request, camera: Camera) -> web.Response: async def handle(self, request: web.Request, camera: Camera) -> web.Response:
"""Serve camera stream, possibly with interval.""" """Serve camera stream, possibly with interval."""
interval = request.query.get("interval") interval_str = request.query.get("interval")
if interval is None: if interval_str is None:
return await camera.handle_async_mjpeg_stream(request) return await camera.handle_async_mjpeg_stream(request)
try: try:
# Compose camera stream from stills # Compose camera stream from stills
interval = float(request.query.get("interval")) interval = float(interval_str)
if interval < MIN_STREAM_INTERVAL: if interval < MIN_STREAM_INTERVAL:
raise ValueError(f"Stream interval must be be > {MIN_STREAM_INTERVAL}") raise ValueError(f"Stream interval must be be > {MIN_STREAM_INTERVAL}")
return await camera.handle_async_still_stream(request, interval) return await camera.handle_async_still_stream(request, interval)
@ -554,7 +558,6 @@ async def websocket_camera_thumbnail(hass, connection, msg):
) )
@websocket_api.async_response
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required("type"): "camera/stream", vol.Required("type"): "camera/stream",
@ -562,6 +565,7 @@ async def websocket_camera_thumbnail(hass, connection, msg):
vol.Optional("format", default="hls"): vol.In(OUTPUT_FORMATS), vol.Optional("format", default="hls"): vol.In(OUTPUT_FORMATS),
} }
) )
@websocket_api.async_response
async def ws_camera_stream(hass, connection, msg): async def ws_camera_stream(hass, connection, msg):
"""Handle get camera stream websocket command. """Handle get camera stream websocket command.
@ -582,17 +586,16 @@ async def ws_camera_stream(hass, connection, msg):
) )
@websocket_api.async_response
@websocket_api.websocket_command( @websocket_api.websocket_command(
{vol.Required("type"): "camera/get_prefs", vol.Required("entity_id"): cv.entity_id} {vol.Required("type"): "camera/get_prefs", vol.Required("entity_id"): cv.entity_id}
) )
@websocket_api.async_response
async def websocket_get_prefs(hass, connection, msg): async def websocket_get_prefs(hass, connection, msg):
"""Handle request for account info.""" """Handle request for account info."""
prefs = hass.data[DATA_CAMERA_PREFS].get(msg["entity_id"]) prefs = hass.data[DATA_CAMERA_PREFS].get(msg["entity_id"])
connection.send_result(msg["id"], prefs.as_dict()) connection.send_result(msg["id"], prefs.as_dict())
@websocket_api.async_response
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required("type"): "camera/update_prefs", vol.Required("type"): "camera/update_prefs",
@ -600,6 +603,7 @@ async def websocket_get_prefs(hass, connection, msg):
vol.Optional("preload_stream"): bool, vol.Optional("preload_stream"): bool,
} }
) )
@websocket_api.async_response
async def websocket_update_prefs(hass, connection, msg): async def websocket_update_prefs(hass, connection, msg):
"""Handle request for account info.""" """Handle request for account info."""
prefs = hass.data[DATA_CAMERA_PREFS] prefs = hass.data[DATA_CAMERA_PREFS]

View File

@ -1,14 +1,16 @@
"""Constants for Camera component.""" """Constants for Camera component."""
DOMAIN = "camera" from typing import Final
DATA_CAMERA_PREFS = "camera_prefs" DOMAIN: Final = "camera"
PREF_PRELOAD_STREAM = "preload_stream" DATA_CAMERA_PREFS: Final = "camera_prefs"
SERVICE_RECORD = "record" PREF_PRELOAD_STREAM: Final = "preload_stream"
CONF_LOOKBACK = "lookback" SERVICE_RECORD: Final = "record"
CONF_DURATION = "duration"
CAMERA_STREAM_SOURCE_TIMEOUT = 10 CONF_LOOKBACK: Final = "lookback"
CAMERA_IMAGE_TIMEOUT = 10 CONF_DURATION: Final = "duration"
CAMERA_STREAM_SOURCE_TIMEOUT: Final = 10
CAMERA_IMAGE_TIMEOUT: Final = 10

View File

@ -710,9 +710,6 @@ ignore_errors = true
[mypy-homeassistant.components.bsblan.*] [mypy-homeassistant.components.bsblan.*]
ignore_errors = true ignore_errors = true
[mypy-homeassistant.components.camera.*]
ignore_errors = true
[mypy-homeassistant.components.canary.*] [mypy-homeassistant.components.canary.*]
ignore_errors = true ignore_errors = true

View File

@ -31,7 +31,6 @@ IGNORED_MODULES: Final[list[str]] = [
"homeassistant.components.bluetooth_tracker.*", "homeassistant.components.bluetooth_tracker.*",
"homeassistant.components.bmw_connected_drive.*", "homeassistant.components.bmw_connected_drive.*",
"homeassistant.components.bsblan.*", "homeassistant.components.bsblan.*",
"homeassistant.components.camera.*",
"homeassistant.components.canary.*", "homeassistant.components.canary.*",
"homeassistant.components.cast.*", "homeassistant.components.cast.*",
"homeassistant.components.cert_expiry.*", "homeassistant.components.cert_expiry.*",