import asyncio import json import warnings from collections.abc import Callable import httpx from homematicip.access_point_update_state import AccessPointUpdateState from homematicip.base.channel_event import ChannelEvent from homematicip.class_maps import * from homematicip.client import Client from homematicip.connection.client_characteristics_builder import ( ClientCharacteristicsBuilder, ) from homematicip.connection.connection_context import ( ConnectionContext, ConnectionContextBuilder, ) from homematicip.connection.connection_factory import ConnectionFactory from homematicip.connection.websocket_handler import WebsocketHandler from homematicip.device import * from homematicip.EventHook import * from homematicip.exceptions.connection_exceptions import ( HmipAuthenticationError, HmipConnectionError, ) from homematicip.exceptions.home_exceptions import HomeNotInitializedError from homematicip.group import * from homematicip.location import Location from homematicip.oauth_otk import OAuthOTK from homematicip.rule import * from homematicip.securityEvent import * from homematicip.weather import Weather LOGGER = logging.getLogger(__name__) class AsyncHome(HomeMaticIPObject): """this class represents the 'Home' of the homematic ip""" _typeClassMap = TYPE_CLASS_MAP _typeGroupMap = TYPE_GROUP_MAP _typeSecurityEventMap = TYPE_SECURITY_EVENT_MAP _typeRuleMap = TYPE_RULE_MAP _typeFunctionalHomeMap = TYPE_FUNCTIONALHOME_MAP def __init__(self, connection=None): super().__init__(connection) # Connection Stuff self._connection_context: ConnectionContext | None = None self._websocket_client: WebsocketHandler | None = None self._auth_token: str | None = None # Events and EventHandler self._on_create = [] self._on_channel_event = [] self.onEvent = EventHook() # Home Attributes self.apExchangeClientId = None self.apExchangeState = ApExchangeState.NONE self.availableAPVersion = None self.carrierSense = None self.connected = None self.currentAPVersion = None self.deviceUpdateStrategy = DeviceUpdateStrategy.MANUALLY self.dutyCycle = None self.id = None self.lastReadyForUpdateTimestamp = None self.location = None self.pinAssigned = None self.powerMeterCurrency = None self.powerMeterUnitPrice = None self.timeZoneId = None self.updateState = HomeUpdateState.UP_TO_DATE self.weather = None self.accessPointUpdateStates = {} # collections of all devices, clients, groups, channels and rules self.devices = [] self.clients = [] self.groups = [] self.channels = [] self.rules = [] self.functionalHomes = [] async def init_async( self, access_point_id: str, auth_token: str | None = None, lookup: bool = True, use_rate_limiting: bool = True, ): """Initializes the home with the given access point id and auth token :param access_point_id: the access point id :param auth_token: the auth token :param lookup: deprecated and ignored — the URL lookup always happens. Pass a pre-built context via :meth:`init_with_context` if you need to skip the lookup. :param use_rate_limiting: if set to true, the connection will be rate limited """ if not lookup: warnings.warn( "init_async(lookup=False) is deprecated and ignored: the URL " "lookup always runs. Use init_with_context() with a pre-built " "ConnectionContext to skip the lookup.", DeprecationWarning, stacklevel=2, ) self._connection_context = await ConnectionContextBuilder.build_context_async(accesspoint_id=access_point_id, auth_token=auth_token) self._connection = ConnectionFactory.create_connection(self._connection_context, use_rate_limiting) def init_with_context( self, context: ConnectionContext, use_rate_limiting: bool = True, httpx_client_session: httpx.AsyncClient | None = None, ): self._connection_context = context self._connection = ConnectionFactory.create_connection(self._connection_context, use_rate_limiting, httpx_client_session) def set_auth_token(self, auth_token): """Sets the auth token for the connection. This is only necessary, if not already set in init function""" if self._connection_context: self._connection_context.auth_token = auth_token self._connection.update_connection_context(self._connection_context) def _clear_configuration(self): """Clears all objects from the home""" self.devices = [] self.clients = [] self.groups = [] self.channels = [] def from_json(self, js_home): super().from_json(js_home) self.weather = Weather(self._connection) self.weather.from_json(js_home["weather"]) if js_home["location"] != None: self.location = Location(self._connection) self.location.from_json(js_home["location"]) self.connected = js_home["connected"] self.currentAPVersion = js_home["currentAPVersion"] self.availableAPVersion = js_home["availableAPVersion"] self.timeZoneId = js_home["timeZoneId"] self.pinAssigned = js_home["pinAssigned"] self.dutyCycle = js_home["dutyCycle"] self.updateState = HomeUpdateState.from_str(js_home["updateState"]) self.powerMeterUnitPrice = js_home["powerMeterUnitPrice"] self.powerMeterCurrency = js_home["powerMeterCurrency"] self.deviceUpdateStrategy = DeviceUpdateStrategy.from_str( js_home["deviceUpdateStrategy"] ) self.lastReadyForUpdateTimestamp = js_home["lastReadyForUpdateTimestamp"] self.apExchangeClientId = js_home["apExchangeClientId"] self.apExchangeState = ApExchangeState.from_str(js_home["apExchangeState"]) self.id = js_home["id"] self.carrierSense = js_home["carrierSense"] for ap, state in js_home["accessPointUpdateStates"].items(): ap_state = AccessPointUpdateState(self._connection) ap_state.from_json(state) self.accessPointUpdateStates[ap] = ap_state self._get_rules(js_home) def on_create(self, handler): """Adds an event handler to the create method. Fires when a device is created.""" self._on_create.append(handler) def fire_create_event(self, *args, **kwargs): """Trigger the method tied to _on_create""" for _handler in self._on_create: _handler(*args, **kwargs) def fire_channel_event(self, *args, **kwargs): """Trigger the method tied to _on_channel_event""" for _handler in self._on_channel_event: _handler(*args, **kwargs) def remove_callback(self, handler): """Remove event handler.""" super().remove_callback(handler) if handler in self._on_create: self._on_create.remove(handler) def on_channel_event(self, handler): """Adds an event handler to the channel event method. Fires when a channel event is received.""" self._on_channel_event.append(handler) def remove_channel_event_handler(self, handler): """Remove event handler.""" if handler in self._on_channel_event: self._on_channel_event.remove(handler) async def download_configuration_async(self) -> dict: """downloads the current configuration from the cloud Returns the downloaded configuration as json Raises: HomeNotInitializedError: if the home is not initialized Exception: if the download fails """ if self._connection_context is None: raise HomeNotInitializedError client_characteristics = ClientCharacteristicsBuilder.get(self._connection_context.accesspoint_id) result = await self._rest_call_async( "home/getCurrentState", client_characteristics ) if not result.success: if result.status == 403: raise HmipAuthenticationError( f"Could not get the current configuration. Error: {result.status} {result.status_text} ({result.text})" ) raise HmipConnectionError( f"Could not get the current configuration. Error: {result.status} {result.status_text}" ) return result.json async def get_current_state_async(self, clear_config: bool = False): """downloads the current configuration and parses it into self Args: clear_config(bool): if set to true, this function will remove all old objects from self.devices, self.client, ... to have a fresh config instead of reparsing them """ logger.debug("Run get_current_state_async") json_state = await self.download_configuration_async() return self.update_home(json_state, clear_config) async def wait_for_websocket_connection_async( self, timeout: float = 120.0, poll_interval: float = 2.0, warning_threshold: float = 60.0, ) -> bool: """Wait up to ``timeout`` seconds for the websocket to report connected. Polls :meth:`websocket_is_connected` every ``poll_interval`` seconds. If still not connected after ``warning_threshold`` seconds, emits a single ``WARNING`` log line. Returns ``True`` once connected, or ``False`` if ``timeout`` elapses first. Designed for post-reconnect recovery in long-running clients (e.g. Home Assistant). Callers typically proceed with a state refresh even on timeout, since a REST call may still recover state. Args: timeout: maximum total wait, in seconds. poll_interval: how often to check :meth:`websocket_is_connected`. Must be > 0. warning_threshold: emit a "still waiting" warning once this many seconds have elapsed without a connection. Raises: ValueError: if ``poll_interval`` is not positive. """ if poll_interval <= 0: raise ValueError("poll_interval must be positive") elapsed = 0.0 warning_logged = False while not self.websocket_is_connected(): if elapsed >= timeout: logger.warning( "Websocket did not reconnect within %s seconds", timeout, ) return False sleep_for = min(poll_interval, timeout - elapsed) await asyncio.sleep(sleep_for) elapsed += sleep_for if ( not warning_logged and warning_threshold <= elapsed < timeout and not self.websocket_is_connected() ): warning_logged = True logger.warning( "Still waiting for websocket reconnect after %s seconds", elapsed, ) return True async def refresh_state_after_reconnect_async( self, clear_config: bool = False, websocket_timeout: float = 120.0, websocket_poll_interval: float = 2.0, websocket_warning_threshold: float = 60.0, retry_initial_delay: float = 8.0, retry_max_delay: float = 1500.0, ) -> None: """Wait (best-effort) for websocket reconnect, then refresh state with retry. Convenience wrapper combining :meth:`wait_for_websocket_connection_async` and :meth:`get_current_state_async_with_retry`. The websocket wait is best-effort: if it times out, the state refresh proceeds anyway, since the REST call is independent and may still recover state. Re-raises ``HmipAuthenticationError``, ``HomeNotInitializedError``, and ``asyncio.CancelledError`` immediately. All other errors trigger retry. Intended for post-reconnect recovery flows in long-running clients. """ await self.wait_for_websocket_connection_async( timeout=websocket_timeout, poll_interval=websocket_poll_interval, warning_threshold=websocket_warning_threshold, ) await self.get_current_state_async_with_retry( clear_config=clear_config, initial_delay=retry_initial_delay, max_delay=retry_max_delay, ) async def get_current_state_async_with_retry( self, clear_config: bool = False, initial_delay: float = 8.0, max_delay: float = 1500.0, ): """Refresh state with exponential backoff retry on transient errors. Wraps :meth:`get_current_state_async` with a retry loop intended for post-reconnect recovery in long-running clients (e.g. Home Assistant). - ``HmipAuthenticationError`` is re-raised immediately so the caller can trigger reauth. - ``HomeNotInitializedError`` is re-raised immediately because it signals a programmer error (calling this before ``init``). - ``asyncio.CancelledError`` is re-raised to propagate cancellation. - All other exceptions, including ``HmipConnectionError`` subclasses and unforeseen errors, trigger a retry with exponential backoff (``initial_delay`` doubling up to ``max_delay`` seconds). Args: clear_config: forwarded to :meth:`get_current_state_async`. initial_delay: seconds to wait before the first retry. max_delay: cap on the backoff delay between retries. """ delay = initial_delay while True: try: return await self.get_current_state_async(clear_config=clear_config) except (HmipAuthenticationError, HomeNotInitializedError): raise except asyncio.CancelledError: raise except HmipConnectionError as err: logger.warning( "get_current_state failed, retrying in %s seconds: %s", delay, err, ) except Exception: logger.exception( "Unexpected error in get_current_state, retrying in %s seconds", delay, ) await asyncio.sleep(delay) delay = min(delay * 2, max_delay) def update_home(self, json_state, clear_config: bool = False): """parse a given json configuration into self. This will update the whole home including devices, clients and groups. Args: json_state(dict): the json configuration as dictionary clear_config(bool): if set to true, this function will remove all old objects from self.devices, self.client, ... to have a fresh config instead of reparsing them """ if clear_config: self._clear_configuration() self._get_devices(json_state) self._get_clients(json_state) self._get_groups(json_state) self._load_functionalChannels() js_home = json_state["home"] return self.update_home_only(js_home, clear_config) def update_home_only(self, js_home, clear_config: bool = False): """parse a given home json configuration into self. This will update only the home without updating devices, clients and groups. Args: js_home(dict): the json configuration as dictionary clear_config(bool): if set to true, this function will remove all old objects from self.devices, self.client, ... to have a fresh config instead of reparsing them """ if clear_config: self.rules = [] self.functionalHomes = [] self.from_json(js_home) self._get_functionalHomes(js_home) return True def _get_devices(self, json_state): self.devices = [x for x in self.devices if x.id in json_state["devices"]] for id_, raw in json_state["devices"].items(): try: _device = self.search_device_by_id(id_) if _device: _device.from_json(raw) else: self.devices.append(self._parse_device(raw)) except Exception as err: LOGGER.exception( "An exception in _get_devices (device-id %s) of type %s occurred", id_, type(err).__name__, ) return def _parse_device(self, json_state): try: deviceType = DeviceType.from_str(json_state["type"]) d = self._typeClassMap[deviceType](self._connection) d.from_json(json_state) return d except: d = self._typeClassMap[DeviceType.BASE_DEVICE](self._connection) d.from_json(json_state) LOGGER.warning("There is no class for device '%s' yet", json_state["type"]) return d def _get_rules(self, json_state): self.rules = [ x for x in self.rules if x.id in json_state["ruleMetaDatas"] ] for id_, raw in json_state["ruleMetaDatas"].items(): _rule = self.search_rule_by_id(id_) if _rule: _rule.from_json(raw) else: self.rules.append(self._parse_rule(raw)) def _parse_rule(self, json_state): try: ruleType = AutomationRuleType.from_str(json_state["type"]) r = self._typeRuleMap[ruleType](self._connection) r.from_json(json_state) return r except: r = Rule(self._connection) r.from_json(json_state) LOGGER.warning("There is no class for rule '%s' yet", json_state["type"]) return r def _get_clients(self, json_state): self.clients = [x for x in self.clients if x.id in json_state["clients"]] for id_, raw in json_state["clients"].items(): _client = self.search_client_by_id(id_) if _client: _client.from_json(raw) else: c = Client(self._connection) c.from_json(raw) self.clients.append(c) def _parse_group(self, json_state): g = None if json_state["type"] == "META": g = MetaGroup(self._connection) g.from_json(json_state, self.devices, self.groups) else: try: groupType = GroupType.from_str(json_state["type"]) g = self._typeGroupMap[groupType](self._connection) g.from_json(json_state, self.devices) except: g = self._typeGroupMap[GroupType.GROUP](self._connection) g.from_json(json_state, self.devices) LOGGER.warning( "There is no class for group '%s' yet", json_state["type"] ) return g def _get_groups(self, json_state): self.groups = [x for x in self.groups if x.id in json_state["groups"]] metaGroups = [] for id_, raw in json_state["groups"].items(): _group = self.search_group_by_id(id_) if _group: if isinstance(_group, MetaGroup): _group.from_json(raw, self.devices, self.groups) else: _group.from_json(raw, self.devices) else: group_type = raw["type"] if group_type == "META": metaGroups.append(raw) else: self.groups.append(self._parse_group(raw)) for mg in metaGroups: self.groups.append(self._parse_group(mg)) def _get_functionalHomes(self, json_state): """loads the functional homes.""" for solution, functionalHome in json_state["functionalHomes"].items(): # Reuse an existing functionalHome with the same solution if any, # so that repeated state updates don't accumulate duplicates. existing = next( (fh for fh in self.functionalHomes if fh.solution == solution), None, ) try: solutionType = FunctionalHomeType.from_str(solution) if existing is None: existing = self._typeFunctionalHomeMap[solutionType]( self._connection ) self.functionalHomes.append(existing) existing.from_json(functionalHome, self.groups) except: if existing is None: existing = FunctionalHome(self._connection) self.functionalHomes.append(existing) LOGGER.warning( "There is no class for functionalHome '%s' yet", solution ) existing.from_json(functionalHome, self.groups) def _load_functionalChannels(self): """loads the functional channels for all devices""" for d in self.devices: d.load_functionalChannels(self.groups, self.channels) def get_functionalHome(self, functionalHomeType: type) -> FunctionalHome | None: """gets the specified functionalHome :param functionalHomeType: the type of the functionalHome which should be returned :return: the FunctionalHome or None if it couldn't be found """ for x in self.functionalHomes: if isinstance(x, functionalHomeType): return x return None def search_device_by_id(self, deviceID) -> Device | None: """searches a device by given id :param deviceID: the device to search for :return: the Device object or None if it couldn't find a device """ for d in self.devices: if d.id == deviceID: return d return None def search_channel(self, device_id, channel_index) -> FunctionalChannel | None: """searches a channel by given deviceID and channelIndex. :param device_id: the device to search for :param channel_index: the channel to search for :return: the FunctionalChannel object or None if it couldn't find a channel """ found_device = [d for d in self.devices if d.id == device_id] d = found_device[0] if found_device else None if d is not None: found_channel = [ch for ch in d.functionalChannels if ch.index == channel_index] return found_channel[0] if found_channel else None return None def search_group_by_id(self, groupID) -> Group | None: """searches a group by given id :param groupID: the group to search for :return: the group object or None if it couldn't find a group """ for g in self.groups: if g.id == groupID: return g return None def search_client_by_id(self, clientID) -> Client | None: """searches a client by given id :param clientID: the client to search for :return: the client object or None if it couldn't find a client """ for c in self.clients: if c.id == clientID: return c return None def search_rule_by_id(self, ruleID) -> Rule | None: """searches a rule by given id :param ruleID: the rule to search for :return: the rule object or None if it couldn't find a rule """ for r in self.rules: if r.id == ruleID: return r return None def get_security_zones_activation(self) -> tuple[bool, bool]: """returns the value of the security zones if they are armed or not :return: internal, external """ internal_active = False external_active = False for g in self.groups: if isinstance(g, SecurityZoneGroup): if g.label == "EXTERNAL": external_active = g.active elif g.label == "INTERNAL": internal_active = g.active return internal_active, external_active async def set_security_zones_activation_async(self, internal=True, external=True): """this function will set the alarm system to armed or disable it Examples: arming while being at home home.set_security_zones_activation(False,True) arming without being at home home.set_security_zones_activation(True,True) disarming the alarm system home.set_security_zones_activation(False,False) :param internal: activates/deactivates the internal zone :param external: activates/deactivates the external zone """ data = {"zonesActivation": {"EXTERNAL": external, "INTERNAL": internal}} return await self._rest_call_async("home/security/setZonesActivation", data) async def set_silent_alarm_async(self, internal=True, external=True): """this function will set the silent alarm for internal or external :param internal: activates/deactivates the silent alarm for internal zone :param external: activates/deactivates the silent alarm for the external zone """ data = {"zonesSilentAlarm": {"EXTERNAL": external, "INTERNAL": internal}} return await self._rest_call_async("home/security/setZonesSilentAlarm", data) async def set_location_async( self, city: str, latitude: str | float, longitude: str | float ): data = {"city": city, "latitude": latitude, "longitude": longitude} return await self._rest_call_async("home/setLocation", data) async def set_cooling_async(self, cooling): data = {"cooling": cooling} return await self._rest_call_async("home/heating/setCooling", data) async def set_intrusion_alert_through_smoke_detectors_async(self, activate: bool = True): """activate or deactivate if smoke detectors should "ring" during an alarm :param activate: True will let the smoke detectors "ring" during an alarm """ data = {"intrusionAlertThroughSmokeDetectors": activate} return await self._rest_call_async( "home/security/setIntrusionAlertThroughSmokeDetectors", data ) async def activate_absence_with_period_async(self, endtime: datetime): """activates the absence mode until the given time :param endtime: the time when the absence should automatically be disabled """ data = {"endTime": endtime.strftime("%Y_%m_%d %H:%M")} return await self._rest_call_async( "home/heating/activateAbsenceWithPeriod", data ) async def activate_absence_permanent_async(self): """activates the absence forever""" return await self._rest_call_async("home/heating/activateAbsencePermanent") async def activate_absence_with_duration_async(self, duration: int): """activates the absence mode for a given time :param duration: the absence duration in minutes """ data = {"duration": duration} return await self._rest_call_async( "home/heating/activateAbsenceWithDuration", data ) async def deactivate_absence_async(self): """deactivates the absence mode immediately""" return await self._rest_call_async("home/heating/deactivateAbsence") async def activate_vacation_async(self, endtime: datetime, temperature: float): """activates the vacation mode until the given time :param endtime: the time when the vacation mode should automatically be disabled :param temperature: the settemperature during the vacation mode """ data = { "endTime": endtime.strftime("%Y_%m_%d %H:%M"), "temperature": temperature, } return await self._rest_call_async("home/heating/activateVacation", data) async def deactivate_vacation_async(self): """deactivates the vacation mode immediately""" return await self._rest_call_async("home/heating/deactivateVacation") async def set_pin_async(self, new_pin: str | None, old_pin: str | None = None) -> dict: """sets a new pin for the home :param new_pin: the new pin :param old_pin: optional, if there is currently a pin active it must be given here. :return: the result of the call """ if new_pin is None: new_pin = "" headers = None if old_pin: # Copy before mutation: self._connection._headers is the live # connection-wide dict, and leaking the old PIN into it would # leak into every subsequent request. headers = dict(self._connection._headers) headers["PIN"] = str(old_pin) result = await self._rest_call_async("home/setPin", body={"pin": new_pin}, custom_header=headers) if not result.success: LOGGER.error("Could not set the pin. Error: %s", result.status_text) return result.json async def set_zone_activation_delay_async(self, delay): data = {"zoneActivationDelay": delay} return await self._rest_call_async( "home/security/setZoneActivationDelay", body=data ) async def get_security_journal_async(self): journal = await self._rest_call_async("home/security/getSecurityJournal") if not journal.success: LOGGER.error( "Could not get the security journal. Error: %s", journal.status_text ) return None ret = [] for entry in journal.json["entries"]: journal_entry = None try: event_type = SecurityEventType(entry["eventType"]) if event_type in self._typeSecurityEventMap: journal_entry = self._typeSecurityEventMap[event_type](self._connection) except: journal_entry = SecurityEvent(self._connection) LOGGER.warning("There is no class for %s yet", entry["eventType"]) if journal_entry is not None: journal_entry.from_json(entry) ret.append(journal_entry) return ret def delete_group(self, group: Group): """deletes the given group from the cloud :param group: the group to delete """ return group.delete() async def get_OAuth_OTK_async(self): token = OAuthOTK(self._connection) result = await self._rest_call_async("home/getOAuthOTK") token.from_json(result.json) return token async def set_timezone_async(self, timezone: str): """sets the timezone for the AP. e.g. "Europe/Berlin" :param timezone: the new timezone """ data = {"timezoneId": timezone} return await self._rest_call_async("home/setTimezone", body=data) async def set_powermeter_unit_price_async(self, price): data = {"powerMeterUnitPrice": price} return await self._rest_call_async("home/setPowerMeterUnitPrice", body=data) async def set_zones_device_assignment_async(self, internal_devices, external_devices) -> dict: """sets the devices for the security zones :param internal_devices: the devices which should be used for the internal zone :param external_devices: the devices which should be used for the external(hull) zone :return: the result of the call """ internal = [x.id for x in internal_devices] external = [x.id for x in external_devices] data = {"zonesDeviceAssignment": {"INTERNAL": internal, "EXTERNAL": external}} return await self._rest_call_async( "home/security/setZonesDeviceAssignment", body=data ) async def start_inclusion_async(self, device_id: str): """start inclusion mode for specific device :param device_id: sgtin of device """ data = {"deviceId": device_id} return await self._rest_call_async( "home/startInclusionModeForDevice", body=data ) async def enable_events(self, additional_message_handler: Callable | None = None): """Connect to Websocket and listen for events""" if self._websocket_client: if self._websocket_client.is_running(): return await self._websocket_client.stop() self._websocket_client = WebsocketHandler() self._websocket_client.add_on_message_handler(self._ws_on_message) if additional_message_handler: self._websocket_client.add_on_message_handler(additional_message_handler) await self._websocket_client.start(self._connection_context) async def disable_events_async(self): """Stop Websocket Connection""" logger.debug("Called disable_events_async") if self._websocket_client: await self._websocket_client.stop() self._websocket_client = None async def _ws_on_message(self, message) -> None: LOGGER.debug(message) js = json.loads(message) event_list = [] for event in js["events"].values(): try: pushEventType = EventType(event["pushEventType"]) LOGGER.debug(pushEventType) obj = None if pushEventType == EventType.GROUP_CHANGED: data = event["group"] obj = self.search_group_by_id(data["id"]) if obj is None: obj = self._parse_group(data) self.groups.append(obj) pushEventType = EventType.GROUP_ADDED self.fire_create_event(obj, event_type=pushEventType, obj=obj) if type(obj) is MetaGroup: obj.from_json(data, self.devices, self.groups) else: obj.from_json(data, self.devices) obj.fire_update_event(data, event_type=pushEventType, obj=obj) elif pushEventType == EventType.HOME_CHANGED: data = event["home"] obj = self obj.update_home_only(data) obj.fire_update_event(data, event_type=pushEventType, obj=obj) elif pushEventType == EventType.CLIENT_ADDED: data = event["client"] obj = Client(self._connection) obj.from_json(data) self.clients.append(obj) elif pushEventType == EventType.CLIENT_CHANGED: data = event["client"] obj = self.search_client_by_id(data["id"]) obj.from_json(data) obj.fire_update_event(data, event_type=pushEventType, obj=obj) elif pushEventType == EventType.CLIENT_REMOVED: obj = self.search_client_by_id(event["id"]) self.clients.remove(obj) obj.fire_remove_event(obj, event_type=pushEventType, obj=obj) elif pushEventType == EventType.DEVICE_ADDED: data = event["device"] obj = self._parse_device(data) obj.load_functionalChannels(self.groups, self.channels) self.devices.append(obj) self.fire_create_event(data, event_type=pushEventType, obj=obj) elif pushEventType == EventType.DEVICE_CHANGED: data = event["device"] obj = self.search_device_by_id(data["id"]) if obj is None: # no DEVICE_ADDED Event? obj = self._parse_device(data) self.devices.append(obj) pushEventType = EventType.DEVICE_ADDED self.fire_create_event(data, event_type=pushEventType, obj=obj) else: obj.from_json(data) obj.load_functionalChannels(self.groups, self.channels) obj.fire_update_event(data, event_type=pushEventType, obj=obj) elif pushEventType == EventType.DEVICE_REMOVED: obj = self.search_device_by_id(event["id"]) obj.fire_remove_event(obj, event_type=pushEventType, obj=obj) self.devices.remove(obj) elif pushEventType == EventType.DEVICE_CHANNEL_EVENT: channel_event = ChannelEvent() channel_event.from_json(event) ch = self.search_channel(channel_event.deviceId, channel_event.channelIndex) if ch is not None: ch.fire_channel_event(channel_event) elif pushEventType == EventType.GROUP_REMOVED: obj = self.search_group_by_id(event["id"]) obj.fire_remove_event(obj, event_type=pushEventType, obj=obj) self.groups.remove(obj) elif pushEventType == EventType.GROUP_ADDED: group = event["group"] obj = self._parse_group(group) self.groups.append(obj) self.fire_create_event(obj, event_type=pushEventType, obj=obj) elif pushEventType == EventType.SECURITY_JOURNAL_CHANGED: pass # data is just none so nothing to do here # TODO: implement INCLUSION_REQUESTED, NONE event_list.append({"eventType": pushEventType, "data": obj}) except ValueError: # pragma: no cover LOGGER.warning( "Unknown EventType '%s' Data: %s", event["pushEventType"], event ) except Exception as err: # pragma: no cover LOGGER.exception(err) self.onEvent.fire(event_list) def websocket_is_connected(self): """returns if the websocket is connected.""" return self._websocket_client.is_connected() if self._websocket_client else False def websocket_last_message_time(self) -> float | None: """Returns the loop timestamp of the last received websocket message.""" if self._websocket_client is None: return None return self._websocket_client.last_message_time() def websocket_message_count(self) -> int: """Returns the number of received websocket messages.""" if self._websocket_client is None: return 0 return self._websocket_client.message_count() def websocket_seconds_since_last_message(self) -> float | None: """Returns the seconds since the last received websocket message.""" if self._websocket_client is None: return None return self._websocket_client.seconds_since_last_message() def websocket_reconnect_attempt_count(self) -> int: """Returns the number of reconnect attempts in the current outage.""" if self._websocket_client is None: return 0 return self._websocket_client.reconnect_attempt_count() def websocket_last_disconnect_reason(self) -> str | None: """Returns the last websocket disconnect or reconnect reason.""" if self._websocket_client is None: return None return self._websocket_client.last_disconnect_reason() def set_on_connected_handler(self, handler: Callable): """Sets a callback that is called when the WebSocket connection is established.""" if self._websocket_client: self._websocket_client.add_on_connected_handler(handler) def set_on_disconnected_handler(self, handler: Callable): """Sets a callback that is called when the WebSocket connection is closed.""" if self._websocket_client: self._websocket_client.add_on_disconnected_handler(handler) def set_on_reconnect_handler(self, handler: Callable): """Sets a callback that is called when the WebSocket connection is trying to reconnect after a disconnect.""" if self._websocket_client: self._websocket_client.add_on_reconnect_handler(handler) def set_on_websocket_stale_handler(self, handler: Callable): """Sets a callback that is called when the WebSocket reports connected but has not received a message past the configured thresholds. Handler signature: ``handler(severity: str, seconds_since: float)``. See :meth:`WebsocketHandler.add_on_stale_handler` for details. """ if self._websocket_client: self._websocket_client.add_on_stale_handler(handler)