From e43802eb0769fb8850f9533de873d94b260bc1e2 Mon Sep 17 00:00:00 2001 From: Robert Svensson Date: Wed, 18 Jan 2023 17:27:13 +0100 Subject: [PATCH] Use more _attrs_* in Axis entities (#85555) * Use _attr_available * Use _attr_is_on * Use _attr_name * Make some values private * Update names of axis entity base classes * Fix review comments --- .../components/axis/binary_sensor.py | 47 +++++++++---------- homeassistant/components/axis/camera.py | 6 +-- homeassistant/components/axis/device.py | 2 +- .../axis/{axis_base.py => entity.py} | 37 ++++++++------- homeassistant/components/axis/light.py | 32 +++++++------ homeassistant/components/axis/switch.py | 18 +++---- tests/components/axis/test_switch.py | 13 +++++ 7 files changed, 86 insertions(+), 69 deletions(-) rename homeassistant/components/axis/{axis_base.py => entity.py} (75%) diff --git a/homeassistant/components/axis/binary_sensor.py b/homeassistant/components/axis/binary_sensor.py index 8c3957e4d19..729d69ed45b 100644 --- a/homeassistant/components/axis/binary_sensor.py +++ b/homeassistant/components/axis/binary_sensor.py @@ -1,6 +1,7 @@ """Support for Axis binary sensors.""" from __future__ import annotations +from collections.abc import Callable from datetime import timedelta from axis.models.event import Event, EventGroup, EventOperation, EventTopic @@ -15,9 +16,9 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.event import async_track_point_in_utc_time from homeassistant.util.dt import utcnow -from .axis_base import AxisEventBase from .const import DOMAIN as AXIS_DOMAIN from .device import AxisNetworkDevice +from .entity import AxisEventEntity DEVICE_CLASS = { EventGroup.INPUT: BinarySensorDeviceClass.CONNECTIVITY, @@ -62,24 +63,23 @@ async def async_setup_entry( ) -class AxisBinarySensor(AxisEventBase, BinarySensorEntity): +class AxisBinarySensor(AxisEventEntity, BinarySensorEntity): """Representation of a binary Axis event.""" def __init__(self, event: Event, device: AxisNetworkDevice) -> None: """Initialize the Axis binary sensor.""" super().__init__(event, device) - self.cancel_scheduled_update = None + self.cancel_scheduled_update: Callable[[], None] | None = None - self._attr_device_class = DEVICE_CLASS.get(self.event.group) + self._attr_device_class = DEVICE_CLASS.get(event.group) self._attr_is_on = event.is_tripped - @callback - def update_callback(self, no_delay=False): - """Update the sensor's state, if needed. + self._set_name(event) - Parameter no_delay is True when device_event_reachable is sent. - """ - self._attr_is_on = self.event.is_tripped + @callback + def async_event_callback(self, event: Event) -> None: + """Update the sensor's state, if needed.""" + self._attr_is_on = event.is_tripped @callback def scheduled_update(now): @@ -91,7 +91,7 @@ class AxisBinarySensor(AxisEventBase, BinarySensorEntity): self.cancel_scheduled_update() self.cancel_scheduled_update = None - if self.is_on or self.device.option_trigger_time == 0 or no_delay: + if self.is_on or self.device.option_trigger_time == 0: self.async_write_ha_state() return @@ -101,17 +101,17 @@ class AxisBinarySensor(AxisEventBase, BinarySensorEntity): utcnow() + timedelta(seconds=self.device.option_trigger_time), ) - @property - def name(self) -> str | None: - """Return the name of the event.""" + @callback + def _set_name(self, event: Event) -> None: + """Set binary sensor name.""" if ( - 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 + event.group == EventGroup.INPUT + and event.id in self.device.api.vapix.ports + and self.device.api.vapix.ports[event.id].name ): - return self.device.api.vapix.ports[self.event.id].name + self._attr_name = self.device.api.vapix.ports[event.id].name - if self.event.group == EventGroup.MOTION: + elif event.group == EventGroup.MOTION: for event_topic, event_data in ( (EventTopic.FENCE_GUARD, self.device.api.vapix.fence_guard), @@ -122,10 +122,9 @@ class AxisBinarySensor(AxisEventBase, BinarySensorEntity): ): if ( - self.event.topic_base == event_topic + event.topic_base == event_topic and event_data - and self.event.id in event_data + and event.id in event_data ): - return f"{self.event_type} {event_data[self.event.id].name}" - - return self._attr_name + self._attr_name = f"{self._event_type} {event_data[event.id].name}" + break diff --git a/homeassistant/components/axis/camera.py b/homeassistant/components/axis/camera.py index b45cfc1ecc2..c593c4fa419 100644 --- a/homeassistant/components/axis/camera.py +++ b/homeassistant/components/axis/camera.py @@ -9,9 +9,9 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .axis_base import AxisEntityBase from .const import DEFAULT_STREAM_PROFILE, DEFAULT_VIDEO_SOURCE, DOMAIN as AXIS_DOMAIN from .device import AxisNetworkDevice +from .entity import AxisEntity async def async_setup_entry( @@ -30,14 +30,14 @@ async def async_setup_entry( async_add_entities([AxisCamera(device)]) -class AxisCamera(AxisEntityBase, MjpegCamera): +class AxisCamera(AxisEntity, MjpegCamera): """Representation of a Axis camera.""" _attr_supported_features = CameraEntityFeature.STREAM def __init__(self, device: AxisNetworkDevice) -> None: """Initialize Axis Communications camera component.""" - AxisEntityBase.__init__(self, device) + AxisEntity.__init__(self, device) MjpegCamera.__init__( self, diff --git a/homeassistant/components/axis/device.py b/homeassistant/components/axis/device.py index 77901f03fc2..2dce4b7692a 100644 --- a/homeassistant/components/axis/device.py +++ b/homeassistant/components/axis/device.py @@ -145,7 +145,7 @@ class AxisNetworkDevice: if self.available != (status == Signal.PLAYING): self.available = not self.available - async_dispatcher_send(self.hass, self.signal_reachable, True) + async_dispatcher_send(self.hass, self.signal_reachable) @staticmethod async def async_new_address_callback( diff --git a/homeassistant/components/axis/axis_base.py b/homeassistant/components/axis/entity.py similarity index 75% rename from homeassistant/components/axis/axis_base.py rename to homeassistant/components/axis/entity.py index fe8a11d1e68..e511ee72d1b 100644 --- a/homeassistant/components/axis/axis_base.py +++ b/homeassistant/components/axis/entity.py @@ -1,5 +1,7 @@ """Base classes for Axis entities.""" +from abc import abstractmethod + from axis.models.event import Event, EventTopic from homeassistant.core import callback @@ -29,7 +31,7 @@ TOPIC_TO_EVENT_TYPE = { } -class AxisEntityBase(Entity): +class AxisEntity(Entity): """Base common to all Axis entities.""" _attr_has_entity_name = True @@ -46,22 +48,20 @@ class AxisEntityBase(Entity): """Subscribe device events.""" self.async_on_remove( async_dispatcher_connect( - self.hass, self.device.signal_reachable, self.update_callback + self.hass, + self.device.signal_reachable, + self.async_signal_reachable_callback, ) ) - @property - def available(self) -> bool: - """Return True if device is available.""" - return self.device.available - @callback - def update_callback(self, no_delay=None) -> None: - """Update the entities state.""" + def async_signal_reachable_callback(self) -> None: + """Call when device connection state change.""" + self._attr_available = self.device.available self.async_write_ha_state() -class AxisEventBase(AxisEntityBase): +class AxisEventEntity(AxisEntity): """Base common to all Axis entities from event stream.""" _attr_should_poll = False @@ -69,19 +69,20 @@ class AxisEventBase(AxisEntityBase): def __init__(self, event: Event, device: AxisNetworkDevice) -> None: """Initialize the Axis event.""" super().__init__(device) - self.event = event - self.event_type = TOPIC_TO_EVENT_TYPE[event.topic_base] - self._attr_name = f"{self.event_type} {event.id}" + self._event_id = event.id + self._event_topic = event.topic_base + 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.value @callback - def async_event_callback(self, event) -> None: + @abstractmethod + def async_event_callback(self, event: Event) -> None: """Update the entities state.""" - self.event = event - self.update_callback() async def async_added_to_hass(self) -> None: """Subscribe sensors events.""" @@ -89,7 +90,7 @@ class AxisEventBase(AxisEntityBase): self.async_on_remove( self.device.api.event.subscribe( self.async_event_callback, - id_filter=self.event.id, - topic_filter=self.event.topic_base, + id_filter=self._event_id, + topic_filter=self._event_topic, ) ) diff --git a/homeassistant/components/axis/light.py b/homeassistant/components/axis/light.py index 24566b71974..10dc8258d7e 100644 --- a/homeassistant/components/axis/light.py +++ b/homeassistant/components/axis/light.py @@ -8,9 +8,9 @@ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .axis_base import AxisEventBase from .const import DOMAIN as AXIS_DOMAIN from .device import AxisNetworkDevice +from .entity import AxisEventEntity async def async_setup_entry( @@ -39,7 +39,7 @@ async def async_setup_entry( ) -class AxisLight(AxisEventBase, LightEntity): +class AxisLight(AxisEventEntity, LightEntity): """Representation of a light Axis event.""" _attr_should_poll = True @@ -48,13 +48,14 @@ class AxisLight(AxisEventBase, LightEntity): """Initialize the Axis light.""" super().__init__(event, device) - self.light_id = f"led{self.event.id}" + self._light_id = f"led{event.id}" self.current_intensity = 0 self.max_intensity = 0 - light_type = device.api.vapix.light_control[self.light_id].light_type - self._attr_name = f"{light_type} {self.event_type} {event.id}" + light_type = device.api.vapix.light_control[self._light_id].light_type + self._attr_name = f"{light_type} {self._event_type} {event.id}" + self._attr_is_on = event.is_tripped self._attr_supported_color_modes = {ColorMode.BRIGHTNESS} self._attr_color_mode = ColorMode.BRIGHTNESS @@ -65,20 +66,21 @@ class AxisLight(AxisEventBase, LightEntity): current_intensity = ( await self.device.api.vapix.light_control.get_current_intensity( - self.light_id + self._light_id ) ) self.current_intensity = current_intensity["data"]["intensity"] max_intensity = await self.device.api.vapix.light_control.get_valid_intensity( - self.light_id + self._light_id ) self.max_intensity = max_intensity["data"]["ranges"][0]["high"] - @property - def is_on(self) -> bool: - """Return true if light is on.""" - return self.event.is_tripped + @callback + def async_event_callback(self, event: Event) -> None: + """Update light state.""" + self._attr_is_on = event.is_tripped + self.async_write_ha_state() @property def brightness(self) -> int: @@ -88,24 +90,24 @@ class AxisLight(AxisEventBase, LightEntity): async def async_turn_on(self, **kwargs: Any) -> None: """Turn on light.""" 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) if ATTR_BRIGHTNESS in kwargs: intensity = int((kwargs[ATTR_BRIGHTNESS] / 255) * self.max_intensity) await self.device.api.vapix.light_control.set_manual_intensity( - self.light_id, intensity + self._light_id, intensity ) async def async_turn_off(self, **kwargs: Any) -> None: """Turn off light.""" 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) -> None: """Update brightness.""" current_intensity = ( await self.device.api.vapix.light_control.get_current_intensity( - self.light_id + self._light_id ) ) self.current_intensity = current_intensity["data"]["intensity"] diff --git a/homeassistant/components/axis/switch.py b/homeassistant/components/axis/switch.py index 05ff8375415..adcd1ba5525 100644 --- a/homeassistant/components/axis/switch.py +++ b/homeassistant/components/axis/switch.py @@ -8,9 +8,9 @@ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .axis_base import AxisEventBase from .const import DOMAIN as AXIS_DOMAIN from .device import AxisNetworkDevice +from .entity import AxisEventEntity async def async_setup_entry( @@ -33,7 +33,7 @@ async def async_setup_entry( ) -class AxisSwitch(AxisEventBase, SwitchEntity): +class AxisSwitch(AxisEventEntity, SwitchEntity): """Representation of a Axis switch.""" def __init__(self, event: Event, device: AxisNetworkDevice) -> None: @@ -42,16 +42,18 @@ class AxisSwitch(AxisEventBase, SwitchEntity): if event.id and device.api.vapix.ports[event.id].name: self._attr_name = device.api.vapix.ports[event.id].name + self._attr_is_on = event.is_tripped - @property - def is_on(self) -> bool: - """Return true if event is active.""" - return self.event.is_tripped + @callback + def async_event_callback(self, event: Event) -> None: + """Update light state.""" + self._attr_is_on = event.is_tripped + self.async_write_ha_state() async def async_turn_on(self, **kwargs: Any) -> None: """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: Any) -> None: """Turn off switch.""" - await self.device.api.vapix.ports[self.event.id].open() + await self.device.api.vapix.ports[self._event_id].open() diff --git a/tests/components/axis/test_switch.py b/tests/components/axis/test_switch.py index 9a334eae2ce..40ecb02a68f 100644 --- a/tests/components/axis/test_switch.py +++ b/tests/components/axis/test_switch.py @@ -137,6 +137,19 @@ async def test_switches_with_port_management(hass, config_entry, mock_rtsp_event assert relay_0.state == STATE_OFF assert relay_0.name == f"{NAME} Doorbell" + # State update + + mock_rtsp_event( + topic="tns1:Device/Trigger/Relay", + data_type="LogicalState", + data_value="active", + source_name="RelayToken", + source_idx="0", + ) + await hass.async_block_till_done() + + assert hass.states.get(f"{SWITCH_DOMAIN}.{NAME}_relay_1").state == STATE_ON + await hass.services.async_call( SWITCH_DOMAIN, SERVICE_TURN_ON,