122 lines
4.2 KiB
Python
122 lines
4.2 KiB
Python
"""Extend the doorbell functions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from pyhap.util import callback as pyhap_callback
|
|
|
|
from homeassistant.const import STATE_ON, STATE_UNAVAILABLE, STATE_UNKNOWN
|
|
from homeassistant.core import (
|
|
Event,
|
|
EventStateChangedData,
|
|
HassJobType,
|
|
State,
|
|
callback as ha_callback,
|
|
)
|
|
from homeassistant.helpers.event import async_track_state_change_event
|
|
|
|
from .accessories import HomeAccessory
|
|
from .const import (
|
|
CHAR_MUTE,
|
|
CHAR_PROGRAMMABLE_SWITCH_EVENT,
|
|
CONF_LINKED_DOORBELL_SENSOR,
|
|
SERV_DOORBELL,
|
|
SERV_SPEAKER,
|
|
SERV_STATELESS_PROGRAMMABLE_SWITCH,
|
|
)
|
|
from .util import state_changed_event_is_same_state
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
DOORBELL_SINGLE_PRESS = 0
|
|
DOORBELL_DOUBLE_PRESS = 1
|
|
DOORBELL_LONG_PRESS = 2
|
|
|
|
|
|
class HomeDoorbellAccessory(HomeAccessory):
|
|
"""Accessory with optional doorbell."""
|
|
|
|
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
|
"""Initialize an Accessory object with optional attached doorbell."""
|
|
super().__init__(*args, **kwargs)
|
|
self._char_doorbell_detected = None
|
|
self._char_doorbell_detected_switch = None
|
|
linked_doorbell_sensor: str | None
|
|
linked_doorbell_sensor = self.config.get(CONF_LINKED_DOORBELL_SENSOR)
|
|
self.linked_doorbell_sensor = linked_doorbell_sensor
|
|
self.doorbell_is_event = False
|
|
if not linked_doorbell_sensor:
|
|
return
|
|
self.doorbell_is_event = linked_doorbell_sensor.startswith("event.")
|
|
if not (state := self.hass.states.get(linked_doorbell_sensor)):
|
|
return
|
|
serv_doorbell = self.add_preload_service(SERV_DOORBELL)
|
|
self.set_primary_service(serv_doorbell)
|
|
self._char_doorbell_detected = serv_doorbell.configure_char(
|
|
CHAR_PROGRAMMABLE_SWITCH_EVENT,
|
|
value=0,
|
|
)
|
|
serv_stateless_switch = self.add_preload_service(
|
|
SERV_STATELESS_PROGRAMMABLE_SWITCH
|
|
)
|
|
self._char_doorbell_detected_switch = serv_stateless_switch.configure_char(
|
|
CHAR_PROGRAMMABLE_SWITCH_EVENT,
|
|
value=0,
|
|
valid_values={"SinglePress": DOORBELL_SINGLE_PRESS},
|
|
)
|
|
serv_speaker = self.add_preload_service(SERV_SPEAKER)
|
|
serv_speaker.configure_char(CHAR_MUTE, value=0)
|
|
self.async_update_doorbell_state(None, state)
|
|
|
|
@ha_callback
|
|
@pyhap_callback # type: ignore[misc]
|
|
def run(self) -> None:
|
|
"""Handle doorbell event."""
|
|
if self._char_doorbell_detected:
|
|
assert self.linked_doorbell_sensor
|
|
self._subscriptions.append(
|
|
async_track_state_change_event(
|
|
self.hass,
|
|
self.linked_doorbell_sensor,
|
|
self.async_update_doorbell_state_event,
|
|
job_type=HassJobType.Callback,
|
|
)
|
|
)
|
|
|
|
super().run()
|
|
|
|
@ha_callback
|
|
def async_update_doorbell_state_event(
|
|
self, event: Event[EventStateChangedData]
|
|
) -> None:
|
|
"""Handle state change event listener callback."""
|
|
if not state_changed_event_is_same_state(event) and (
|
|
new_state := event.data["new_state"]
|
|
):
|
|
self.async_update_doorbell_state(event.data["old_state"], new_state)
|
|
|
|
@ha_callback
|
|
def async_update_doorbell_state(
|
|
self, old_state: State | None, new_state: State
|
|
) -> None:
|
|
"""Handle link doorbell sensor state change to update HomeKit value."""
|
|
assert self._char_doorbell_detected
|
|
assert self._char_doorbell_detected_switch
|
|
state = new_state.state
|
|
if state == STATE_ON or (
|
|
self.doorbell_is_event
|
|
and old_state is not None
|
|
and old_state.state != STATE_UNAVAILABLE
|
|
and state not in (STATE_UNKNOWN, STATE_UNAVAILABLE)
|
|
):
|
|
self._char_doorbell_detected.set_value(DOORBELL_SINGLE_PRESS)
|
|
self._char_doorbell_detected_switch.set_value(DOORBELL_SINGLE_PRESS)
|
|
_LOGGER.debug(
|
|
"%s: Set linked doorbell %s sensor to %d",
|
|
self.entity_id,
|
|
self.linked_doorbell_sensor,
|
|
DOORBELL_SINGLE_PRESS,
|
|
)
|