event_source.py

Go to the documentation of this file.

"""Axis network device abstraction."""
 
from __future__ import annotations
 
import axis
from axis.errors import Unauthorized
from axis.interfaces.mqtt import mqtt_json_to_event
from axis.models.mqtt import ClientState
from axis.stream_manager import Signal, State
 
from homeassistant.components import mqtt
from homeassistant.components.mqtt import DOMAIN as MQTT_DOMAIN, ReceiveMessage
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.setup import async_when_setup
 
 
class AxisEventSource:
    """Manage connection to event sources from an Axis device."""
 
    def __init__(
        self, hass: HomeAssistant, config_entry: ConfigEntry, api: axis.AxisDevice
    ) -> None:
        """Initialize the device."""
        self.hasshass = hass
        self.config_entryconfig_entry = config_entry
        self.apiapi = api
 
        self.signal_reachablesignal_reachable = f"axis_reachable_{config_entry.entry_id}"
 
        self.availableavailable = True
 
    @callback
    def setup(self) -> None:
        """Set up the device events."""
        self.apiapi.stream.connection_status_callback.append(self._connection_status_cb_connection_status_cb)
        self.apiapi.enable_events()
        self.apiapi.stream.start()
 
        if self.apiapi.vapix.mqtt.supported:
            async_when_setup(self.hasshass, MQTT_DOMAIN, self._async_use_mqtt_async_use_mqtt)
 
    @callback
    def teardown(self) -> None:
        """Tear down connections."""
        self._disconnect_from_stream_disconnect_from_stream()
 
    @callback
    def _disconnect_from_stream(self) -> None:
        """Stop stream."""
        if self.apiapi.stream.state != State.STOPPED:
            self.apiapi.stream.connection_status_callback.clear()
        self.apiapi.stream.stop()
 
    async def _async_use_mqtt(self, hass: HomeAssistant, component: str) -> None:
        """Set up to use MQTT."""
        try:
            status = await self.apiapi.vapix.mqtt.get_client_status()
        except Unauthorized:
            # This means the user has too low privileges
            return
 
        if status.status.state == ClientState.ACTIVE:
            self.config_entryconfig_entry.async_on_unload(
                await mqtt.async_subscribe(
                    hass, f"{status.config.device_topic_prefix}/#", self._mqtt_message_mqtt_message
                )
            )
 
    @callback
    def _mqtt_message(self, message: ReceiveMessage) -> None:
        """Receive Axis MQTT message."""
        self._disconnect_from_stream_disconnect_from_stream()
 
        if message.topic.endswith("event/connection"):
            return
 
        event = mqtt_json_to_event(message.payload)
        self.apiapi.event.handler(event)
 
    @callback
    def _connection_status_cb(self, status: Signal) -> None:
        """Handle signals of device connection status.
 
        This is called on every RTSP keep-alive message.
        Only signal state change if state change is true.
        """
 
        if self.availableavailable != (status == Signal.PLAYING):
            self.availableavailable = not self.availableavailable
            async_dispatcher_send(self.hasshass, self.signal_reachablesignal_reachable)