Refactor _logs_missing()

Author: bouwew

plugwise_usb/nodes/helpers/pulses.py

@@ -805,122 +805,91 @@ def _first_log_reference(
 
     def _logs_missing(self, from_timestamp: datetime) -> list[int] | None:
         """Calculate list of missing log addresses."""
-        if self._logs is None:
+        if self._logs is None or self.collected_logs < 2:
             self._log_addresses_missing = None
             return None
 
-        if self.collected_logs < 2:
-            return None
+        first = self._first_log_reference()
+        last = self._last_log_reference()
 
-        last_address, last_slot = self._last_log_reference()
-        if last_address is None or last_slot is None:
+        if None in first or None in last:
             _LOGGER.debug(
-                "_logs_missing | %s | last_address=%s, last_slot=%s",
-                self._mac,
-                last_address,
-                last_slot,
+                "_logs_missing | %s | first=%s, last=%s", self._mac, first, last
             )
             return None
 
-        first_address, first_slot = self._first_log_reference()
-        if first_address is None or first_slot is None:
-            _LOGGER.debug(
-                "_logs_missing | %s | first_address=%s, first_slot=%s",
-                self._mac,
-                first_address,
-                first_slot,
-            )
-            return None
+        first_address, first_slot = first
+        last_address, last_slot = last
+
+        if (
+            first == last and
+            self._logs[first_address][first_slot].timestamp == self._logs[last_address][last_slot].timestamp
+        ):
+            return None  # Only one unique log, not enough to work with
 
         missing = []
         _LOGGER.debug(
             "_logs_missing | %s | first_address=%s, last_address=%s",
-            self._mac,
-            first_address,
-            last_address,
+            self._mac, first_address, last_address
         )
 
-        if (
-            last_address == first_address
-            and last_slot == first_slot
-            and self._logs[first_address][first_slot].timestamp
-            == self._logs[last_address][last_slot].timestamp
-        ):
-            # Power consumption logging, so we need at least 4 logs.
-            return None
-
-        # Collect any missing address in current range
-        address = last_address
-        slot = last_slot
-        while not (address == first_address and slot == first_slot):
+        # Check for missing logs in the collected range
+        address, slot = last_address, last_slot
+        while (address, slot) != (first_address, first_slot):
             address, slot = calc_log_address(address, slot, -1)
             if address in missing:
                 continue
             if not self._log_exists(address, slot):
                 missing.append(address)
-                continue
-            if self._logs[address][slot].timestamp <= from_timestamp:
+            elif self._logs[address][slot].timestamp <= from_timestamp:
                 break
 
-        _LOGGER.debug(
-           "_logs_missing | %s | missing in range=%s", self._mac, missing
-        )
-
-        # return missing logs in range first
-        if len(missing) > 0:
+        if missing:
+            _LOGGER.debug("_logs_missing | %s | missing in range=%s", self._mac, missing)
             return missing
 
-        if first_address not in self._logs:
-            return missing
-
-        if first_slot not in self._logs[first_address]:
-            return missing
-
-        if self._logs[first_address][first_slot].timestamp < from_timestamp:
-            return missing
-
-        # Check if we are able to calculate log interval
-        address, slot = calc_log_address(first_address, first_slot, -1)
-        log_interval: int | None = None
-        if self._log_interval_consumption is not None:
-            log_interval = self._log_interval_consumption
-        elif self._log_interval_production is not None:
-            log_interval = self._log_interval_production
+        # Check if first slot exists and is recent enough
+        if (
+            first_address not in self._logs or
+            first_slot not in self._logs[first_address] or
+            self._logs[first_address][first_slot].timestamp < from_timestamp
+        ):
+            return []
 
+        # Estimate log interval
+        log_interval = self._log_interval_consumption or self._log_interval_production
         if (
-            self._log_interval_production is not None
-            and log_interval is not None
-            and self._log_interval_production < log_interval
+            self._log_interval_production is not None and
+            log_interval is not None and
+            self._log_interval_production < log_interval
         ):
             log_interval = self._log_interval_production
 
         if log_interval is None:
             return None
 
-        # We have an suspected interval, so try to calculate missing log addresses prior to first collected log
-        calculated_timestamp = self._logs[first_address][
-            first_slot
-        ].timestamp - timedelta(minutes=log_interval)
+        # Walk backward before the first known log to find additional missing logs
+        timestamp = self._logs[first_address][first_slot].timestamp
+        calc_timestamp = timestamp - timedelta(minutes=log_interval)
+        address, slot = calc_log_address(first_address, first_slot, -1)
+
         _LOGGER.debug(
-                "_logs_missing | %s | calculated timestamp=%s",
-                self._mac,
-                calculated_timestamp,
+            "_logs_missing | %s | calculated timestamp=%s", self._mac, calc_timestamp
         )
-        while from_timestamp < calculated_timestamp:
-            if (
-                address == self._first_empty_log_address
-                and slot == self._first_empty_log_slot
-            ):
+
+        while calc_timestamp > from_timestamp:
+            if (address, slot) == (self._first_empty_log_address, self._first_empty_log_slot):
                 break
             if address not in missing:
                 missing.append(address)
-            calculated_timestamp -= timedelta(minutes=log_interval)
+            calc_timestamp -= timedelta(minutes=log_interval)
             address, slot = calc_log_address(address, slot, -1)
 
         missing.sort(reverse=True)
         _LOGGER.debug("_logs_missing | %s | calculated missing=%s", self._mac, missing)
         return missing
 
+
     def _last_known_duration(self) -> timedelta:
         """Duration for last known logs."""
         if self._logs is None: