Merge pull request #247 from plugwise/enable_prod_2

Implement setting of energy logging intervals

Author: bouwew

CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 
-## v0.40.1
+## v0.41.0
+
+- Implement setting of energy logging intervals [#247](https://github.com/plugwise/python-plugwise-usb/pull/247)
+
+## v0.40.1 (not released)
 
 - Improve device Name and Model detection for Switch [#248](https://github.com/plugwise/python-plugwise-usb/pull/248)
 

plugwise_usb/__init__.py

@@ -210,6 +210,16 @@ async def set_accept_join_request(self, state: bool) -> bool:
             raise NodeError(f"Failed setting accept joining: {exc}") from exc
         return True
 
+    async def set_energy_intervals(
+        self, mac: str, cons_interval: int, prod_interval: int
+    ) -> bool:
+        """Configure the energy logging interval settings."""
+        try:
+            await self._network.set_energy_intervals(mac, cons_interval, prod_interval)
+        except (MessageError, NodeError, ValueError) as exc:
+            raise NodeError(f"{exc}") from exc
+        return True
+
     async def clear_cache(self) -> None:
         """Clear current cache."""
         if self._network is not None:

plugwise_usb/messages/requests.py

@@ -1264,7 +1264,7 @@ class CircleMeasureIntervalRequest(PlugwiseRequest):
 
     FIXME: Make sure production interval is a multiply of consumption !!
 
-    Response message: Ack message with ???  TODO:
+    Response message: NodeResponse with ack-type POWER_LOG_INTERVAL_ACCEPTED 
     """
 
     _identifier = b"0057"
@@ -1281,6 +1281,17 @@ def __init__(
         self._args.append(Int(consumption, length=4))
         self._args.append(Int(production, length=4))
 
+    async def send(self) -> NodeResponse | None:
+        """Send request."""
+        result = await self._send_request()
+        if isinstance(result, NodeResponse):
+            return result
+        if result is None:
+            return None
+        raise MessageError(
+            f"Invalid response message. Received {result.__class__.__name__}, expected NodeResponse"
+        )
+
 
 class NodeClearGroupMacRequest(PlugwiseRequest):
     """TODO: usage?.

plugwise_usb/messages/responses.py

@@ -56,9 +56,10 @@ class StickResponseType(bytes, Enum):
 class NodeResponseType(bytes, Enum):
     """Response types of a 'NodeResponse' reply message."""
 
-    CIRCLE_PLUS = b"00DD"  # type for CirclePlusAllowJoiningRequest with state false
+    CIRCLE_PLUS = b"00DD"  # ack for CirclePlusAllowJoiningRequest with state false
     CLOCK_ACCEPTED = b"00D7"
-    JOIN_ACCEPTED = b"00D9"  # type for CirclePlusAllowJoiningRequest with state true
+    JOIN_ACCEPTED = b"00D9"  # ack for CirclePlusAllowJoiningRequest with state true
+    POWER_LOG_INTERVAL_ACCEPTED = b"00F8"  # ack for CircleMeasureIntervalRequest
     RELAY_SWITCHED_OFF = b"00DE"
     RELAY_SWITCHED_ON = b"00D8"
     RELAY_SWITCH_FAILED = b"00E2"
@@ -68,7 +69,6 @@ class NodeResponseType(bytes, Enum):
 
     # TODO: Validate these response types
     SED_CONFIG_FAILED = b"00F7"
-    POWER_LOG_INTERVAL_ACCEPTED = b"00F8"
     POWER_CALIBRATION_ACCEPTED = b"00DA"
 
 

plugwise_usb/network/__init__.py

@@ -14,7 +14,11 @@
 from ..connection import StickController
 from ..constants import UTF8
 from ..exceptions import CacheError, MessageError, NodeError, StickError, StickTimeout
-from ..messages.requests import CirclePlusAllowJoiningRequest, NodePingRequest
+from ..messages.requests import (
+    CirclePlusAllowJoiningRequest,
+    CircleMeasureIntervalRequest,
+    NodePingRequest,
+)
 from ..messages.responses import (
     NODE_AWAKE_RESPONSE_ID,
     NODE_JOIN_ID,
@@ -537,6 +541,34 @@ async def allow_join_requests(self, state: bool) -> None:
         _LOGGER.debug("Sent AllowJoiningRequest to Circle+ with state=%s", state)
         self.accept_join_request = state
 
+    async def set_energy_intervals(
+        self, mac: str, consumption: int, production: int
+    ) -> None:
+        """Set the logging intervals for both energy consumption and production.
+
+        Default: consumption = 60, production = 0.
+        For logging energy in both directions set both to 60.
+        """
+        # Validate input parameters
+        if consumption <= 0:
+            raise ValueError("Consumption interval must be positive")
+        if production < 0:
+            raise ValueError("Production interval must be non-negative")
+        if production > 0 and production % consumption != 0:
+            raise ValueError("Production interval must be a multiple of consumption interval")
+
+        _LOGGER.debug("set_energy_intervals | cons=%s, prod=%s", consumption, production)
+        request = CircleMeasureIntervalRequest(
+            self._controller.send, bytes(mac, UTF8), consumption, production
+        )
+        if (response := await request.send()) is None:
+            raise NodeError("No response for CircleMeasureIntervalRequest.")
+
+        if response.response_type != NodeResponseType.POWER_LOG_INTERVAL_ACCEPTED:
+            raise MessageError(
+                f"Unknown NodeResponseType '{response.response_type.name}' received"
+            )
+
     def subscribe_to_node_events(
         self,
         node_event_callback: Callable[[NodeEvent, str], Coroutine[Any, Any, None]],

plugwise_usb/nodes/helpers/pulses.py

@@ -88,6 +88,8 @@ def __init__(self, mac: str) -> None:
         self._rollover_consumption = False
         self._rollover_production = False
 
+        self._first_next_log_processed = False
+        self._first_prev_log_processed = False
         self._logs: dict[int, dict[int, PulseLogRecord]] | None = None
         self._log_addresses_missing: list[int] | None = None
         self._log_production: bool | None = None
@@ -377,6 +379,8 @@ def _detect_rollover(
                     )
                 return False
 
+        return False
+
     def add_empty_log(self, address: int, slot: int) -> None:
         """Add empty energy log record to mark any start of beginning of energy log collection."""
         recalculate = False
@@ -439,11 +443,12 @@ def add_log(
             self.recalculate_missing_log_addresses()
 
         _LOGGER.debug(
-            "add_log | pulses=%s | address=%s | slot= %s |time:%s",
+            "add_log | pulses=%s | address=%s | slot=%s | time=%s, direction=%s",
             pulses,
             address,
             slot,
             timestamp,
+            direction,
         )
         return True
 
@@ -504,37 +509,70 @@ def _update_log_direction(
         if self._logs is None:
             return
 
+        prev_timestamp = self._check_prev_production(address, slot, timestamp)
+        next_timestamp = self._check_next_production(address, slot, timestamp)
+        if self._first_prev_log_processed and self._first_next_log_processed:
+            # _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 _check_prev_production(
+        self, address: int, slot: int, timestamp: datetime
+    ) -> datetime | None:
+        """Check the previous slot for production pulses."""
         prev_address, prev_slot = calc_log_address(address, slot, -1)
         if self._log_exists(prev_address, prev_slot):
-            if self._logs[prev_address][prev_slot].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
+            prev_timestamp = self._logs[prev_address][prev_slot].timestamp
+            if not self._first_prev_log_processed:
+                self._first_prev_log_processed = 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
+            return prev_timestamp
+
+        if self._first_prev_log_processed:
+            self._first_prev_log_processed = False
+            return None
 
+    def _check_next_production(
+        self, address: int, slot: int, timestamp: datetime
+    ) -> datetime | None:
+        """Check the next slot for production pulses."""
         next_address, next_slot = calc_log_address(address, slot, 1)
         if self._log_exists(next_address, next_slot):
-            if self._logs[next_address][next_slot].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
+            next_timestamp = self._logs[next_address][next_slot].timestamp
+            if not self._first_next_log_processed:
+                self._first_next_log_processed = 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
+            return next_timestamp
+
+        if self._first_next_log_processed:
+            self._first_next_log_processed = False
+            return None
 
     def _update_log_interval(self) -> None:
         """Update the detected log interval based on the most recent two logs."""

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name            = "plugwise_usb"
-version         = "v0.40.1b1"
+version         = "v0.41.0"
 license         = "MIT"
 keywords        = ["home", "automation", "plugwise", "module", "usb"]
 classifiers     = [

scripts/tests_and_coverage.sh

@@ -23,7 +23,8 @@ set +u
 
 if [ -z "${GITHUB_ACTIONS}" ] || [ "$1" == "test_and_coverage" ] ; then
     # Python tests (rerun with debug if failures)
-    PYTHONPATH=$(pwd) pytest -qx tests/ --cov='.' --no-cov-on-fail --cov-report term-missing || PYTHONPATH=$(pwd) pytest -xrpP --log-level debug tests/
+    # PYTHONPATH=$(pwd) pytest -qx tests/ --cov='.' --no-cov-on-fail --cov-report term-missing || 
+    PYTHONPATH=$(pwd) pytest -xrpP --log-level debug tests/
 fi
 
 if [ -z "${GITHUB_ACTIONS}" ] || [ "$1" == "linting" ] ; then

tests/test_usb.py

@@ -1218,15 +1218,15 @@ def test_pulse_collection_production(self, monkeypatch: pytest.MonkeyPatch) -> N
         # Test consumption & production - Log import #1 - production
         # Missing addresses can not be determined yet
         test_timestamp = fixed_this_hour - td(hours=1)
-        tst_production.add_log(200, 2, test_timestamp, 2000)
+        tst_production.add_log(200, 2, test_timestamp, -2000)
         assert tst_production.log_addresses_missing is None
         assert tst_production.production_logging is None
 
         # Test consumption & production - Log import #2 - consumption
         # production must be enabled & intervals are unknown
         # Log at address 200 is known and expect production logs too
         test_timestamp = fixed_this_hour - td(hours=1)
-        tst_production.add_log(200, 1, test_timestamp, 1000)
+        tst_production.add_log(200, 1, test_timestamp, 0)
         assert tst_production.log_addresses_missing is None
         assert tst_production.log_interval_consumption is None
         assert tst_production.log_interval_production is None
@@ -1235,7 +1235,7 @@ def test_pulse_collection_production(self, monkeypatch: pytest.MonkeyPatch) -> N
         # Test consumption & production - Log import #3 - production
         # Interval of consumption is not yet available
         test_timestamp = fixed_this_hour - td(hours=2)  # type: ignore[unreachable]
-        tst_production.add_log(199, 4, test_timestamp, 4000)
+        tst_production.add_log(199, 4, test_timestamp, -2200)
         missing_check = list(range(199, 157, -1))
         assert tst_production.log_addresses_missing == missing_check
         assert tst_production.log_interval_consumption is None
@@ -1245,32 +1245,32 @@ def test_pulse_collection_production(self, monkeypatch: pytest.MonkeyPatch) -> N
         # Test consumption & production - Log import #4
         # Interval of consumption is available
         test_timestamp = fixed_this_hour - td(hours=2)
-        tst_production.add_log(199, 3, test_timestamp, 3000)
+        tst_production.add_log(199, 3, test_timestamp, 0)
         assert tst_production.log_addresses_missing == missing_check
         assert tst_production.log_interval_consumption == 60
         assert tst_production.log_interval_production == 60
         assert tst_production.production_logging
 
         pulse_update_1 = fixed_this_hour + td(minutes=5)
-        tst_production.update_pulse_counter(100, 50, pulse_update_1)
+        tst_production.update_pulse_counter(0, -500, pulse_update_1)
         assert tst_production.collected_pulses(
             fixed_this_hour, is_consumption=True
-        ) == (100, pulse_update_1)
+        ) == (0, pulse_update_1)
         assert tst_production.collected_pulses(
             fixed_this_hour, is_consumption=False
-        ) == (50, pulse_update_1)
+        ) == (500, pulse_update_1)
         assert tst_production.collected_pulses(
             fixed_this_hour - td(hours=1), is_consumption=True
-        ) == (100, pulse_update_1)
+        ) == (0, pulse_update_1)
         assert tst_production.collected_pulses(
             fixed_this_hour - td(hours=2), is_consumption=True
-        ) == (1000 + 100, pulse_update_1)
+        ) == (0 + 0, pulse_update_1)
         assert tst_production.collected_pulses(
             fixed_this_hour - td(hours=1), is_consumption=False
-        ) == (50, pulse_update_1)
+        ) == (500, pulse_update_1)
         assert tst_production.collected_pulses(
             fixed_this_hour - td(hours=2), is_consumption=False
-        ) == (2000 + 50, pulse_update_1)
+        ) == (2000 + 500, pulse_update_1)
 
     _pulse_update = 0