Improve type hints in axis (#75910)

This commit is contained in:
epenet 2022-07-30 11:04:31 +02:00 committed by GitHub
parent bb3e094552
commit 8181da7090
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 61 additions and 40 deletions

View File

@ -26,9 +26,8 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
except AuthenticationRequired as err: except AuthenticationRequired as err:
raise ConfigEntryAuthFailed from err raise ConfigEntryAuthFailed from err
device = hass.data[AXIS_DOMAIN][config_entry.unique_id] = AxisNetworkDevice( device = AxisNetworkDevice(hass, config_entry, api)
hass, config_entry, api hass.data[AXIS_DOMAIN][config_entry.unique_id] = device
)
await device.async_update_device_registry() await device.async_update_device_registry()
await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS) await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
device.async_setup_events() device.async_setup_events()
@ -43,7 +42,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload Axis device config entry.""" """Unload Axis device config entry."""
device = hass.data[AXIS_DOMAIN].pop(config_entry.unique_id) device: AxisNetworkDevice = hass.data[AXIS_DOMAIN].pop(config_entry.unique_id)
return await device.async_reset() return await device.async_reset()

View File

@ -1,10 +1,12 @@
"""Base classes for Axis entities.""" """Base classes for Axis entities."""
from axis.event_stream import AxisEvent
from homeassistant.core import callback from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo, Entity from homeassistant.helpers.entity import DeviceInfo, Entity
from .const import DOMAIN as AXIS_DOMAIN from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
class AxisEntityBase(Entity): class AxisEntityBase(Entity):
@ -12,7 +14,7 @@ class AxisEntityBase(Entity):
_attr_has_entity_name = True _attr_has_entity_name = True
def __init__(self, device): def __init__(self, device: AxisNetworkDevice) -> None:
"""Initialize the Axis event.""" """Initialize the Axis event."""
self.device = device self.device = device
@ -20,7 +22,7 @@ class AxisEntityBase(Entity):
identifiers={(AXIS_DOMAIN, device.unique_id)} identifiers={(AXIS_DOMAIN, device.unique_id)}
) )
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Subscribe device events.""" """Subscribe device events."""
self.async_on_remove( self.async_on_remove(
async_dispatcher_connect( async_dispatcher_connect(
@ -29,12 +31,12 @@ class AxisEntityBase(Entity):
) )
@property @property
def available(self): def available(self) -> bool:
"""Return True if device is available.""" """Return True if device is available."""
return self.device.available return self.device.available
@callback @callback
def update_callback(self, no_delay=None): def update_callback(self, no_delay=None) -> None:
"""Update the entities state.""" """Update the entities state."""
self.async_write_ha_state() self.async_write_ha_state()
@ -44,7 +46,7 @@ class AxisEventBase(AxisEntityBase):
_attr_should_poll = False _attr_should_poll = False
def __init__(self, event, device): def __init__(self, event: AxisEvent, device: AxisNetworkDevice) -> None:
"""Initialize the Axis event.""" """Initialize the Axis event."""
super().__init__(device) super().__init__(device)
self.event = event self.event = event

View File

@ -1,4 +1,6 @@
"""Support for Axis binary sensors.""" """Support for Axis binary sensors."""
from __future__ import annotations
from datetime import timedelta from datetime import timedelta
from axis.event_stream import ( from axis.event_stream import (
@ -8,6 +10,8 @@ from axis.event_stream import (
CLASS_OUTPUT, CLASS_OUTPUT,
CLASS_PTZ, CLASS_PTZ,
CLASS_SOUND, CLASS_SOUND,
AxisBinaryEvent,
AxisEvent,
FenceGuard, FenceGuard,
LoiteringGuard, LoiteringGuard,
MotionGuard, MotionGuard,
@ -28,6 +32,7 @@ from homeassistant.util.dt import utcnow
from .axis_base import AxisEventBase from .axis_base import AxisEventBase
from .const import DOMAIN as AXIS_DOMAIN from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
DEVICE_CLASS = { DEVICE_CLASS = {
CLASS_INPUT: BinarySensorDeviceClass.CONNECTIVITY, CLASS_INPUT: BinarySensorDeviceClass.CONNECTIVITY,
@ -43,12 +48,12 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up a Axis binary sensor.""" """Set up a Axis binary sensor."""
device = hass.data[AXIS_DOMAIN][config_entry.unique_id] device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.unique_id]
@callback @callback
def async_add_sensor(event_id): def async_add_sensor(event_id):
"""Add binary sensor from Axis device.""" """Add binary sensor from Axis device."""
event = device.api.event[event_id] event: AxisEvent = device.api.event[event_id]
if event.CLASS not in (CLASS_OUTPUT, CLASS_PTZ) and not ( if event.CLASS not in (CLASS_OUTPUT, CLASS_PTZ) and not (
event.CLASS == CLASS_LIGHT and event.TYPE == "Light" event.CLASS == CLASS_LIGHT and event.TYPE == "Light"
@ -63,7 +68,9 @@ async def async_setup_entry(
class AxisBinarySensor(AxisEventBase, BinarySensorEntity): class AxisBinarySensor(AxisEventBase, BinarySensorEntity):
"""Representation of a binary Axis event.""" """Representation of a binary Axis event."""
def __init__(self, event, device): event: AxisBinaryEvent
def __init__(self, event: AxisEvent, device: AxisNetworkDevice) -> None:
"""Initialize the Axis binary sensor.""" """Initialize the Axis binary sensor."""
super().__init__(event, device) super().__init__(event, device)
self.cancel_scheduled_update = None self.cancel_scheduled_update = None
@ -98,12 +105,12 @@ class AxisBinarySensor(AxisEventBase, BinarySensorEntity):
) )
@property @property
def is_on(self): def is_on(self) -> bool:
"""Return true if event is active.""" """Return true if event is active."""
return self.event.is_tripped return self.event.is_tripped
@property @property
def name(self): def name(self) -> str | None:
"""Return the name of the event.""" """Return the name of the event."""
if ( if (
self.event.CLASS == CLASS_INPUT self.event.CLASS == CLASS_INPUT

View File

@ -11,6 +11,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .axis_base import AxisEntityBase from .axis_base import AxisEntityBase
from .const import DEFAULT_STREAM_PROFILE, DEFAULT_VIDEO_SOURCE, DOMAIN as AXIS_DOMAIN from .const import DEFAULT_STREAM_PROFILE, DEFAULT_VIDEO_SOURCE, DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
async def async_setup_entry( async def async_setup_entry(
@ -21,7 +22,7 @@ async def async_setup_entry(
"""Set up the Axis camera video stream.""" """Set up the Axis camera video stream."""
filter_urllib3_logging() filter_urllib3_logging()
device = hass.data[AXIS_DOMAIN][config_entry.unique_id] device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.unique_id]
if not device.api.vapix.params.image_format: if not device.api.vapix.params.image_format:
return return
@ -34,7 +35,7 @@ class AxisCamera(AxisEntityBase, MjpegCamera):
_attr_supported_features = CameraEntityFeature.STREAM _attr_supported_features = CameraEntityFeature.STREAM
def __init__(self, device): def __init__(self, device: AxisNetworkDevice) -> None:
"""Initialize Axis Communications camera component.""" """Initialize Axis Communications camera component."""
AxisEntityBase.__init__(self, device) AxisEntityBase.__init__(self, device)
@ -49,7 +50,7 @@ class AxisCamera(AxisEntityBase, MjpegCamera):
self._attr_unique_id = f"{device.unique_id}-camera" self._attr_unique_id = f"{device.unique_id}-camera"
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Subscribe camera events.""" """Subscribe camera events."""
self.async_on_remove( self.async_on_remove(
async_dispatcher_connect( async_dispatcher_connect(

View File

@ -51,7 +51,9 @@ from .errors import AuthenticationRequired, CannotConnect
class AxisNetworkDevice: class AxisNetworkDevice:
"""Manages a Axis device.""" """Manages a Axis device."""
def __init__(self, hass, config_entry, api): def __init__(
self, hass: HomeAssistant, config_entry: ConfigEntry, api: axis.AxisDevice
) -> None:
"""Initialize the device.""" """Initialize the device."""
self.hass = hass self.hass = hass
self.config_entry = config_entry self.config_entry = config_entry
@ -167,11 +169,11 @@ class AxisNetworkDevice:
This is a static method because a class method (bound method), This is a static method because a class method (bound method),
can not be used with weak references. can not be used with weak references.
""" """
device = hass.data[AXIS_DOMAIN][entry.unique_id] device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][entry.unique_id]
device.api.config.host = device.host device.api.config.host = device.host
async_dispatcher_send(hass, device.signal_new_address) async_dispatcher_send(hass, device.signal_new_address)
async def async_update_device_registry(self): async def async_update_device_registry(self) -> None:
"""Update device registry.""" """Update device registry."""
device_registry = dr.async_get(self.hass) device_registry = dr.async_get(self.hass)
device_registry.async_get_or_create( device_registry.async_get_or_create(
@ -224,17 +226,17 @@ class AxisNetworkDevice:
async_when_setup(self.hass, MQTT_DOMAIN, self.async_use_mqtt) async_when_setup(self.hass, MQTT_DOMAIN, self.async_use_mqtt)
@callback @callback
def disconnect_from_stream(self): def disconnect_from_stream(self) -> None:
"""Stop stream.""" """Stop stream."""
if self.api.stream.state != STATE_STOPPED: if self.api.stream.state != STATE_STOPPED:
self.api.stream.connection_status_callback.clear() self.api.stream.connection_status_callback.clear()
self.api.stream.stop() self.api.stream.stop()
async def shutdown(self, event): async def shutdown(self, event) -> None:
"""Stop the event stream.""" """Stop the event stream."""
self.disconnect_from_stream() self.disconnect_from_stream()
async def async_reset(self): async def async_reset(self) -> bool:
"""Reset this device to default state.""" """Reset this device to default state."""
self.disconnect_from_stream() self.disconnect_from_stream()

View File

@ -9,6 +9,7 @@ from homeassistant.const import CONF_MAC, CONF_PASSWORD, CONF_UNIQUE_ID, CONF_US
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from .const import DOMAIN as AXIS_DOMAIN from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
REDACT_CONFIG = {CONF_MAC, CONF_PASSWORD, CONF_UNIQUE_ID, CONF_USERNAME} REDACT_CONFIG = {CONF_MAC, CONF_PASSWORD, CONF_UNIQUE_ID, CONF_USERNAME}
REDACT_BASIC_DEVICE_INFO = {"SerialNumber", "SocSerialNumber"} REDACT_BASIC_DEVICE_INFO = {"SerialNumber", "SocSerialNumber"}
@ -19,7 +20,7 @@ async def async_get_config_entry_diagnostics(
hass: HomeAssistant, config_entry: ConfigEntry hass: HomeAssistant, config_entry: ConfigEntry
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Return diagnostics for a config entry.""" """Return diagnostics for a config entry."""
device = hass.data[AXIS_DOMAIN][config_entry.unique_id] device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.unique_id]
diag: dict[str, Any] = {} diag: dict[str, Any] = {}
diag["config"] = async_redact_data(config_entry.as_dict(), REDACT_CONFIG) diag["config"] = async_redact_data(config_entry.as_dict(), REDACT_CONFIG)

View File

@ -1,5 +1,7 @@
"""Support for Axis lights.""" """Support for Axis lights."""
from axis.event_stream import CLASS_LIGHT from typing import Any
from axis.event_stream import CLASS_LIGHT, AxisBinaryEvent, AxisEvent
from homeassistant.components.light import ATTR_BRIGHTNESS, ColorMode, LightEntity from homeassistant.components.light import ATTR_BRIGHTNESS, ColorMode, LightEntity
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
@ -9,6 +11,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .axis_base import AxisEventBase from .axis_base import AxisEventBase
from .const import DOMAIN as AXIS_DOMAIN from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
async def async_setup_entry( async def async_setup_entry(
@ -17,7 +20,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up a Axis light.""" """Set up a Axis light."""
device = hass.data[AXIS_DOMAIN][config_entry.unique_id] device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.unique_id]
if ( if (
device.api.vapix.light_control is None device.api.vapix.light_control is None
@ -28,7 +31,7 @@ async def async_setup_entry(
@callback @callback
def async_add_sensor(event_id): def async_add_sensor(event_id):
"""Add light from Axis device.""" """Add light from Axis device."""
event = device.api.event[event_id] event: AxisEvent = device.api.event[event_id]
if event.CLASS == CLASS_LIGHT and event.TYPE == "Light": if event.CLASS == CLASS_LIGHT and event.TYPE == "Light":
async_add_entities([AxisLight(event, device)]) async_add_entities([AxisLight(event, device)])
@ -42,8 +45,9 @@ class AxisLight(AxisEventBase, LightEntity):
"""Representation of a light Axis event.""" """Representation of a light Axis event."""
_attr_should_poll = True _attr_should_poll = True
event: AxisBinaryEvent
def __init__(self, event, device): def __init__(self, event: AxisEvent, device: AxisNetworkDevice) -> None:
"""Initialize the Axis light.""" """Initialize the Axis light."""
super().__init__(event, device) super().__init__(event, device)
@ -75,16 +79,16 @@ class AxisLight(AxisEventBase, LightEntity):
self.max_intensity = max_intensity["data"]["ranges"][0]["high"] self.max_intensity = max_intensity["data"]["ranges"][0]["high"]
@property @property
def is_on(self): def is_on(self) -> bool:
"""Return true if light is on.""" """Return true if light is on."""
return self.event.is_tripped return self.event.is_tripped
@property @property
def brightness(self): def brightness(self) -> int:
"""Return the brightness of this light between 0..255.""" """Return the brightness of this light between 0..255."""
return int((self.current_intensity / self.max_intensity) * 255) return int((self.current_intensity / self.max_intensity) * 255)
async def async_turn_on(self, **kwargs): async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on light.""" """Turn on light."""
if not self.is_on: if not self.is_on:
await self.device.api.vapix.light_control.activate_light(self.light_id) await self.device.api.vapix.light_control.activate_light(self.light_id)
@ -95,12 +99,12 @@ class AxisLight(AxisEventBase, LightEntity):
self.light_id, intensity self.light_id, intensity
) )
async def async_turn_off(self, **kwargs): async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off light.""" """Turn off light."""
if self.is_on: if self.is_on:
await self.device.api.vapix.light_control.deactivate_light(self.light_id) await self.device.api.vapix.light_control.deactivate_light(self.light_id)
async def async_update(self): async def async_update(self) -> None:
"""Update brightness.""" """Update brightness."""
current_intensity = ( current_intensity = (
await self.device.api.vapix.light_control.get_current_intensity( await self.device.api.vapix.light_control.get_current_intensity(

View File

@ -1,5 +1,7 @@
"""Support for Axis switches.""" """Support for Axis switches."""
from axis.event_stream import CLASS_OUTPUT from typing import Any
from axis.event_stream import CLASS_OUTPUT, AxisBinaryEvent, AxisEvent
from homeassistant.components.switch import SwitchEntity from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
@ -9,6 +11,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .axis_base import AxisEventBase from .axis_base import AxisEventBase
from .const import DOMAIN as AXIS_DOMAIN from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
async def async_setup_entry( async def async_setup_entry(
@ -17,12 +20,12 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up a Axis switch.""" """Set up a Axis switch."""
device = hass.data[AXIS_DOMAIN][config_entry.unique_id] device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.unique_id]
@callback @callback
def async_add_switch(event_id): def async_add_switch(event_id):
"""Add switch from Axis device.""" """Add switch from Axis device."""
event = device.api.event[event_id] event: AxisEvent = device.api.event[event_id]
if event.CLASS == CLASS_OUTPUT: if event.CLASS == CLASS_OUTPUT:
async_add_entities([AxisSwitch(event, device)]) async_add_entities([AxisSwitch(event, device)])
@ -35,7 +38,9 @@ async def async_setup_entry(
class AxisSwitch(AxisEventBase, SwitchEntity): class AxisSwitch(AxisEventBase, SwitchEntity):
"""Representation of a Axis switch.""" """Representation of a Axis switch."""
def __init__(self, event, device): event: AxisBinaryEvent
def __init__(self, event: AxisEvent, device: AxisNetworkDevice) -> None:
"""Initialize the Axis switch.""" """Initialize the Axis switch."""
super().__init__(event, device) super().__init__(event, device)
@ -43,14 +48,14 @@ class AxisSwitch(AxisEventBase, SwitchEntity):
self._attr_name = device.api.vapix.ports[event.id].name self._attr_name = device.api.vapix.ports[event.id].name
@property @property
def is_on(self): def is_on(self) -> bool:
"""Return true if event is active.""" """Return true if event is active."""
return self.event.is_tripped return self.event.is_tripped
async def async_turn_on(self, **kwargs): async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on switch.""" """Turn on switch."""
await self.device.api.vapix.ports[self.event.id].close() await self.device.api.vapix.ports[self.event.id].close()
async def async_turn_off(self, **kwargs): async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off switch.""" """Turn off switch."""
await self.device.api.vapix.ports[self.event.id].open() await self.device.api.vapix.ports[self.event.id].open()