jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-25 09:47:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bump axis to v46 (#85431)

Browse Source
This commit is contained in:
Robert Svensson 2023-01-09 12:43:40 +01:00 committed by GitHub
parent 54168c9bdb
commit 1cdd535f21
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 111 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
homeassistant/components/axis/axis_base.py
View File

@ -1,5 +1,6 @@
"""Base classes for Axis entities."""
from axis.event_stream import AxisEvent
from axis.models.event import Event, EventTopic
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
@ -8,6 +9,25 @@ from homeassistant.helpers.entity import DeviceInfo, Entity
from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
TOPIC_TO_EVENT_TYPE = {
EventTopic.DAY_NIGHT_VISION: "DayNight",
EventTopic.FENCE_GUARD: "Fence Guard",
EventTopic.LIGHT_STATUS: "Light",
EventTopic.LOITERING_GUARD: "Loitering Guard",
EventTopic.MOTION_DETECTION: "Motion",
EventTopic.MOTION_DETECTION_3: "VMD3",
EventTopic.MOTION_DETECTION_4: "VMD4",
EventTopic.MOTION_GUARD: "Motion Guard",
EventTopic.OBJECT_ANALYTICS: "Object Analytics",
EventTopic.PIR: "PIR",
EventTopic.PORT_INPUT: "Input",
EventTopic.PORT_SUPERVISED_INPUT: "Supervised Input",
EventTopic.PTZ_IS_MOVING: "is_moving",
EventTopic.PTZ_ON_PRESET: "on_preset",
EventTopic.RELAY: "Relay",
EventTopic.SOUND_TRIGGER_LEVEL: "Sound",
}
class AxisEntityBase(Entity):
"""Base common to all Axis entities."""
@ -46,21 +66,30 @@ class AxisEventBase(AxisEntityBase):
_attr_should_poll = False
def __init__(self, event: AxisEvent, device: AxisNetworkDevice) -> None:
def __init__(self, event: Event, device: AxisNetworkDevice) -> None:
"""Initialize the Axis event."""
super().__init__(device)
self.event = event
self._attr_name = f"{event.type} {event.id}"
self.event_type = TOPIC_TO_EVENT_TYPE[event.topic_base]
self._attr_name = f"{self.event_type} {event.id}"
self._attr_unique_id = f"{device.unique_id}-{event.topic}-{event.id}"
self._attr_device_class = event.group
self._attr_device_class = event.group.value
@callback
def async_event_callback(self, event) -> None:
"""Update the entities state."""
self.event = event
self.update_callback()
async def async_added_to_hass(self) -> None:
"""Subscribe sensors events."""
self.event.register_callback(self.update_callback)
await super().async_added_to_hass()
async def async_will_remove_from_hass(self) -> None:
"""Disconnect device object when removed."""
self.event.remove_callback(self.update_callback)
self.async_on_remove(
self.device.api.event.subscribe(
self.async_event_callback,
id_filter=self.event.id,
topic_filter=self.event.topic_base,
)
)

82
homeassistant/components/axis/binary_sensor.py
View File

@ -3,21 +3,7 @@ from __future__ import annotations
from datetime import timedelta
from axis.event_stream import (
CLASS_INPUT,
CLASS_LIGHT,
CLASS_MOTION,
CLASS_OUTPUT,
CLASS_PTZ,
CLASS_SOUND,
AxisBinaryEvent,
AxisEvent,
FenceGuard,
LoiteringGuard,
MotionGuard,
ObjectAnalytics,
Vmd4,
)
from axis.models.event import Event, EventGroup, EventOperation, EventTopic
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@ -25,7 +11,6 @@ from homeassistant.components.binary_sensor import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_point_in_utc_time
from homeassistant.util.dt import utcnow
@ -35,12 +20,27 @@ from .const import DOMAIN as AXIS_DOMAIN
from .device import AxisNetworkDevice
DEVICE_CLASS = {
CLASS_INPUT: BinarySensorDeviceClass.CONNECTIVITY,
CLASS_LIGHT: BinarySensorDeviceClass.LIGHT,
CLASS_MOTION: BinarySensorDeviceClass.MOTION,
CLASS_SOUND: BinarySensorDeviceClass.SOUND,
EventGroup.INPUT: BinarySensorDeviceClass.CONNECTIVITY,
EventGroup.LIGHT: BinarySensorDeviceClass.LIGHT,
EventGroup.MOTION: BinarySensorDeviceClass.MOTION,
EventGroup.SOUND: BinarySensorDeviceClass.SOUND,
}
EVENT_TOPICS = (
EventTopic.DAY_NIGHT_VISION,
EventTopic.FENCE_GUARD,
EventTopic.LOITERING_GUARD,
EventTopic.MOTION_DETECTION,
EventTopic.MOTION_DETECTION_3,
EventTopic.MOTION_DETECTION_4,
EventTopic.MOTION_GUARD,
EventTopic.OBJECT_ANALYTICS,
EventTopic.PIR,
EventTopic.PORT_INPUT,
EventTopic.PORT_SUPERVISED_INPUT,
EventTopic.SOUND_TRIGGER_LEVEL,
)
async def async_setup_entry(
hass: HomeAssistant,
@ -51,26 +51,21 @@ async def async_setup_entry(
device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.unique_id]
@callback
def async_add_sensor(event_id):
"""Add binary sensor from Axis device."""
event: AxisEvent = device.api.event[event_id]
def async_create_entity(event: Event) -> None:
"""Create Axis binary sensor entity."""
async_add_entities([AxisBinarySensor(event, device)])
if event.group not in (CLASS_OUTPUT, CLASS_PTZ) and not (
event.group == CLASS_LIGHT and event.type == "Light"
):
async_add_entities([AxisBinarySensor(event, device)])
config_entry.async_on_unload(
async_dispatcher_connect(hass, device.signal_new_event, async_add_sensor)
device.api.event.subscribe(
async_create_entity,
topic_filter=EVENT_TOPICS,
operation_filter=EventOperation.INITIALIZED,
)
class AxisBinarySensor(AxisEventBase, BinarySensorEntity):
"""Representation of a binary Axis event."""
event: AxisBinaryEvent
def __init__(self, event: AxisEvent, device: AxisNetworkDevice) -> None:
def __init__(self, event: Event, device: AxisNetworkDevice) -> None:
"""Initialize the Axis binary sensor."""
super().__init__(event, device)
self.cancel_scheduled_update = None
@ -110,26 +105,27 @@ class AxisBinarySensor(AxisEventBase, BinarySensorEntity):
def name(self) -> str | None:
"""Return the name of the event."""
if (
self.event.group == CLASS_INPUT
self.event.group == EventGroup.INPUT
and self.event.id in self.device.api.vapix.ports
and self.device.api.vapix.ports[self.event.id].name
):
return self.device.api.vapix.ports[self.event.id].name
if self.event.group == CLASS_MOTION:
if self.event.group == EventGroup.MOTION:
for event_class, event_data in (
(FenceGuard, self.device.api.vapix.fence_guard),
(LoiteringGuard, self.device.api.vapix.loitering_guard),
(MotionGuard, self.device.api.vapix.motion_guard),
(ObjectAnalytics, self.device.api.vapix.object_analytics),
(Vmd4, self.device.api.vapix.vmd4),
for event_topic, event_data in (
(EventTopic.FENCE_GUARD, self.device.api.vapix.fence_guard),
(EventTopic.LOITERING_GUARD, self.device.api.vapix.loitering_guard),
(EventTopic.MOTION_GUARD, self.device.api.vapix.motion_guard),
(EventTopic.OBJECT_ANALYTICS, self.device.api.vapix.object_analytics),
(EventTopic.MOTION_DETECTION_4, self.device.api.vapix.vmd4),
):
if (
isinstance(self.event, event_class)
self.event.topic_base == event_topic
and event_data
and self.event.id in event_data
):
return f"{self.event.type} {event_data[self.event.id].name}"
return f"{self.event_type} {event_data[self.event.id].name}"
return self._attr_name

22
homeassistant/components/axis/device.py
View File

@ -8,8 +8,7 @@ import async_timeout
import axis
from axis.configuration import Configuration
from axis.errors import Unauthorized
from axis.event_stream import OPERATION_INITIALIZED
from axis.streammanager import SIGNAL_PLAYING, STATE_STOPPED
from axis.stream_manager import Signal, State
from axis.vapix.interfaces.mqtt import mqtt_json_to_event
from homeassistant.components import mqtt
@ -129,11 +128,6 @@ class AxisNetworkDevice:
"""Device specific event to signal a change in connection status."""
return f"axis_reachable_{self.unique_id}"
@property
def signal_new_event(self):
"""Device specific event to signal new device event available."""
return f"axis_new_event_{self.unique_id}"
@property
def signal_new_address(self):
"""Device specific event to signal a change in device address."""
@ -149,16 +143,10 @@ class AxisNetworkDevice:
Only signal state change if state change is true.
"""
if self.available != (status == SIGNAL_PLAYING):
if self.available != (status == Signal.PLAYING):
self.available = not self.available
async_dispatcher_send(self.hass, self.signal_reachable, True)
@callback
def async_event_callback(self, action, event_id):
"""Call to configure events when initialized on event stream."""
if action == OPERATION_INITIALIZED:
async_dispatcher_send(self.hass, self.signal_new_event, event_id)
@staticmethod
async def async_new_address_callback(
hass: HomeAssistant, entry: ConfigEntry
@ -208,7 +196,7 @@ class AxisNetworkDevice:
self.disconnect_from_stream()
event = mqtt_json_to_event(message.payload)
self.api.event.update([event])
self.api.event.handler(event)
# Setup and teardown methods
@ -219,7 +207,7 @@ class AxisNetworkDevice:
self.api.stream.connection_status_callback.append(
self.async_connection_status_callback
)
self.api.enable_events(event_callback=self.async_event_callback)
self.api.enable_events()
self.api.stream.start()
if self.api.vapix.mqtt:
@ -228,7 +216,7 @@ class AxisNetworkDevice:
@callback
def disconnect_from_stream(self) -> None:
"""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.stop()

23
homeassistant/components/axis/light.py
View File

@ -1,12 +1,11 @@
"""Support for Axis lights."""
from typing import Any
from axis.event_stream import CLASS_LIGHT, AxisBinaryEvent, AxisEvent
from axis.models.event import Event, EventOperation, EventTopic
from homeassistant.components.light import ATTR_BRIGHTNESS, ColorMode, LightEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .axis_base import AxisEventBase
@ -29,15 +28,14 @@ async def async_setup_entry(
return
@callback
def async_add_sensor(event_id):
"""Add light from Axis device."""
event: AxisEvent = device.api.event[event_id]
def async_create_entity(event: Event) -> None:
"""Create Axis light entity."""
async_add_entities([AxisLight(event, device)])
if event.group == CLASS_LIGHT and event.type == "Light":
async_add_entities([AxisLight(event, device)])
config_entry.async_on_unload(
async_dispatcher_connect(hass, device.signal_new_event, async_add_sensor)
device.api.event.subscribe(
async_create_entity,
topic_filter=EventTopic.LIGHT_STATUS,
operation_filter=EventOperation.INITIALIZED,
)
@ -45,9 +43,8 @@ class AxisLight(AxisEventBase, LightEntity):
"""Representation of a light Axis event."""
_attr_should_poll = True
event: AxisBinaryEvent
def __init__(self, event: AxisEvent, device: AxisNetworkDevice) -> None:
def __init__(self, event: Event, device: AxisNetworkDevice) -> None:
"""Initialize the Axis light."""
super().__init__(event, device)
@ -57,7 +54,7 @@ class AxisLight(AxisEventBase, LightEntity):
self.max_intensity = 0
light_type = device.api.vapix.light_control[self.light_id].light_type
self._attr_name = f"{light_type} {event.type} {event.id}"
self._attr_name = f"{light_type} {self.event_type} {event.id}"
self._attr_supported_color_modes = {ColorMode.BRIGHTNESS}
self._attr_color_mode = ColorMode.BRIGHTNESS

2
homeassistant/components/axis/manifest.json
View File

@ -3,7 +3,7 @@
"name": "Axis",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/axis",
"requirements": ["axis==45"],
"requirements": ["axis==46"],
"dhcp": [
{
"registered_devices": true

22
homeassistant/components/axis/switch.py
View File

@ -1,12 +1,11 @@
"""Support for Axis switches."""
from typing import Any
from axis.event_stream import CLASS_OUTPUT, AxisBinaryEvent, AxisEvent
from axis.models.event import Event, EventOperation, EventTopic
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .axis_base import AxisEventBase
@ -23,24 +22,21 @@ async def async_setup_entry(
device: AxisNetworkDevice = hass.data[AXIS_DOMAIN][config_entry.unique_id]
@callback
def async_add_switch(event_id):
"""Add switch from Axis device."""
event: AxisEvent = device.api.event[event_id]
def async_create_entity(event: Event) -> None:
"""Create Axis switch entity."""
async_add_entities([AxisSwitch(event, device)])
if event.group == CLASS_OUTPUT:
async_add_entities([AxisSwitch(event, device)])
config_entry.async_on_unload(
async_dispatcher_connect(hass, device.signal_new_event, async_add_switch)
device.api.event.subscribe(
async_create_entity,
topic_filter=EventTopic.RELAY,
operation_filter=EventOperation.INITIALIZED,
)
class AxisSwitch(AxisEventBase, SwitchEntity):
"""Representation of a Axis switch."""
event: AxisBinaryEvent
def __init__(self, event: AxisEvent, device: AxisNetworkDevice) -> None:
def __init__(self, event: Event, device: AxisNetworkDevice) -> None:
"""Initialize the Axis switch."""
super().__init__(event, device)

2
requirements_all.txt
View File

@ -395,7 +395,7 @@ aurorapy==0.2.7
# avion==0.10
# homeassistant.components.axis
axis==45
axis==46
# homeassistant.components.azure_event_hub
azure-eventhub==5.7.0

2
requirements_test_all.txt
View File

@ -340,7 +340,7 @@ auroranoaa==0.0.2
aurorapy==0.2.7
# homeassistant.components.axis
axis==45
axis==46
# homeassistant.components.azure_event_hub
azure-eventhub==5.7.0

20
tests/components/axis/conftest.py
View File

@ -3,13 +3,7 @@ from __future__ import annotations
from unittest.mock import patch
from axis.rtsp import (
SIGNAL_DATA,
SIGNAL_FAILED,
SIGNAL_PLAYING,
STATE_PLAYING,
STATE_STOPPED,
)
from axis.rtsp import Signal, State
import pytest
from tests.components.light.conftest import mock_light_profiles # noqa: F401
@ -18,19 +12,19 @@ from tests.components.light.conftest import mock_light_profiles # noqa: F401
@pytest.fixture(autouse=True)
def mock_axis_rtspclient():
"""No real RTSP communication allowed."""
with patch("axis.streammanager.RTSPClient") as rtsp_client_mock:
with patch("axis.stream_manager.RTSPClient") as rtsp_client_mock:
rtsp_client_mock.return_value.session.state = STATE_STOPPED
rtsp_client_mock.return_value.session.state = State.STOPPED
async def start_stream():
"""Set state to playing when calling RTSPClient.start."""
rtsp_client_mock.return_value.session.state = STATE_PLAYING
rtsp_client_mock.return_value.session.state = State.PLAYING
rtsp_client_mock.return_value.start = start_stream
def stop_stream():
"""Set state to stopped when calling RTSPClient.stop."""
rtsp_client_mock.return_value.session.state = STATE_STOPPED
rtsp_client_mock.return_value.session.state = State.STOPPED
rtsp_client_mock.return_value.stop = stop_stream
@ -40,7 +34,7 @@ def mock_axis_rtspclient():
if data:
rtsp_client_mock.return_value.rtp.data = data
axis_streammanager_session_callback(signal=SIGNAL_DATA)
axis_streammanager_session_callback(signal=Signal.DATA)
elif state:
axis_streammanager_session_callback(signal=state)
else:
@ -106,7 +100,7 @@ def mock_rtsp_signal_state(mock_axis_rtspclient):
def send_signal(connected: bool) -> None:
"""Signal state change of RTSP connection."""
signal = SIGNAL_PLAYING if connected else SIGNAL_FAILED
signal = Signal.PLAYING if connected else Signal.FAILED
mock_axis_rtspclient(state=signal)
yield send_signal

16
tests/components/axis/test_device.py
View File

@ -4,7 +4,6 @@ from unittest import mock
from unittest.mock import Mock, patch
import axis as axislib
from axis.event_stream import OPERATION_INITIALIZED
import pytest
import respx
@ -463,21 +462,6 @@ async def test_device_unknown_error(hass):
assert hass.data[AXIS_DOMAIN] == {}
async def test_new_event_sends_signal(hass):
"""Make sure that new event send signal."""
entry = Mock()
entry.data = ENTRY_CONFIG
axis_device = axis.device.AxisNetworkDevice(hass, entry, Mock())
with patch.object(axis.device, "async_dispatcher_send") as mock_dispatch_send:
axis_device.async_event_callback(action=OPERATION_INITIALIZED, event_id="event")
await hass.async_block_till_done()
assert len(mock_dispatch_send.mock_calls) == 1
assert len(mock_dispatch_send.mock_calls[0]) == 3
async def test_shutdown():
"""Successful shutdown."""
hass = Mock()