Improve

Author: bouwew

plugwise_usb/nodes/helpers/pulses.py

@@ -91,6 +91,8 @@ def __init__(self, mac: str) -> None:
         self._logs: dict[int, dict[int, PulseLogRecord]] | None = None
         self._log_addresses_missing: list[int] | None = None
         self._log_production: bool | None = None
+        self._next_log_exists = False
+        self._prev_log_exists = False
         self._pulses_consumption: int | None = None
         self._pulses_production: int | None = None
         self._pulses_timestamp: datetime | None = None
@@ -511,61 +513,66 @@ def _update_log_direction(
             slot,
             timestamp,
         )
-        prev_exists = next_exists = False
         prev_address, prev_slot = calc_log_address(address, slot, -1)
         if self._log_exists(prev_address, prev_slot):
-            prev_exists = True
             prev_timestamp = self._logs[prev_address][prev_slot].timestamp
             _LOGGER.debug(
                 "_update_log_direction | pr_address=%s | pr_slot=%s | timestamp=%s",
                 prev_address,
                 prev_slot,
                 prev_timestamp,
             )
-            if prev_timestamp == timestamp:
-                # Given log is the second log with same timestamp,
-                # mark direction as production
-                self._logs[address][slot].is_consumption = False
-                self._logs[prev_address][prev_slot].is_consumption = True
-                self._log_production = True
-            elif self._log_production:
-                self._logs[address][slot].is_consumption = True
-                if self._logs[prev_address][prev_slot].is_consumption:
-                    self._logs[prev_address][prev_slot].is_consumption = False
-                    self._reset_log_references()
-            elif self._log_production is None:
-                self._log_production = False
+            if not self._prev_log_exists:
+                self._prev_log_exists = True
+                if prev_timestamp == timestamp:
+                    # Given log is the second log with same timestamp,
+                    # mark direction as production
+                    self._logs[address][slot].is_consumption = False
+                    self._logs[prev_address][prev_slot].is_consumption = True
+                    self._log_production = True
+                elif self._log_production:
+                    self._logs[address][slot].is_consumption = True
+                    if self._logs[prev_address][prev_slot].is_consumption:
+                        self._logs[prev_address][prev_slot].is_consumption = False
+                        self._reset_log_references()
+                elif self._log_production is None:
+                    self._log_production = False
+        elif self._prev_log_exists:    
+            self._prev_log_exists = False
 
         next_address, next_slot = calc_log_address(address, slot, 1)
         if self._log_exists(next_address, next_slot):
-            next_exists = True
             next_timestamp = self._logs[next_address][next_slot].timestamp
             _LOGGER.debug(
                 "_update_log_direction | nxt_address=%s | nxt_slot=%s | timestamp=%s",
                 next_address,
                 next_slot,
                 next_timestamp,
             )
-            if next_timestamp == timestamp:
-                # Given log is the first log with same timestamp,
-                # mark direction as production of next log
-                self._logs[address][slot].is_consumption = True
-                if self._logs[next_address][next_slot].is_consumption:
-                    self._logs[next_address][next_slot].is_consumption = False
-                    self._reset_log_references()
-                self._log_production = True
-            elif self._log_production:
-                self._logs[address][slot].is_consumption = False
-                self._logs[next_address][next_slot].is_consumption = True
-            elif self._log_production is None:
-                self._log_production = False
-
-            if prev_exists and next_exists:
-                # _log_production is True when 2 out of 3 consecutive slots have
-                # the same timestamp, otherwise it is False
-                self._log_production = (
-                    next_timestamp == timestamp and prev_timestamp != timestamp
-                ) or (next_timestamp == prev_timestamp and next_timestamp != timestamp)
+            if not self._next_log_exists:
+                self._next_log_exists = True
+                if next_timestamp == timestamp:
+                    # Given log is the first log with same timestamp,
+                    # mark direction as production of next log
+                    self._logs[address][slot].is_consumption = True
+                    if self._logs[next_address][next_slot].is_consumption:
+                        self._logs[next_address][next_slot].is_consumption = False
+                        self._reset_log_references()
+                    self._log_production = True
+                elif self._log_production:
+                    self._logs[address][slot].is_consumption = False
+                    self._logs[next_address][next_slot].is_consumption = True
+                elif self._log_production is None:
+                    self._log_production = False
+        elif self._next_log_exists:
+            self._next_log_exists = False
+
+        if self._prev_log_exists and self._next_log_exists:
+            # _log_production is True when 2 out of 3 consecutive slots have
+            # the same timestamp, otherwise it is False
+            self._log_production = (
+                next_timestamp == timestamp and prev_timestamp != timestamp
+            ) or (next_timestamp == prev_timestamp and next_timestamp != timestamp)
 
     def _update_log_interval(self) -> None:
         """Update the detected log interval based on the most recent two logs."""