"""Extend the basic Accessory and Bridge functions.""" 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.alarm_control_panel import ( DOMAIN as ALARM_CONTROL_PANEL_DOMAIN, ) from homeassistant.components.automation import DOMAIN as AUTOMATION_DOMAIN from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN from homeassistant.components.climate import ( DOMAIN as CLIMATE_DOMAIN, ClimateEntityFeature, ) from homeassistant.components.cover import ( DOMAIN as COVER_DOMAIN, CoverDeviceClass, CoverEntityFeature, ) from homeassistant.components.device_tracker import DOMAIN as DEVICE_TRACKER_DOMAIN from homeassistant.components.fan import DOMAIN as FAN_DOMAIN from homeassistant.components.humidifier import DOMAIN as HUMIDIFIER_DOMAIN from homeassistant.components.input_boolean import DOMAIN as INPUT_BOOLEAN_DOMAIN from homeassistant.components.input_button import DOMAIN as INPUT_BUTTON_DOMAIN from homeassistant.components.input_select import DOMAIN as INPUT_SELECT_DOMAIN from homeassistant.components.lawn_mower import ( DOMAIN as LAWN_MOWER_DOMAIN, LawnMowerEntityFeature, ) from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN from homeassistant.components.media_player import ( DOMAIN as MEDIA_PLAYER_DOMAIN, MediaPlayerDeviceClass, ) from homeassistant.components.person import DOMAIN as PERSON_DOMAIN from homeassistant.components.remote import DOMAIN as REMOTE_DOMAIN, RemoteEntityFeature from homeassistant.components.scene import DOMAIN as SCENE_DOMAIN from homeassistant.components.script import DOMAIN as SCRIPT_DOMAIN from homeassistant.components.select import DOMAIN as SELECT_DOMAIN from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN, SensorDeviceClass from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN, SwitchDeviceClass from homeassistant.components.vacuum import DOMAIN as VACUUM_DOMAIN from homeassistant.components.valve import DOMAIN as VALVE_DOMAIN from homeassistant.components.water_heater import DOMAIN as WATER_HEATER_DOMAIN from homeassistant.const import ( ATTR_BATTERY_CHARGING, ATTR_BATTERY_LEVEL, ATTR_ENTITY_ID, ATTR_HW_VERSION, ATTR_MANUFACTURER, ATTR_MODEL, ATTR_SERVICE, ATTR_SW_VERSION, CONF_NAME, CONF_TYPE, LIGHT_LUX, PERCENTAGE, STATE_ON, STATE_UNAVAILABLE, STATE_UNKNOWN, EntityStateAttribute, UnitOfTemperature, __version__, ) from homeassistant.core import ( CALLBACK_TYPE, Context, Event, EventStateChangedData, HassJobType, HomeAssistant, State, callback as ha_callback, split_entity_id, ) from homeassistant.exceptions import HomeAssistantError 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 .aidmanager import AccessoryAidStorage from .climate_util import ( get_fan_modes_and_speeds, get_swing_on_mode, has_swing_off_mode, ) 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_AIR_PURIFIER, TYPE_FAN, TYPE_FAUCET, TYPE_HEATER_COOLER, TYPE_OUTLET, TYPE_SHOWER, TYPE_SPRINKLER, TYPE_SWITCH, TYPE_THERMOSTAT, 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", } FAN_TYPES = { TYPE_AIR_PURIFIER: "AirPurifier", TYPE_FAN: "Fan", } CLIMATE_TYPES = { TYPE_HEATER_COOLER: "HeaterCooler", TYPE_THERMOSTAT: "Thermostat", } TYPES: Registry[str, type[HomeAccessory]] = Registry() RELOAD_ON_CHANGE_ATTRS = ( EntityStateAttribute.SUPPORTED_FEATURES, EntityStateAttribute.DEVICE_CLASS, EntityStateAttribute.UNIT_OF_MEASUREMENT, ) def climate_controls_target_humidity(state: State) -> bool: """Return True when a climate entity exposes a humidity setpoint. HeaterCooler cannot control a humidity setpoint; entities that expose one (e.g. econet) stay on the Thermostat, which can. """ features = state.attributes.get(EntityStateAttribute.SUPPORTED_FEATURES, 0) return bool(features & ClimateEntityFeature.TARGET_HUMIDITY) def climate_supports_heater_cooler(state: State) -> bool: """Return True when a climate entity fits the HeaterCooler accessory.""" attributes = state.attributes features = attributes.get(EntityStateAttribute.SUPPORTED_FEATURES, 0) # Timing fan modes like auto or circulate do not count as speeds. has_fan = bool(features & ClimateEntityFeature.FAN_MODE) and ( len(get_fan_modes_and_speeds(attributes)[1]) >= 2 ) # The binary swing control writes the off mode back, so automatic # routing requires the entity to advertise one. has_swing = bool(features & ClimateEntityFeature.SWING_MODE) and ( get_swing_on_mode(attributes) is not None and has_swing_off_mode(attributes) ) return (has_fan or has_swing) and not ( features & ClimateEntityFeature.TARGET_HUMIDITY ) @ha_callback def async_resolve_accessory_type( aid_storage: AccessoryAidStorage, state: State, conf: dict[str, Any], *, allow_auto: bool, ) -> str | None: """Resolve which accessory an entity uses into conf. Some domains can be represented by more than one HomeKit accessory; climate is the only such domain today. Returns the accessory type the caller must record with async_set_accessory_type once the accessory is successfully created, so a failed creation is not sticky across restarts; a stored routing the entity can no longer support is dropped immediately instead. """ if state.domain != CLIMATE_DOMAIN: return None return _async_resolve_climate_type(aid_storage, state, conf, allow_auto=allow_auto) @ha_callback def _async_resolve_climate_type( aid_storage: AccessoryAidStorage, state: State, conf: dict[str, Any], *, allow_auto: bool, ) -> str | None: """Resolve which accessory a climate entity uses into conf. An explicit type in the entity config always wins, even for entities with a humidity setpoint, and updates the stored routing, so switching back to automatic keeps the accessory the entity already uses. In bridge mode an entity that has never been bridged gets the HeaterCooler when capable. Anything else keeps the Thermostat; the accessory type can be changed at any time from the bridge options. """ entity_id = state.entity_id if climate_type := conf.get(CONF_TYPE): # The explicit type is recorded by the caller like the automatic # one, so every path that sets a type defers persistence until # the accessory exists. return cast(str, climate_type) stored_type = aid_storage.get_accessory_type(entity_id) if stored_type == TYPE_HEATER_COOLER: if not climate_controls_target_humidity(state): conf[CONF_TYPE] = TYPE_HEATER_COOLER return None # A humidity setpoint gained since the choice was stored cannot # be represented by the HeaterCooler, so the routing is dropped. aid_storage.async_set_accessory_type(entity_id, None) if not climate_supports_heater_cooler(state): return None if ( stored_type is None and allow_auto and not aid_storage.entity_is_allocated(entity_id) ): # A stored Thermostat choice survives even when the entity looks # new again, like an accessory mode pairing reset, so Automatic # keeps the accessory the entity already uses. conf[CONF_TYPE] = TYPE_HEATER_COOLER return TYPE_HEATER_COOLER return None 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(EntityStateAttribute.SUPPORTED_FEATURES, 0) if state.domain == ALARM_CONTROL_PANEL_DOMAIN: a_type = "SecuritySystem" elif state.domain in (BINARY_SENSOR_DOMAIN, DEVICE_TRACKER_DOMAIN, PERSON_DOMAIN): a_type = "BinarySensor" elif state.domain == CLIMATE_DOMAIN: # The type is resolved by the bridge before the accessory is created. a_type = CLIMATE_TYPES[config.get(CONF_TYPE, TYPE_THERMOSTAT)] elif state.domain == COVER_DOMAIN: device_class = state.attributes.get(EntityStateAttribute.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_DOMAIN: if fan_type := config.get(CONF_TYPE): a_type = FAN_TYPES[fan_type] else: a_type = "Fan" elif state.domain == HUMIDIFIER_DOMAIN: a_type = "HumidifierDehumidifier" elif state.domain == LIGHT_DOMAIN: a_type = "Light" elif state.domain == LOCK_DOMAIN: a_type = "Lock" elif state.domain == MEDIA_PLAYER_DOMAIN: device_class = state.attributes.get(EntityStateAttribute.DEVICE_CLASS) feature_list = config.get(CONF_FEATURE_LIST, []) if device_class == MediaPlayerDeviceClass.RECEIVER: a_type = "ReceiverMediaPlayer" elif device_class in ( MediaPlayerDeviceClass.TV, MediaPlayerDeviceClass.PROJECTOR, ): a_type = "TelevisionMediaPlayer" elif validate_media_player_features(state, feature_list): a_type = "MediaPlayer" elif state.domain == SENSOR_DOMAIN: device_class = state.attributes.get(EntityStateAttribute.DEVICE_CLASS) unit = state.attributes.get(EntityStateAttribute.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: a_type = "PM10Sensor" elif device_class == SensorDeviceClass.PM25: 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: a_type = "AirQualitySensor" elif device_class == SensorDeviceClass.CO: a_type = "CarbonMonoxideSensor" elif device_class == SensorDeviceClass.CO2: a_type = "CarbonDioxideSensor" elif device_class == SensorDeviceClass.ILLUMINANCE or unit == LIGHT_LUX: a_type = "LightSensor" # Fallbacks based on entity_id elif SensorDeviceClass.PM10 in state.entity_id: a_type = "PM10Sensor" elif SensorDeviceClass.PM25 in state.entity_id: a_type = "PM25Sensor" elif SensorDeviceClass.GAS in state.entity_id: a_type = "AirQualitySensor" elif "co2" in state.entity_id: a_type = "CarbonDioxideSensor" else: _LOGGER.debug( "%s: Unsupported sensor type (device_class=%s) (unit=%s)", state.entity_id, device_class, unit, ) elif state.domain == SWITCH_DOMAIN: if switch_type := config.get(CONF_TYPE): a_type = SWITCH_TYPES[switch_type] elif ( state.attributes.get(EntityStateAttribute.DEVICE_CLASS) == SwitchDeviceClass.OUTLET ): a_type = "Outlet" else: a_type = "Switch" elif state.domain == VALVE_DOMAIN: a_type = "Valve" elif state.domain == VACUUM_DOMAIN: a_type = "Vacuum" elif ( state.domain == LAWN_MOWER_DOMAIN and features & LawnMowerEntityFeature.DOCK and features & LawnMowerEntityFeature.START_MOWING ): a_type = "LawnMower" elif state.domain == REMOTE_DOMAIN and features & RemoteEntityFeature.ACTIVITY: a_type = "ActivityRemote" elif state.domain in ( AUTOMATION_DOMAIN, BUTTON_DOMAIN, INPUT_BOOLEAN_DOMAIN, INPUT_BUTTON_DOMAIN, REMOTE_DOMAIN, SCENE_DOMAIN, SCRIPT_DOMAIN, ): a_type = "Switch" elif state.domain in (INPUT_SELECT_DOMAIN, SELECT_DOMAIN): a_type = "SelectSwitch" elif state.domain == WATER_HEATER_DOMAIN: a_type = "WaterHeater" elif state.domain == CAMERA_DOMAIN: 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[str] = 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[untyped-decorator] 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], value: Any | None = None, ) -> None: """Fire event and call service for changes from HomeKit.""" self.hass.async_create_task( self.async_call_service_and_wait(domain, service, service_data, value), eager_start=True, ) async def async_call_service_and_wait( self, domain: str, service: str, service_data: dict[str, Any], value: Any | None = None, ) -> bool: """Fire event and call service, returning True when it succeeded. blocking=True so the handler's exception reaches us (the non-blocking path swallows it); on failure we resync so pyhap's optimistic target characteristic doesn't strand the tile on the requested action. """ event_data = { ATTR_ENTITY_ID: service_data.get(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) try: await self.hass.services.async_call( domain, service, service_data, blocking=True, context=context ) except HomeAssistantError as err: _LOGGER.warning( "%s: %s.%s failed (%s); re-syncing HomeKit state", self.entity_id, domain, service, err, ) except Exception: _LOGGER.exception( "%s: %s.%s raised unexpectedly; re-syncing HomeKit state", self.entity_id, domain, service, ) else: return True # This coroutine often runs fire-and-forget, so failures must be # logged here instead of by the loop's default task handler. try: if (state := self.hass.states.get(self.entity_id)) is not None: self.async_update_state(state) else: _LOGGER.debug( "%s: cannot re-sync HomeKit state; entity has no state", self.entity_id, ) except Exception: _LOGGER.exception("%s: re-syncing HomeKit state failed", self.entity_id) return False @ha_callback def async_reload(self) -> None: """Reload and recreate an accessory. 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[untyped-decorator] 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[untyped-decorator] 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