Merge pull request #12 from rlippmann/1.2.9-fixes

1.2.9

Author: rlippmann

CHANGELOG.md

@@ -1,3 +1,14 @@
+## 1.2.9 (2024-04-21)
+
+* ignore query string in check_login_errors().  This should fix a bug where the task was logged out
+  but not correctly being identified
+* remove unnecessary warning in alarm status check
+* add arm night
+* refactor update_alarm_from_etree()
+* bump to newer user agent
+* skip sync check if it will back off
+* fix linter issue in _initialize_sites
+
 ## 1.2.8 (2024-03-07)
 
 * add more detail to "invalid sync check" error logging

pyadtpulse/alarm_panel.py

@@ -21,6 +21,7 @@
 ADT_ALARM_UNKNOWN = "unknown"
 ADT_ALARM_ARMING = "arming"
 ADT_ALARM_DISARMING = "disarming"
+ADT_ALARM_NIGHT = "night"
 
 ALARM_STATUSES = (
     ADT_ALARM_AWAY,
@@ -29,8 +30,16 @@
     ADT_ALARM_UNKNOWN,
     ADT_ALARM_ARMING,
     ADT_ALARM_DISARMING,
+    ADT_ALARM_NIGHT,
 )
 
+ALARM_POSSIBLE_STATUS_MAP = {
+    "Disarmed": (ADT_ALARM_OFF, ADT_ALARM_ARMING),
+    "Armed Away": (ADT_ALARM_AWAY, ADT_ALARM_DISARMING),
+    "Armed Stay": (ADT_ALARM_HOME, ADT_ALARM_DISARMING),
+    "Armed Night": (ADT_ALARM_NIGHT, ADT_ALARM_DISARMING),
+}
+
 ADT_ARM_DISARM_TIMEOUT: float = 20
 
 
@@ -129,6 +138,16 @@ def is_disarming(self) -> bool:
         with self._state_lock:
             return self._status == ADT_ALARM_DISARMING
 
+    @property
+    def is_armed_night(self) -> bool:
+        """Return if system is in night mode.
+
+        Returns:
+            bool: True if system is in night mode
+        """
+        with self._state_lock:
+            return self._status == ADT_ALARM_NIGHT
+
     @property
     def last_update(self) -> float:
         """Return last update time.
@@ -198,7 +217,7 @@ async def _arm(
             if arm_result is not None:
                 error_block = arm_result.find(".//div")
                 if error_block is not None:
-                    error_text = arm_result.text_contents().replace(
+                    error_text = arm_result.text_content().replace(
                         "Arm AnywayCancel\n\n", ""
                     )
                     LOG.warning(
@@ -240,6 +259,18 @@ def arm_away(self, connection: PulseConnection, force_arm: bool = False) -> bool
         """
         return self._sync_set_alarm_mode(connection, ADT_ALARM_AWAY, force_arm)
 
+    @typechecked
+    def arm_night(self, connection: PulseConnection, force_arm: bool = False) -> bool:
+        """Arm the alarm in Night mode.
+
+        Args:
+            force_arm (bool, Optional): force system to arm
+
+        Returns:
+            bool: True if arm succeeded
+        """
+        return self._sync_set_alarm_mode(connection, ADT_ALARM_NIGHT, force_arm)
+
     @typechecked
     def arm_home(self, connection: PulseConnection, force_arm: bool = False) -> bool:
         """Arm the alarm in Home mode.
@@ -288,6 +319,19 @@ async def async_arm_home(
         """
         return await self._arm(connection, ADT_ALARM_HOME, force_arm)
 
+    @typechecked
+    async def async_arm_night(
+        self, connection: PulseConnection, force_arm: bool = False
+    ) -> bool:
+        """Arm alarm night async.
+
+        Args:
+            force_arm (bool, Optional): force system to arm
+        Returns:
+            bool: True if arm succeeded
+        """
+        return await self._arm(connection, ADT_ALARM_NIGHT, force_arm)
+
     @typechecked
     async def async_disarm(self, connection: PulseConnection) -> bool:
         """Disarm alarm async.
@@ -313,51 +357,44 @@ def update_alarm_from_etree(self, summary_html_etree: html.HtmlElement) -> None:
         value = summary_html_etree.find(".//span[@class='p_boldNormalTextLarge']")
         sat_location = "security_button_0"
         with self._state_lock:
+            status_found = False
+            last_updated = int(time())
             if value is not None:
                 text = value.text_content().lstrip().splitlines()[0]
-                last_updated = int(time())
-
-                if text.startswith("Disarmed"):
-                    if (
-                        self._status != ADT_ALARM_ARMING
-                        or last_updated - self._last_arm_disarm > ADT_ARM_DISARM_TIMEOUT
-                    ):
-                        self._status = ADT_ALARM_OFF
-                        self._last_arm_disarm = last_updated
-                elif text.startswith("Armed Away"):
-                    if (
-                        self._status != ADT_ALARM_DISARMING
-                        or last_updated - self._last_arm_disarm > ADT_ARM_DISARM_TIMEOUT
-                    ):
-                        self._status = ADT_ALARM_AWAY
-                        self._last_arm_disarm = last_updated
-                elif text.startswith("Armed Stay"):
-                    if (
-                        self._status != ADT_ALARM_DISARMING
-                        or last_updated - self._last_arm_disarm > ADT_ARM_DISARM_TIMEOUT
-                    ):
-                        self._status = ADT_ALARM_HOME
-                        self._last_arm_disarm = last_updated
-                else:
+
+                for (
+                    current_status,
+                    possible_statuses,
+                ) in ALARM_POSSIBLE_STATUS_MAP.items():
+                    if text.startswith(current_status):
+                        status_found = True
+                        if (
+                            self._status != possible_statuses[1]
+                            or last_updated - self._last_arm_disarm
+                            > ADT_ARM_DISARM_TIMEOUT
+                        ):
+                            self._status = possible_statuses[0]
+                            self._last_arm_disarm = last_updated
+                        break
+
+            if value is None or not status_found:
+                if not text.startswith("Status Unavailable"):
                     LOG.warning("Failed to get alarm status from '%s'", text)
-                    self._status = ADT_ALARM_UNKNOWN
-                    self._last_arm_disarm = last_updated
-                    return
-                LOG.debug("Alarm status = %s", self._status)
+                self._status = ADT_ALARM_UNKNOWN
+                self._last_arm_disarm = last_updated
+                return
+            LOG.debug("Alarm status = %s", self._status)
             sat_string = f'.//input[@id="{sat_location}"]'
             sat_button = summary_html_etree.find(sat_string)
             if sat_button is not None and "onclick" in sat_button.attrib:
                 on_click = sat_button.attrib["onclick"]
                 match = re.search(r"sat=([a-z0-9\-]+)", on_click)
                 if match:
                     self._sat = match.group(1)
-            elif len(self._sat) == 0:
-                LOG.warning("No sat recorded and was unable extract sat.")
-
-            if len(self._sat) > 0:
-                LOG.debug("Extracted sat = %s", self._sat)
+            if not self._sat:
+                LOG.warning("No sat recorded and was unable to extract sat.")
             else:
-                LOG.warning("Unable to extract sat")
+                LOG.debug("Extracted sat = %s", self._sat)
 
     @typechecked
     def set_alarm_attributes(self, alarm_attributes: dict[str, str]) -> None:

pyadtpulse/const.py

@@ -1,6 +1,6 @@
 """Constants for pyadtpulse."""
 
-__version__ = "1.2.8"
+__version__ = "1.2.9"
 
 DEFAULT_API_HOST = "https://portal.adtpulse.com"
 API_HOST_CA = "https://portal-ca.adtpulse.com"  # Canada
@@ -31,12 +31,12 @@
 # than that
 ADT_DEFAULT_POLL_INTERVAL = 2.0
 ADT_GATEWAY_MAX_OFFLINE_POLL_INTERVAL = 600.0
-ADT_MAX_BACKOFF: float = 15.0 * 60.0
+ADT_MAX_BACKOFF: float = 5.0 * 60.0
 ADT_DEFAULT_HTTP_USER_AGENT = {
     "User-Agent": (
-        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
         "AppleWebKit/537.36 (KHTML, like Gecko) "
-        "Chrome/100.0.4896.127 Safari/537.36 Edg/100.0.1185.44"
+        "Chrome/122.0.0.0 Safari/537.36"
     )
 }
 

pyadtpulse/pulse_connection.py

@@ -120,7 +120,7 @@ def determine_error_type():
             """
             self._login_in_progress = False
             url = self._connection_properties.make_url(ADT_LOGIN_URI)
-            if url == response_url_string:
+            if response_url_string.startswith(url):
                 error = tree.find(".//div[@id='warnMsgContents']")
                 if error is not None:
                     error_text = error.text_content()

pyadtpulse/pyadtpulse_async.py

@@ -171,13 +171,15 @@ async def _initialize_sites(self, tree: html.HtmlElement) -> None:
         """
         # typically, ADT Pulse accounts have only a single site (premise/location)
         single_premise = tree.find(".//span[@id='p_singlePremise']")
-        if single_premise is not None:
+        if single_premise is not None and single_premise.text:
             site_name = single_premise.text
             start_time = 0.0
             if self._pulse_connection.detailed_debug_logging:
                 start_time = time.time()
-            # FIXME: this code works, but it doesn't pass the linter
-            signout_link = str(tree.find(".//a[@class='p_signoutlink']").get("href"))
+            temp = tree.find(".//a[@class='p_signoutlink']")
+            signout_link = None
+            if temp is not None:
+                signout_link = str(temp.get("href"))
             if signout_link:
                 m = re.search("networkid=(.+)&", signout_link)
                 if m and m.group(1) and m.group(1):
@@ -272,9 +274,13 @@ def should_relogin(relogin_interval: int) -> bool:
             relogin_interval = self._pulse_properties.relogin_interval * 60
             try:
                 await asyncio.sleep(self._pulse_properties.keepalive_interval * 60)
-                if self._pulse_connection_status.retry_after > time.time():
+                if (
+                    self._pulse_connection_status.retry_after > time.time()
+                    or self._pulse_connection_status.get_backoff().backoff_count
+                    > WARN_TRANSIENT_FAILURE_THRESHOLD
+                ):
                     LOG.debug(
-                        "%s: Skipping actions because retry_after > now", task_name
+                        "%s: Skipping actions because query will backoff", task_name
                     )
                     continue
                 if not self._pulse_connection.is_connected:

pyadtpulse/site.py

@@ -78,6 +78,13 @@ async def async_arm_away(self, force_arm: bool = False) -> bool:
             self._pulse_connection, force_arm=force_arm
         )
 
+    @typechecked
+    async def async_arm_night(self, force_arm: bool = False) -> bool:
+        """Arm system away async."""
+        return await self.alarm_control_panel.async_arm_night(
+            self._pulse_connection, force_arm=force_arm
+        )
+
     async def async_disarm(self) -> bool:
         """Disarm system async."""
         return await self.alarm_control_panel.async_disarm(self._pulse_connection)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pyadtpulse"
-version = "1.2.8"
+version = "1.2.9"
 description = "Python interface for ADT Pulse security systems"
 authors = ["Ryan Snodgrass"]
 maintainers = ["Robert Lippmann"]