mirror of https://github.com/home-assistant/core
757 lines
26 KiB
Python
757 lines
26 KiB
Python
"""Extend the basic Accessory and Bridge functions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, cast
|
|
from uuid import UUID
|
|
|
|
from pyhap.accessory import Accessory, Bridge
|
|
from pyhap.accessory_driver import AccessoryDriver
|
|
from pyhap.characteristic import Characteristic
|
|
from pyhap.const import CATEGORY_OTHER
|
|
from pyhap.iid_manager import IIDManager
|
|
from pyhap.service import Service
|
|
from pyhap.util import callback as pyhap_callback
|
|
|
|
from homeassistant.components.cover import CoverDeviceClass, CoverEntityFeature
|
|
from homeassistant.components.media_player import MediaPlayerDeviceClass
|
|
from homeassistant.components.remote import RemoteEntityFeature
|
|
from homeassistant.components.sensor import SensorDeviceClass
|
|
from homeassistant.components.switch import SwitchDeviceClass
|
|
from homeassistant.const import (
|
|
ATTR_BATTERY_CHARGING,
|
|
ATTR_BATTERY_LEVEL,
|
|
ATTR_DEVICE_CLASS,
|
|
ATTR_ENTITY_ID,
|
|
ATTR_HW_VERSION,
|
|
ATTR_MANUFACTURER,
|
|
ATTR_MODEL,
|
|
ATTR_SERVICE,
|
|
ATTR_SUPPORTED_FEATURES,
|
|
ATTR_SW_VERSION,
|
|
ATTR_UNIT_OF_MEASUREMENT,
|
|
CONF_NAME,
|
|
CONF_TYPE,
|
|
LIGHT_LUX,
|
|
PERCENTAGE,
|
|
STATE_ON,
|
|
STATE_UNAVAILABLE,
|
|
STATE_UNKNOWN,
|
|
UnitOfTemperature,
|
|
__version__,
|
|
)
|
|
from homeassistant.core import (
|
|
CALLBACK_TYPE,
|
|
Context,
|
|
Event,
|
|
EventStateChangedData,
|
|
HassJobType,
|
|
HomeAssistant,
|
|
State,
|
|
callback as ha_callback,
|
|
split_entity_id,
|
|
)
|
|
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
|
from homeassistant.helpers.event import async_track_state_change_event
|
|
from homeassistant.util.decorator import Registry
|
|
|
|
from .const import (
|
|
ATTR_DISPLAY_NAME,
|
|
ATTR_INTEGRATION,
|
|
ATTR_VALUE,
|
|
BRIDGE_MODEL,
|
|
BRIDGE_SERIAL_NUMBER,
|
|
CHAR_BATTERY_LEVEL,
|
|
CHAR_CHARGING_STATE,
|
|
CHAR_HARDWARE_REVISION,
|
|
CHAR_STATUS_LOW_BATTERY,
|
|
CONF_FEATURE_LIST,
|
|
CONF_LINKED_BATTERY_CHARGING_SENSOR,
|
|
CONF_LINKED_BATTERY_SENSOR,
|
|
CONF_LOW_BATTERY_THRESHOLD,
|
|
DEFAULT_LOW_BATTERY_THRESHOLD,
|
|
EMPTY_MAC,
|
|
EVENT_HOMEKIT_CHANGED,
|
|
HK_CHARGING,
|
|
HK_NOT_CHARGABLE,
|
|
HK_NOT_CHARGING,
|
|
MANUFACTURER,
|
|
MAX_MANUFACTURER_LENGTH,
|
|
MAX_MODEL_LENGTH,
|
|
MAX_SERIAL_LENGTH,
|
|
MAX_VERSION_LENGTH,
|
|
SERV_ACCESSORY_INFO,
|
|
SERV_BATTERY_SERVICE,
|
|
SIGNAL_RELOAD_ENTITIES,
|
|
TYPE_FAUCET,
|
|
TYPE_OUTLET,
|
|
TYPE_SHOWER,
|
|
TYPE_SPRINKLER,
|
|
TYPE_SWITCH,
|
|
TYPE_VALVE,
|
|
)
|
|
from .iidmanager import AccessoryIIDStorage
|
|
from .util import (
|
|
accessory_friendly_name,
|
|
async_dismiss_setup_message,
|
|
async_show_setup_message,
|
|
cleanup_name_for_homekit,
|
|
convert_to_float,
|
|
format_version,
|
|
validate_media_player_features,
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
SWITCH_TYPES = {
|
|
TYPE_FAUCET: "ValveSwitch",
|
|
TYPE_OUTLET: "Outlet",
|
|
TYPE_SHOWER: "ValveSwitch",
|
|
TYPE_SPRINKLER: "ValveSwitch",
|
|
TYPE_SWITCH: "Switch",
|
|
TYPE_VALVE: "ValveSwitch",
|
|
}
|
|
TYPES: Registry[str, type[HomeAccessory]] = Registry()
|
|
|
|
RELOAD_ON_CHANGE_ATTRS = (
|
|
ATTR_SUPPORTED_FEATURES,
|
|
ATTR_DEVICE_CLASS,
|
|
ATTR_UNIT_OF_MEASUREMENT,
|
|
)
|
|
|
|
|
|
def get_accessory( # noqa: C901
|
|
hass: HomeAssistant, driver: HomeDriver, state: State, aid: int | None, config: dict
|
|
) -> HomeAccessory | None:
|
|
"""Take state and return an accessory object if supported."""
|
|
if not aid:
|
|
_LOGGER.warning(
|
|
(
|
|
'The entity "%s" is not supported, since it '
|
|
"generates an invalid aid, please change it"
|
|
),
|
|
state.entity_id,
|
|
)
|
|
return None
|
|
|
|
a_type = None
|
|
name = config.get(CONF_NAME, state.name)
|
|
features = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
|
|
|
|
if state.domain == "alarm_control_panel":
|
|
a_type = "SecuritySystem"
|
|
|
|
elif state.domain in ("binary_sensor", "device_tracker", "person"):
|
|
a_type = "BinarySensor"
|
|
|
|
elif state.domain == "climate":
|
|
a_type = "Thermostat"
|
|
|
|
elif state.domain == "cover":
|
|
device_class = state.attributes.get(ATTR_DEVICE_CLASS)
|
|
|
|
if device_class in (
|
|
CoverDeviceClass.GARAGE,
|
|
CoverDeviceClass.GATE,
|
|
) and features & (CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE):
|
|
a_type = "GarageDoorOpener"
|
|
elif (
|
|
device_class == CoverDeviceClass.WINDOW
|
|
and features & CoverEntityFeature.SET_POSITION
|
|
):
|
|
a_type = "Window"
|
|
elif (
|
|
device_class == CoverDeviceClass.DOOR
|
|
and features & CoverEntityFeature.SET_POSITION
|
|
):
|
|
a_type = "Door"
|
|
elif features & CoverEntityFeature.SET_POSITION:
|
|
a_type = "WindowCovering"
|
|
elif features & (CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE):
|
|
a_type = "WindowCoveringBasic"
|
|
elif features & CoverEntityFeature.SET_TILT_POSITION:
|
|
# WindowCovering and WindowCoveringBasic both support tilt
|
|
# only WindowCovering can handle the covers that are missing
|
|
# CoverEntityFeature.SET_POSITION, CoverEntityFeature.OPEN,
|
|
# and CoverEntityFeature.CLOSE
|
|
a_type = "WindowCovering"
|
|
|
|
elif state.domain == "fan":
|
|
a_type = "Fan"
|
|
|
|
elif state.domain == "humidifier":
|
|
a_type = "HumidifierDehumidifier"
|
|
|
|
elif state.domain == "light":
|
|
a_type = "Light"
|
|
|
|
elif state.domain == "lock":
|
|
a_type = "Lock"
|
|
|
|
elif state.domain == "media_player":
|
|
device_class = state.attributes.get(ATTR_DEVICE_CLASS)
|
|
feature_list = config.get(CONF_FEATURE_LIST, [])
|
|
|
|
if device_class == MediaPlayerDeviceClass.RECEIVER:
|
|
a_type = "ReceiverMediaPlayer"
|
|
elif device_class == MediaPlayerDeviceClass.TV:
|
|
a_type = "TelevisionMediaPlayer"
|
|
elif validate_media_player_features(state, feature_list):
|
|
a_type = "MediaPlayer"
|
|
|
|
elif state.domain == "sensor":
|
|
device_class = state.attributes.get(ATTR_DEVICE_CLASS)
|
|
unit = state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
|
|
|
|
if device_class == SensorDeviceClass.TEMPERATURE or unit in (
|
|
UnitOfTemperature.CELSIUS,
|
|
UnitOfTemperature.FAHRENHEIT,
|
|
):
|
|
a_type = "TemperatureSensor"
|
|
elif device_class == SensorDeviceClass.HUMIDITY and unit == PERCENTAGE:
|
|
a_type = "HumiditySensor"
|
|
elif (
|
|
device_class == SensorDeviceClass.PM10
|
|
or SensorDeviceClass.PM10 in state.entity_id
|
|
):
|
|
a_type = "PM10Sensor"
|
|
elif (
|
|
device_class == SensorDeviceClass.PM25
|
|
or SensorDeviceClass.PM25 in state.entity_id
|
|
):
|
|
a_type = "PM25Sensor"
|
|
elif device_class == SensorDeviceClass.NITROGEN_DIOXIDE:
|
|
a_type = "NitrogenDioxideSensor"
|
|
elif device_class == SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS:
|
|
a_type = "VolatileOrganicCompoundsSensor"
|
|
elif (
|
|
device_class == SensorDeviceClass.GAS
|
|
or SensorDeviceClass.GAS in state.entity_id
|
|
):
|
|
a_type = "AirQualitySensor"
|
|
elif device_class == SensorDeviceClass.CO:
|
|
a_type = "CarbonMonoxideSensor"
|
|
elif device_class == SensorDeviceClass.CO2 or "co2" in state.entity_id:
|
|
a_type = "CarbonDioxideSensor"
|
|
elif device_class == SensorDeviceClass.ILLUMINANCE or unit == LIGHT_LUX:
|
|
a_type = "LightSensor"
|
|
|
|
elif state.domain == "switch":
|
|
if switch_type := config.get(CONF_TYPE):
|
|
a_type = SWITCH_TYPES[switch_type]
|
|
elif state.attributes.get(ATTR_DEVICE_CLASS) == SwitchDeviceClass.OUTLET:
|
|
a_type = "Outlet"
|
|
else:
|
|
a_type = "Switch"
|
|
|
|
elif state.domain == "valve":
|
|
a_type = "Valve"
|
|
|
|
elif state.domain == "vacuum":
|
|
a_type = "Vacuum"
|
|
|
|
elif state.domain == "remote" and features & RemoteEntityFeature.ACTIVITY:
|
|
a_type = "ActivityRemote"
|
|
|
|
elif state.domain in (
|
|
"automation",
|
|
"button",
|
|
"input_boolean",
|
|
"input_button",
|
|
"remote",
|
|
"scene",
|
|
"script",
|
|
):
|
|
a_type = "Switch"
|
|
|
|
elif state.domain in ("input_select", "select"):
|
|
a_type = "SelectSwitch"
|
|
|
|
elif state.domain == "water_heater":
|
|
a_type = "WaterHeater"
|
|
|
|
elif state.domain == "camera":
|
|
a_type = "Camera"
|
|
|
|
if a_type is None:
|
|
return None
|
|
|
|
_LOGGER.debug('Add "%s" as "%s"', state.entity_id, a_type)
|
|
return TYPES[a_type](hass, driver, name, state.entity_id, aid, config)
|
|
|
|
|
|
class HomeAccessory(Accessory): # type: ignore[misc]
|
|
"""Adapter class for Accessory."""
|
|
|
|
driver: HomeDriver
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
driver: HomeDriver,
|
|
name: str,
|
|
entity_id: str,
|
|
aid: int,
|
|
config: dict[str, Any],
|
|
*args: Any,
|
|
category: int = CATEGORY_OTHER,
|
|
device_id: str | None = None,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
"""Initialize a Accessory object."""
|
|
super().__init__(
|
|
driver=driver,
|
|
display_name=cleanup_name_for_homekit(name),
|
|
aid=aid,
|
|
iid_manager=HomeIIDManager(driver.iid_storage),
|
|
*args, # noqa: B026
|
|
**kwargs,
|
|
)
|
|
self._reload_on_change_attrs = list(RELOAD_ON_CHANGE_ATTRS)
|
|
self.config = config or {}
|
|
if device_id:
|
|
self.device_id: str | None = device_id
|
|
serial_number = device_id
|
|
domain = None
|
|
else:
|
|
self.device_id = None
|
|
serial_number = entity_id
|
|
domain = split_entity_id(entity_id)[0].replace("_", " ")
|
|
|
|
if self.config.get(ATTR_MANUFACTURER) is not None:
|
|
manufacturer = str(self.config[ATTR_MANUFACTURER])
|
|
elif self.config.get(ATTR_INTEGRATION) is not None:
|
|
manufacturer = self.config[ATTR_INTEGRATION].replace("_", " ").title()
|
|
elif domain:
|
|
manufacturer = f"{MANUFACTURER} {domain}".title()
|
|
else:
|
|
manufacturer = MANUFACTURER
|
|
if self.config.get(ATTR_MODEL) is not None:
|
|
model = str(self.config[ATTR_MODEL])
|
|
elif domain:
|
|
model = domain.title()
|
|
else:
|
|
model = MANUFACTURER
|
|
sw_version = None
|
|
if self.config.get(ATTR_SW_VERSION) is not None:
|
|
sw_version = format_version(self.config[ATTR_SW_VERSION])
|
|
if sw_version is None:
|
|
sw_version = format_version(__version__)
|
|
assert sw_version is not None
|
|
hw_version = None
|
|
if self.config.get(ATTR_HW_VERSION) is not None:
|
|
hw_version = format_version(self.config[ATTR_HW_VERSION])
|
|
|
|
self.set_info_service(
|
|
manufacturer=manufacturer[:MAX_MANUFACTURER_LENGTH],
|
|
model=model[:MAX_MODEL_LENGTH],
|
|
serial_number=serial_number[:MAX_SERIAL_LENGTH],
|
|
firmware_revision=sw_version[:MAX_VERSION_LENGTH],
|
|
)
|
|
if hw_version:
|
|
serv_info = self.get_service(SERV_ACCESSORY_INFO)
|
|
char = self.driver.loader.get_char(CHAR_HARDWARE_REVISION)
|
|
serv_info.add_characteristic(char)
|
|
serv_info.configure_char(
|
|
CHAR_HARDWARE_REVISION, value=hw_version[:MAX_VERSION_LENGTH]
|
|
)
|
|
char.broker = self
|
|
self.iid_manager.assign(char)
|
|
|
|
self.category = category
|
|
self.entity_id = entity_id
|
|
self.hass = hass
|
|
self._subscriptions: list[CALLBACK_TYPE] = []
|
|
|
|
if device_id:
|
|
return
|
|
|
|
self._char_battery = None
|
|
self._char_charging = None
|
|
self._char_low_battery = None
|
|
self.linked_battery_sensor = self.config.get(CONF_LINKED_BATTERY_SENSOR)
|
|
self.linked_battery_charging_sensor = self.config.get(
|
|
CONF_LINKED_BATTERY_CHARGING_SENSOR
|
|
)
|
|
self.low_battery_threshold = self.config.get(
|
|
CONF_LOW_BATTERY_THRESHOLD, DEFAULT_LOW_BATTERY_THRESHOLD
|
|
)
|
|
|
|
"""Add battery service if available"""
|
|
state = self.hass.states.get(self.entity_id)
|
|
self._update_available_from_state(state)
|
|
assert state is not None
|
|
entity_attributes = state.attributes
|
|
battery_found = entity_attributes.get(ATTR_BATTERY_LEVEL)
|
|
|
|
if self.linked_battery_sensor:
|
|
state = self.hass.states.get(self.linked_battery_sensor)
|
|
if state is not None:
|
|
battery_found = state.state
|
|
else:
|
|
_LOGGER.warning(
|
|
"%s: Battery sensor state missing: %s",
|
|
self.entity_id,
|
|
self.linked_battery_sensor,
|
|
)
|
|
self.linked_battery_sensor = None
|
|
|
|
if not battery_found:
|
|
return
|
|
|
|
_LOGGER.debug("%s: Found battery level", self.entity_id)
|
|
|
|
if self.linked_battery_charging_sensor:
|
|
state = self.hass.states.get(self.linked_battery_charging_sensor)
|
|
if state is None:
|
|
self.linked_battery_charging_sensor = None
|
|
_LOGGER.warning(
|
|
"%s: Battery charging binary_sensor state missing: %s",
|
|
self.entity_id,
|
|
self.linked_battery_charging_sensor,
|
|
)
|
|
else:
|
|
_LOGGER.debug("%s: Found battery charging", self.entity_id)
|
|
|
|
serv_battery = self.add_preload_service(SERV_BATTERY_SERVICE)
|
|
self._char_battery = serv_battery.configure_char(CHAR_BATTERY_LEVEL, value=0)
|
|
self._char_charging = serv_battery.configure_char(
|
|
CHAR_CHARGING_STATE, value=HK_NOT_CHARGABLE
|
|
)
|
|
self._char_low_battery = serv_battery.configure_char(
|
|
CHAR_STATUS_LOW_BATTERY, value=0
|
|
)
|
|
|
|
def _update_available_from_state(self, new_state: State | None) -> None:
|
|
"""Update the available property based on the state."""
|
|
self._available = new_state is not None and new_state.state != STATE_UNAVAILABLE
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
"""Return if accessory is available."""
|
|
return self._available
|
|
|
|
@ha_callback
|
|
@pyhap_callback # type: ignore[misc]
|
|
def run(self) -> None:
|
|
"""Handle accessory driver started event."""
|
|
if state := self.hass.states.get(self.entity_id):
|
|
self.async_update_state_callback(state)
|
|
self._update_available_from_state(state)
|
|
self._subscriptions.append(
|
|
async_track_state_change_event(
|
|
self.hass,
|
|
[self.entity_id],
|
|
self.async_update_event_state_callback,
|
|
job_type=HassJobType.Callback,
|
|
)
|
|
)
|
|
|
|
battery_charging_state = None
|
|
battery_state = None
|
|
if self.linked_battery_sensor and (
|
|
linked_battery_sensor_state := self.hass.states.get(
|
|
self.linked_battery_sensor
|
|
)
|
|
):
|
|
battery_state = linked_battery_sensor_state.state
|
|
battery_charging_state = linked_battery_sensor_state.attributes.get(
|
|
ATTR_BATTERY_CHARGING
|
|
)
|
|
self._subscriptions.append(
|
|
async_track_state_change_event(
|
|
self.hass,
|
|
[self.linked_battery_sensor],
|
|
self.async_update_linked_battery_callback,
|
|
job_type=HassJobType.Callback,
|
|
)
|
|
)
|
|
elif state is not None:
|
|
battery_state = state.attributes.get(ATTR_BATTERY_LEVEL)
|
|
if self.linked_battery_charging_sensor:
|
|
state = self.hass.states.get(self.linked_battery_charging_sensor)
|
|
battery_charging_state = state and state.state == STATE_ON
|
|
self._subscriptions.append(
|
|
async_track_state_change_event(
|
|
self.hass,
|
|
[self.linked_battery_charging_sensor],
|
|
self.async_update_linked_battery_charging_callback,
|
|
job_type=HassJobType.Callback,
|
|
)
|
|
)
|
|
elif battery_charging_state is None and state is not None:
|
|
battery_charging_state = state.attributes.get(ATTR_BATTERY_CHARGING)
|
|
|
|
if battery_state is not None or battery_charging_state is not None:
|
|
self.async_update_battery(battery_state, battery_charging_state)
|
|
|
|
@ha_callback
|
|
def async_update_event_state_callback(
|
|
self, event: Event[EventStateChangedData]
|
|
) -> None:
|
|
"""Handle state change event listener callback."""
|
|
new_state = event.data["new_state"]
|
|
old_state = event.data["old_state"]
|
|
self._update_available_from_state(new_state)
|
|
if (
|
|
new_state
|
|
and old_state
|
|
and STATE_UNAVAILABLE not in (old_state.state, new_state.state)
|
|
):
|
|
old_attributes = old_state.attributes
|
|
new_attributes = new_state.attributes
|
|
for attr in self._reload_on_change_attrs:
|
|
if old_attributes.get(attr) != new_attributes.get(attr):
|
|
_LOGGER.debug(
|
|
"%s: Reloading HomeKit accessory since %s has changed from %s -> %s",
|
|
self.entity_id,
|
|
attr,
|
|
old_attributes.get(attr),
|
|
new_attributes.get(attr),
|
|
)
|
|
self.async_reload()
|
|
return
|
|
self.async_update_state_callback(new_state)
|
|
|
|
@ha_callback
|
|
def async_update_state_callback(self, new_state: State | None) -> None:
|
|
"""Handle state change listener callback."""
|
|
_LOGGER.debug("New_state: %s", new_state)
|
|
# HomeKit handles unavailable state via the available property
|
|
# so we should not propagate it here
|
|
if new_state is None or new_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
|
|
return
|
|
battery_state = None
|
|
battery_charging_state = None
|
|
if (
|
|
not self.linked_battery_sensor
|
|
and ATTR_BATTERY_LEVEL in new_state.attributes
|
|
):
|
|
battery_state = new_state.attributes.get(ATTR_BATTERY_LEVEL)
|
|
if (
|
|
not self.linked_battery_charging_sensor
|
|
and ATTR_BATTERY_CHARGING in new_state.attributes
|
|
):
|
|
battery_charging_state = new_state.attributes.get(ATTR_BATTERY_CHARGING)
|
|
if battery_state is not None or battery_charging_state is not None:
|
|
self.async_update_battery(battery_state, battery_charging_state)
|
|
self.async_update_state(new_state)
|
|
|
|
@ha_callback
|
|
def async_update_linked_battery_callback(
|
|
self, event: Event[EventStateChangedData]
|
|
) -> None:
|
|
"""Handle linked battery sensor state change listener callback."""
|
|
if (new_state := event.data["new_state"]) is None:
|
|
return
|
|
if self.linked_battery_charging_sensor:
|
|
battery_charging_state = None
|
|
else:
|
|
battery_charging_state = new_state.attributes.get(ATTR_BATTERY_CHARGING)
|
|
self.async_update_battery(new_state.state, battery_charging_state)
|
|
|
|
@ha_callback
|
|
def async_update_linked_battery_charging_callback(
|
|
self, event: Event[EventStateChangedData]
|
|
) -> None:
|
|
"""Handle linked battery charging sensor state change listener callback."""
|
|
if (new_state := event.data["new_state"]) is None:
|
|
return
|
|
self.async_update_battery(None, new_state.state == STATE_ON)
|
|
|
|
@ha_callback
|
|
def async_update_battery(self, battery_level: Any, battery_charging: Any) -> None:
|
|
"""Update battery service if available.
|
|
|
|
Only call this function if self._support_battery_level is True.
|
|
"""
|
|
if not self._char_battery or not self._char_low_battery:
|
|
# Battery appeared after homekit was started
|
|
return
|
|
|
|
battery_level = convert_to_float(battery_level)
|
|
if battery_level is not None:
|
|
if self._char_battery.value != battery_level:
|
|
self._char_battery.set_value(battery_level)
|
|
is_low_battery = 1 if battery_level < self.low_battery_threshold else 0
|
|
if self._char_low_battery.value != is_low_battery:
|
|
self._char_low_battery.set_value(is_low_battery)
|
|
_LOGGER.debug(
|
|
"%s: Updated battery level to %d", self.entity_id, battery_level
|
|
)
|
|
|
|
# Charging state can appear after homekit was started
|
|
if battery_charging is None or not self._char_charging:
|
|
return
|
|
|
|
hk_charging = HK_CHARGING if battery_charging else HK_NOT_CHARGING
|
|
if self._char_charging.value != hk_charging:
|
|
self._char_charging.set_value(hk_charging)
|
|
_LOGGER.debug(
|
|
"%s: Updated battery charging to %d", self.entity_id, hk_charging
|
|
)
|
|
|
|
@ha_callback
|
|
def async_update_state(self, new_state: State) -> None:
|
|
"""Handle state change to update HomeKit value.
|
|
|
|
Overridden by accessory types.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@ha_callback
|
|
def async_call_service(
|
|
self,
|
|
domain: str,
|
|
service: str,
|
|
service_data: dict[str, Any] | None,
|
|
value: Any | None = None,
|
|
) -> None:
|
|
"""Fire event and call service for changes from HomeKit."""
|
|
event_data = {
|
|
ATTR_ENTITY_ID: self.entity_id,
|
|
ATTR_DISPLAY_NAME: self.display_name,
|
|
ATTR_SERVICE: service,
|
|
ATTR_VALUE: value,
|
|
}
|
|
context = Context()
|
|
|
|
self.hass.bus.async_fire(EVENT_HOMEKIT_CHANGED, event_data, context=context)
|
|
self.hass.async_create_task(
|
|
self.hass.services.async_call(
|
|
domain, service, service_data, context=context
|
|
),
|
|
eager_start=True,
|
|
)
|
|
|
|
@ha_callback
|
|
def async_reload(self) -> None:
|
|
"""Reload and recreate an accessory and update the c# value in the mDNS record."""
|
|
async_dispatcher_send(
|
|
self.hass,
|
|
SIGNAL_RELOAD_ENTITIES.format(self.driver.entry_id),
|
|
(self.entity_id,),
|
|
)
|
|
|
|
@ha_callback
|
|
def async_stop(self) -> None:
|
|
"""Cancel any subscriptions when the bridge is stopped."""
|
|
while self._subscriptions:
|
|
self._subscriptions.pop(0)()
|
|
|
|
async def stop(self) -> None:
|
|
"""Stop the accessory.
|
|
|
|
This is overrides the parent class to call async_stop
|
|
since pyhap will call this function to stop the accessory
|
|
but we want to use our async_stop method since we need
|
|
it to be a callback to avoid races in reloading accessories.
|
|
"""
|
|
self.async_stop()
|
|
|
|
|
|
class HomeBridge(Bridge): # type: ignore[misc]
|
|
"""Adapter class for Bridge."""
|
|
|
|
def __init__(self, hass: HomeAssistant, driver: HomeDriver, name: str) -> None:
|
|
"""Initialize a Bridge object."""
|
|
super().__init__(driver, name, iid_manager=HomeIIDManager(driver.iid_storage))
|
|
self.set_info_service(
|
|
firmware_revision=format_version(__version__),
|
|
manufacturer=MANUFACTURER,
|
|
model=BRIDGE_MODEL,
|
|
serial_number=BRIDGE_SERIAL_NUMBER,
|
|
)
|
|
self.hass = hass
|
|
|
|
def setup_message(self) -> None:
|
|
"""Prevent print of pyhap setup message to terminal."""
|
|
|
|
async def async_get_snapshot(self, info: dict) -> bytes:
|
|
"""Get snapshot from accessory if supported."""
|
|
if (acc := self.accessories.get(info["aid"])) is None:
|
|
raise ValueError("Requested snapshot for missing accessory")
|
|
if not hasattr(acc, "async_get_snapshot"):
|
|
raise ValueError(
|
|
"Got a request for snapshot, but the Accessory "
|
|
'does not define a "async_get_snapshot" method'
|
|
)
|
|
return cast(bytes, await acc.async_get_snapshot(info))
|
|
|
|
|
|
class HomeDriver(AccessoryDriver): # type: ignore[misc]
|
|
"""Adapter class for AccessoryDriver."""
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
entry_id: str,
|
|
bridge_name: str,
|
|
entry_title: str,
|
|
iid_storage: AccessoryIIDStorage,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
"""Initialize a AccessoryDriver object."""
|
|
# Always set an empty mac of pyhap will incur
|
|
# the cost of generating a new one for every driver
|
|
super().__init__(**kwargs, mac=EMPTY_MAC)
|
|
self.hass = hass
|
|
self.entry_id = entry_id
|
|
self._bridge_name = bridge_name
|
|
self._entry_title = entry_title
|
|
self.iid_storage = iid_storage
|
|
|
|
@pyhap_callback # type: ignore[misc]
|
|
def pair(
|
|
self, client_username_bytes: bytes, client_public: str, client_permissions: int
|
|
) -> bool:
|
|
"""Override super function to dismiss setup message if paired."""
|
|
success = super().pair(client_username_bytes, client_public, client_permissions)
|
|
if success:
|
|
async_dismiss_setup_message(self.hass, self.entry_id)
|
|
return cast(bool, success)
|
|
|
|
@pyhap_callback # type: ignore[misc]
|
|
def unpair(self, client_uuid: UUID) -> None:
|
|
"""Override super function to show setup message if unpaired."""
|
|
super().unpair(client_uuid)
|
|
|
|
if self.state.paired:
|
|
return
|
|
|
|
async_show_setup_message(
|
|
self.hass,
|
|
self.entry_id,
|
|
accessory_friendly_name(self._entry_title, self.accessory),
|
|
self.state.pincode,
|
|
self.accessory.xhm_uri(),
|
|
)
|
|
|
|
|
|
class HomeIIDManager(IIDManager): # type: ignore[misc]
|
|
"""IID Manager that remembers IIDs between restarts."""
|
|
|
|
def __init__(self, iid_storage: AccessoryIIDStorage) -> None:
|
|
"""Initialize a IIDManager object."""
|
|
super().__init__()
|
|
self._iid_storage = iid_storage
|
|
|
|
def get_iid_for_obj(self, obj: Characteristic | Service) -> int:
|
|
"""Get IID for object."""
|
|
aid = obj.broker.aid
|
|
if isinstance(obj, Characteristic):
|
|
service: Service = obj.service
|
|
iid = self._iid_storage.get_or_allocate_iid(
|
|
aid, service.type_id, service.unique_id, obj.type_id, obj.unique_id
|
|
)
|
|
else:
|
|
iid = self._iid_storage.get_or_allocate_iid(
|
|
aid, obj.type_id, obj.unique_id, None, None
|
|
)
|
|
if iid in self.objs:
|
|
raise RuntimeError(
|
|
f"Cannot assign IID {iid} to {obj} as it is already in use by:"
|
|
f" {self.objs[iid]}"
|
|
)
|
|
return iid
|