Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Mark UVC as strict typed (#123239)

pull/125698/head
Joost Lekkerkerker 2024-09-10 22:02:46 +02:00 committed by GitHub
parent 377ae75e60
commit 688da5389c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 52 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -480,6 +480,7 @@ homeassistant.components.update.*
homeassistant.components.uptime.*
homeassistant.components.uptimerobot.*
homeassistant.components.usb.*
homeassistant.components.uvc.*
homeassistant.components.vacuum.*
homeassistant.components.vallox.*
homeassistant.components.valve.*

90
homeassistant/components/uvc/camera.py
View File

@ -5,9 +5,11 @@ from __future__ import annotations
from datetime import datetime
import logging
import re
from typing import Any, cast
import requests
from uvcclient import camera as uvc_camera, nvr
from uvcclient.camera import UVCCameraClient
from uvcclient.nvr import UVCRemote
import voluptuous as vol
from homeassistant.components.camera import (
@ -57,11 +59,11 @@ def setup_platform(
ssl = config[CONF_SSL]
try:
# Exceptions may be raised in all method calls to the nvr library.
nvrconn = nvr.UVCRemote(addr, port, key, ssl=ssl)
# Exceptions may be raised in all method calls to the nvr library.
cameras = nvrconn.index()
identifier = "id" if nvrconn.server_version >= (3, 2, 0) else "uuid"
identifier = nvrconn.camera_identifier
# Filter out airCam models, which are not supported in the latest
# version of UnifiVideo and which are EOL by Ubiquiti
cameras = [
@ -75,15 +77,12 @@ def setup_platform(
except nvr.NvrError as ex:
_LOGGER.error("NVR refuses to talk to me: %s", str(ex))
raise PlatformNotReady from ex
except requests.exceptions.ConnectionError as ex:
_LOGGER.error("Unable to connect to NVR: %s", str(ex))
raise PlatformNotReady from ex
add_entities(
[
(
UnifiVideoCamera(nvrconn, camera[identifier], camera["name"], password)
for camera in cameras
],
),
True,
)
@ -92,24 +91,19 @@ class UnifiVideoCamera(Camera):
"""A Ubiquiti Unifi Video Camera."""
_attr_should_poll = True # Cameras default to False
_attr_brand = "Ubiquiti"
_attr_is_streaming = False
_caminfo: dict[str, Any]
def __init__(self, camera, uuid, name, password):
def __init__(self, camera: UVCRemote, uuid: str, name: str, password: str) -> None:
"""Initialize an Unifi camera."""
super().__init__()
self._nvr = camera
self._uuid = uuid
self._name = name
self._uuid = self._attr_unique_id = uuid
self._attr_name = name
self._password = password
self._attr_is_streaming = False
self._connect_addr = None
self._camera = None
self._motion_status = False
self._caminfo = None
@property
def name(self):
"""Return the name of this camera."""
return self._name
self._connect_addr: str | None = None
self._camera: UVCCameraClient | None = None
@property
def supported_features(self) -> CameraEntityFeature:
@ -122,7 +116,7 @@ class UnifiVideoCamera(Camera):
return CameraEntityFeature(0)
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the camera state attributes."""
attr = {}
if self.motion_detection_enabled:
@ -145,24 +139,14 @@ class UnifiVideoCamera(Camera):
@property
def motion_detection_enabled(self) -> bool:
"""Camera Motion Detection Status."""
return self._caminfo["recordingSettings"]["motionRecordEnabled"]
return bool(self._caminfo["recordingSettings"]["motionRecordEnabled"])
@property
def unique_id(self) -> str:
"""Return a unique identifier for this client."""
return self._uuid
@property
def brand(self):
"""Return the brand of this camera."""
return "Ubiquiti"
@property
def model(self):
def model(self) -> str:
"""Return the model of this camera."""
return self._caminfo["model"]
return cast(str, self._caminfo["model"])
def _login(self):
def _login(self) -> bool:
"""Login to the camera."""
caminfo = self._caminfo
if self._connect_addr:
@ -170,6 +154,7 @@ class UnifiVideoCamera(Camera):
else:
addrs = [caminfo["host"], caminfo["internalHost"]]
client_cls: type[uvc_camera.UVCCameraClient]
if self._nvr.server_version >= (3, 2, 0):
client_cls = uvc_camera.UVCCameraClientV320
else:
@ -178,15 +163,14 @@ class UnifiVideoCamera(Camera):
if caminfo["username"] is None:
caminfo["username"] = "ubnt"
assert isinstance(caminfo["username"], str)
camera = None
for addr in addrs:
try:
camera = client_cls(addr, caminfo["username"], self._password)
camera.login()
_LOGGER.debug(
"Logged into UVC camera %(name)s via %(addr)s",
{"name": self._name, "addr": addr},
)
_LOGGER.debug("Logged into UVC camera %s via %s", self._attr_name, addr)
self._connect_addr = addr
break
except OSError:
@ -197,7 +181,7 @@ class UnifiVideoCamera(Camera):
pass
if not self._connect_addr:
_LOGGER.error("Unable to login to camera")
return None
return False
self._camera = camera
self._caminfo = caminfo
@ -210,11 +194,13 @@ class UnifiVideoCamera(Camera):
if not self._camera and not self._login():
return None
def _get_image(retry=True):
def _get_image(retry: bool = True) -> bytes | None:
assert self._camera is not None
try:
return self._camera.get_snapshot()
except uvc_camera.CameraConnectError:
_LOGGER.error("Unable to contact camera")
return None
except uvc_camera.CameraAuthError:
if retry:
self._login()
@ -224,13 +210,12 @@ class UnifiVideoCamera(Camera):
return _get_image()
def set_motion_detection(self, mode):
def set_motion_detection(self, mode: bool) -> None:
"""Set motion detection on or off."""
set_mode = "motion" if mode is True else "none"
try:
self._nvr.set_recordmode(self._uuid, set_mode)
self._motion_status = mode
except nvr.NvrError as err:
_LOGGER.error("Unable to set recordmode to %s", set_mode)
_LOGGER.debug(err)
@ -243,16 +228,19 @@ class UnifiVideoCamera(Camera):
"""Disable motion detection in camera."""
self.set_motion_detection(False)
async def stream_source(self):
async def stream_source(self) -> str | None:
"""Return the source of the stream."""
for channel in self._caminfo["channels"]:
if channel["isRtspEnabled"]:
return next(
(
uri
for i, uri in enumerate(channel["rtspUris"])
if re.search(self._nvr._host, uri) # noqa: SLF001
)
return cast(
str,
next(
(
uri
for i, uri in enumerate(channel["rtspUris"])
if re.search(self._nvr._host, uri) # noqa: SLF001
)
),
)
return None

10
mypy.ini
View File

@ -4558,6 +4558,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.uvc.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.vacuum.*]
check_untyped_defs = true
disallow_incomplete_defs = true

5
tests/components/uvc/test_camera.py
View File

@ -4,7 +4,6 @@ from datetime import UTC, datetime, timedelta
from unittest.mock import call, patch
import pytest
import requests
from uvcclient import camera, nvr
from homeassistant.components.camera import (
@ -46,6 +45,7 @@ def mock_remote_fixture(camera_info):
]
mock_remote.return_value.index.return_value = mock_cameras
mock_remote.return_value.server_version = (3, 2, 0)
mock_remote.return_value.camera_identifier = "id"
yield mock_remote
@ -205,6 +205,7 @@ async def test_setup_partial_config_v31x(
"""Test the setup with a v3.1.x server."""
config = {"platform": "uvc", "nvr": "foo", "key": "secret"}
mock_remote.return_value.server_version = (3, 1, 3)
mock_remote.return_value.camera_identifier = "uuid"
assert await async_setup_component(hass, "camera", {"camera": config})
await hass.async_block_till_done()
@ -260,7 +261,6 @@ async def test_setup_incomplete_config(
[
(nvr.NotAuthorized, 0),
(nvr.NvrError, 2),
(requests.exceptions.ConnectionError, 2),
],
)
async def test_setup_nvr_errors_during_indexing(
@ -293,7 +293,6 @@ async def test_setup_nvr_errors_during_indexing(
[
(nvr.NotAuthorized, 0),
(nvr.NvrError, 2),
(requests.exceptions.ConnectionError, 2),
],
)
async def test_setup_nvr_errors_during_initialization(