Add request deduplication support to Shelly emulator

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changelog
 
 ## Next
+- Added `DEDUPE_TIME_WINDOW` support to the Shelly emulator to drop burst-repeat requests from the same battery IP; the value can be set under `[GENERAL]` to apply regardless of which device type is emulated
 - Added opt-in web-based configuration editor (`WEB_CONFIG_ENABLED = True` in `[GENERAL]`) accessible at `http://<host>:52500/config`; supports editing all config sections and keys with type-appropriate inputs, comment preservation, and a Save & Restart button
 - **Breaking:** Rebrand project from "B2500 Meter" to "AstraMeter" (formerly b2500-meter). Package renamed to `astrameter`, CLI commands are now `astrameter` and `astra-sim`. Docker image moved from `ghcr.io/tomquist/b2500-meter` to `ghcr.io/tomquist/astrameter` (the legacy `ghcr.io/tomquist/b2500-meter` image is still published in parallel for backward compatibility). Home Assistant users must update their app repository URL to `https://github.com/tomquist/astrameter#main`.
 - Added CT002/CT003 emulation for steering multiple Marstek storage devices over the Marstek CT UDP protocol, with opt-in efficiency optimization that concentrates power on fewer batteries at low demand and rotates fairly over time (`MIN_EFFICIENT_POWER`, `EFFICIENCY_ROTATION_INTERVAL`, and related tuning options)

README.md

@@ -182,30 +182,32 @@ THROTTLE_INTERVAL = 0
 # doesn't add latency to every CT002 response. Default: true.
 # Can be overridden per powermeter section.
 #WAIT_FOR_NEXT_MESSAGE = true
+# Ignore repeated requests from the same emulator client within this window
+# (seconds). Applies to CT002/CT003 (keyed by consumer id) and Shelly (keyed
+# by battery IP). Can be overridden in the [CT002]/[CT003] section. 0 disables.
+#DEDUPE_TIME_WINDOW = 0
 ```
 
-Per-powermeter options (apply in any powermeter section, e.g. `[TASMOTA]` or `[HOMEASSISTANT]`):
+Per-powermeter options (apply in any powermeter section, e.g. `[TASMOTA]` or `[HOMEASSISTANT]`, or globally under `[GENERAL]`):
 - **THROTTLE_INTERVAL** — Override global throttling for this powermeter
 - **WAIT_FOR_NEXT_MESSAGE** — Override the global wait-for-fresh-push behaviour
   for this powermeter (set to `false` to opt out of the wait entirely)
+- **SMOOTH_TARGET_ALPHA** (default 0 = disabled) — EMA factor for the powermeter
+  reading in (0, 1]. Higher values track load changes faster; lower values filter
+  noise but add lag. Values close to 1.0 work well when the powermeter updates at
+  ≥ 1 Hz; reduce toward 0.3 if it updates significantly slower than 1 Hz.
+- **MAX_SMOOTH_STEP** (default 0 = unlimited) — Maximum watts the smoothed reading
+  may change per request cycle when `SMOOTH_TARGET_ALPHA` is active. Acts as a
+  slew-rate limit.
+- **DEADBAND** (default 0 = disabled, W) — When the absolute reading is below this
+  value, the wrapper emits zeros instead of chasing noise. Keeps batteries from
+  hunting around the zero-crossing; 10–30 W is a sensible range.
 
 CT002/CT003 active-steering options (all under `[CT002]` or `[CT003]`):
 - **ACTIVE_CONTROL** — When true (default), the emulator smooths the grid reading, splits
   the target across batteries, and balances their load.
   When false, the emulator relays raw meter values and batteries decide on their own.
 
-*Smoothing — how fast the emulator tracks grid changes:*
-- **SMOOTH_TARGET_ALPHA** (default 0.9) — EMA factor for the grid reading. Higher values
-  track load changes faster; lower values filter noise but add lag. The battery's own ramp
-  rate already filters noise, so values close to 1.0 work well when the powermeter updates
-  at ≥ 1 Hz. Reduce toward 0.3 if your powermeter updates significantly slower than 1 Hz
-  (higher lag in readings needs heavier smoothing to avoid oscillation).
-- **DEADBAND** (default 20 W) — When the grid total is within ± this value, the smoothed
-  target decays toward zero instead of chasing noise. Keeps batteries from hunting around
-  the zero-crossing. 10–30 W is a sensible range; set to 0 to disable.
-- **MAX_SMOOTH_STEP** (default 0 = unlimited) — Maximum watts the smoothed target may
-  change per request cycle. Acts as a slew-rate limit. Rarely needed at high alpha.
-
 *Fair distribution — balancing load across multiple batteries:*
 - **FAIR_DISTRIBUTION** (default true) — Adjust each battery's target so they share the
   load evenly. Only matters with two or more batteries.
@@ -303,7 +305,9 @@ CT_MAC = 001122334455
 UDP_PORT = 12345
 # WiFi RSSI reported to the storage system
 WIFI_RSSI = -50
-# Ignore repeated requests from the same client within this window (seconds)
+# Ignore repeated requests from the same consumer within this window (seconds).
+# Also supported by the Shelly emulator (keyed by battery IP); set it under
+# [GENERAL] to apply regardless of the emulated device type.
 DEDUPE_TIME_WINDOW = 0
 # Forget consumers after this many seconds without updates (multi-consumer support)
 CONSUMER_TTL = 120

config.ini.example

@@ -26,6 +26,10 @@ THROTTLE_INTERVAL = 0
 # response than a fresh one. Defaults to true.
 # Can be overridden per powermeter section.
 #WAIT_FOR_NEXT_MESSAGE = true
+# Ignore repeated requests from the same emulator client within this window
+# (seconds). Applies to CT002/CT003 (keyed by consumer id) and Shelly (keyed
+# by battery IP). Can be overridden in the [CT002]/[CT003] section. 0 disables.
+#DEDUPE_TIME_WINDOW = 0
 
 #[CT002]
 ## CT type is derived from the emulated device (ct002 -> HME-4, ct003 -> HME-3).
@@ -38,7 +42,8 @@ THROTTLE_INTERVAL = 0
 #UDP_PORT = 12345
 ## WiFi RSSI reported to the storage system
 #WIFI_RSSI = -50
-## Ignore repeated requests from the same client within this window (seconds)
+## Ignore repeated requests from the same consumer within this window (seconds).
+## Overrides the [GENERAL] DEDUPE_TIME_WINDOW for this section.
 #DEDUPE_TIME_WINDOW = 0
 ## Forget consumers after this many seconds without updates (multi-consumer support)
 #CONSUMER_TTL = 120

ha_addon/config.yaml

@@ -56,5 +56,9 @@ schema:
   log_level: list(critical|error|warning|info|debug)
   power_offset: str?
   power_multiplier: str?
+  dedupe_time_window: float(0,)?
+  smooth_target_alpha: float(0,1)?
+  max_smooth_step: float(0,)?
+  deadband: float(0,)?
   custom_config: str?
   mqtt_uri: str?

ha_addon/run.sh

@@ -85,6 +85,9 @@ else
         echo "[GENERAL]"
         echo "DEVICE_TYPE=$(bashio::config 'device_types')"
         echo "THROTTLE_INTERVAL=$(bashio::config 'throttle_interval')"
+        if bashio::config.has_value 'dedupe_time_window'; then
+            echo "DEDUPE_TIME_WINDOW=$(bashio::config 'dedupe_time_window')"
+        fi
         echo "ENABLE_WEB_SERVER=true"
         echo ""
         if [ "$has_ct002" -eq 1 ] && [ "$has_ct003" -eq 1 ]; then
@@ -143,6 +146,15 @@ else
             power_multiplier="$(bashio::config 'power_multiplier' | tr -d '\r\n')"
             echo "POWER_MULTIPLIER=$power_multiplier"
         fi
+        if bashio::config.has_value 'smooth_target_alpha'; then
+            echo "SMOOTH_TARGET_ALPHA=$(bashio::config 'smooth_target_alpha')"
+        fi
+        if bashio::config.has_value 'max_smooth_step'; then
+            echo "MAX_SMOOTH_STEP=$(bashio::config 'max_smooth_step')"
+        fi
+        if bashio::config.has_value 'deadband'; then
+            echo "DEADBAND=$(bashio::config 'deadband')"
+        fi
 
         # Fetch this add-on's slug from the supervisor so MQTT discovery can
         # link discovered meter devices to the add-on device via_device.

ha_addon/translations/en.yaml

@@ -46,4 +46,16 @@ configuration:
     description: "Optional. Added to each power value after the multiplier is applied. Single value (e.g. -50) or a comma-separated list (one per phase, e.g. -50,-30,-40). Leave empty to disable."
   power_multiplier:
     name: Power Multiplier
-    description: "Optional. Scales each power value (formula: value * POWER_MULTIPLIER + POWER_OFFSET). Single value (e.g. 1.05) or a comma-separated list (one per phase, e.g. 1.05,1.02,1.03). Use -1 to flip polarity. Leave empty to disable."
+    description: "Optional. Scales each power value (formula: value * POWER_MULTIPLIER + POWER_OFFSET). Single value (e.g. 1.05) or a comma-separated list (one per phase, e.g. 1.05,1.02,1.03). Use -1 to flip polarity. Leave empty to disable."
+  dedupe_time_window:
+    name: Deduplication Time Window
+    description: "Optional. Seconds during which repeat requests from the same battery are ignored. Applies to both CT002/CT003 (keyed by consumer id) and Shelly (keyed by battery IP) emulators. Set to 0 or leave empty to disable."
+  smooth_target_alpha:
+    name: Smoothing Alpha (EMA)
+    description: "Optional. Exponential moving-average factor in (0, 1] applied to power readings from Home Assistant. Smaller values smooth more heavily (e.g. 0.1 = heavy smoothing, 0.5 = light). Leave empty to disable."
+  max_smooth_step:
+    name: Max Smoothing Step
+    description: "Optional. Caps the maximum watt change per cycle when smoothing is active. Use together with Smoothing Alpha to bound per-step jumps (e.g. 200). Leave empty or 0 for unlimited."
+  deadband:
+    name: Deadband (W)
+    description: "Optional. Readings with absolute value below this threshold are reported as 0 W to stop battery micro-cycling near zero demand (e.g. 5). Leave empty or 0 to disable."

src/astrameter/ct002/balancer.py

@@ -53,7 +53,6 @@ class BalancerConfig:
     error_reduce_threshold: float = 20
     max_correction_per_step: float = 80
     max_target_step: float = 0
-    deadband: float = 20
     min_efficient_power: float = 0
     probe_min_power: float = 80
     efficiency_rotation_interval: float = 900
@@ -74,7 +73,6 @@ def _clamp(name: str, lo: float, hi: float) -> None:
         _clamp("error_reduce_threshold", 0, float("inf"))
         _clamp("max_correction_per_step", 0, float("inf"))
         _clamp("max_target_step", 0, float("inf"))
-        _clamp("deadband", 0, float("inf"))
         _clamp("min_efficient_power", 0, float("inf"))
         _clamp("probe_min_power", 0, float("inf"))
         _clamp("efficiency_rotation_interval", 1, float("inf"))

src/astrameter/ct002/ct002.py

@@ -10,6 +10,7 @@
 from typing import Any
 
 from astrameter.config.logger import logger
+from astrameter.request_dedupe import RequestDeduplicator
 
 from .balancer import (
     SATURATION_GRACE_SECONDS,
@@ -109,7 +110,6 @@ def __init__(
         error_boost_max=0.5,
         error_reduce_threshold=20,
         balance_deadband=15,
-        deadband=20,
         max_correction_per_step=80,
         max_target_step=0,
         saturation_detection=True,
@@ -142,7 +142,12 @@ def __init__(
         self._device_id = device_id
         self._consumers: dict[str, Consumer] = {}
         self._info_idx_counter = 0
-        self._last_response_time: dict[tuple, float] = {}
+        # Use wall-clock (time.time) so the dedup shares a timebase with
+        # _cleanup_consumers' purge; RequestDeduplicator would otherwise
+        # default to time.monotonic and mix timebases across the class.
+        self._dedup: RequestDeduplicator[str] = RequestDeduplicator(
+            dedupe_time_window, clock=clock or time.time
+        )
         self._transport = None
         self._protocol: _CT002Protocol | None = None
         self._cleanup_task = None
@@ -168,7 +173,6 @@ def __init__(
                 error_reduce_threshold=error_reduce_threshold,
                 max_correction_per_step=max_correction_per_step,
                 max_target_step=max_target_step,
-                deadband=deadband,
                 min_efficient_power=min_efficient_power,
                 probe_min_power=probe_min_power,
                 efficiency_rotation_interval=efficiency_rotation_interval,
@@ -301,13 +305,7 @@ def _cleanup_consumers(self):
             self._call_event_listener(key, {"_removed": True})
             del self._consumers[key]
             self._balancer.remove_consumer(key)
-        stale_addrs = [
-            addr
-            for addr, ts in self._last_response_time.items()
-            if now - ts > self.dedupe_time_window
-        ]
-        for addr in stale_addrs:
-            self._last_response_time.pop(addr, None)
+        self._dedup.purge_older_than(self.consumer_ttl)
 
     def _consumer_mode(self, consumer_id: str | None) -> ConsumerMode:
         if not consumer_id:
@@ -565,13 +563,15 @@ async def _handle_request(self, data, addr, transport):
             " in inspection mode" if in_inspection_mode else "",
         )
 
-        # Deduplication check
-        current_time = time.time()
-        last_time = self._last_response_time.get(addr)
-        if last_time and (current_time - last_time) < self.dedupe_time_window:
-            logger.debug("Ignoring request from %s due to dedupe window", addr)
+        # Deduplication check (keyed by consumer id so repeats from the
+        # same battery are suppressed regardless of source UDP port).
+        if not self._dedup.should_process(consumer_id):
+            logger.debug(
+                "Ignoring request from %s (consumer=%s) due to dedupe window",
+                addr,
+                consumer_id,
+            )
             return
-        self._last_response_time[addr] = current_time
 
         meter_dev_type = fields[0] if len(fields) > 0 else ""
         self._update_consumer_report(

src/astrameter/ct002/ct002_test.py

@@ -0,0 +1,32 @@
+from astrameter.ct002.ct002 import CT002
+
+
+class FakeClock:
+    def __init__(self) -> None:
+        self.now = 0.0
+
+    def __call__(self) -> float:
+        return self.now
+
+
+def test_dedup_uses_consumer_id_key_and_injected_clock() -> None:
+    clock = FakeClock()
+    ct = CT002(dedupe_time_window=1.0, clock=clock)
+
+    # Same consumer within the window → dropped.
+    assert ct._dedup.should_process("consumer-A") is True
+    clock.now += 0.5
+    assert ct._dedup.should_process("consumer-A") is False
+
+    # Different consumers are independent, even within the window.
+    assert ct._dedup.should_process("consumer-B") is True
+
+    # After the window elapses, the same consumer is accepted again.
+    clock.now += 1.0
+    assert ct._dedup.should_process("consumer-A") is True
+
+
+def test_dedup_window_zero_disables() -> None:
+    ct = CT002(dedupe_time_window=0.0)
+    for _ in range(3):
+        assert ct._dedup.should_process("consumer-A") is True

src/astrameter/main.py

@@ -123,14 +123,18 @@ async def run_device(
 
     device: CT002 | Shelly
 
+    global_dedupe_time_window = cfg.getfloat(
+        "GENERAL", "DEDUPE_TIME_WINDOW", fallback=0.0
+    )
+
     if device_type in ["ct002", "ct003"]:
         ct_section = get_ct_section(device_type, cfg)
         ct_type = "HME-4" if device_type == "ct002" else "HME-3"
         ct_mac = cfg.get(ct_section, "CT_MAC", fallback="")
         ct_udp_port = cfg.getint(ct_section, "UDP_PORT", fallback=UDP_PORT)
         wifi_rssi = cfg.getint(ct_section, "WIFI_RSSI", fallback=-50)
         dedupe_time_window = cfg.getfloat(
-            ct_section, "DEDUPE_TIME_WINDOW", fallback=0.0
+            ct_section, "DEDUPE_TIME_WINDOW", fallback=global_dedupe_time_window
         )
         consumer_ttl = cfg.getint(ct_section, "CONSUMER_TTL", fallback=120)
         debug_status = cfg.getboolean(ct_section, "DEBUG_STATUS", fallback=False)
@@ -149,7 +153,6 @@ async def run_device(
             ct_section, "ERROR_REDUCE_THRESHOLD", fallback=20
         )
         balance_deadband = cfg.getint(ct_section, "BALANCE_DEADBAND", fallback=15)
-        deadband = cfg.getint(ct_section, "DEADBAND", fallback=20)
         max_correction_per_step = cfg.getint(
             ct_section, "MAX_CORRECTION_PER_STEP", fallback=80
         )
@@ -223,7 +226,6 @@ async def run_device(
             error_boost_max=error_boost_max,
             error_reduce_threshold=error_reduce_threshold,
             balance_deadband=balance_deadband,
-            deadband=deadband,
             max_correction_per_step=max_correction_per_step,
             max_target_step=max_target_step,
             saturation_detection=saturation_detection,
@@ -260,22 +262,42 @@ def _ct002_event_listener(dev_id, consumer_id, data):
     elif device_type == "shellypro3em_old":
         logger.debug("Shelly Pro 3EM Settings:")
         logger.debug(f"Device ID: {device_id}")
-        device = Shelly(powermeters=powermeters, device_id=device_id, udp_port=1010)
+        device = Shelly(
+            powermeters=powermeters,
+            device_id=device_id,
+            udp_port=1010,
+            dedupe_time_window=global_dedupe_time_window,
+        )
 
     elif device_type == "shellypro3em_new":
         logger.debug("Shelly Pro 3EM Settings:")
         logger.debug(f"Device ID: {device_id}")
-        device = Shelly(powermeters=powermeters, device_id=device_id, udp_port=2220)
+        device = Shelly(
+            powermeters=powermeters,
+            device_id=device_id,
+            udp_port=2220,
+            dedupe_time_window=global_dedupe_time_window,
+        )
 
     elif device_type == "shellyemg3":
         logger.debug("Shelly EM Gen3 Settings:")
         logger.debug(f"Device ID: {device_id}")
-        device = Shelly(powermeters=powermeters, device_id=device_id, udp_port=2222)
+        device = Shelly(
+            powermeters=powermeters,
+            device_id=device_id,
+            udp_port=2222,
+            dedupe_time_window=global_dedupe_time_window,
+        )
 
     elif device_type == "shellyproem50":
         logger.debug("Shelly Pro EM 50 Settings:")
         logger.debug(f"Device ID: {device_id}")
-        device = Shelly(powermeters=powermeters, device_id=device_id, udp_port=2223)
+        device = Shelly(
+            powermeters=powermeters,
+            device_id=device_id,
+            udp_port=2223,
+            dedupe_time_window=global_dedupe_time_window,
+        )
 
     else:
         raise ValueError(f"Unsupported device type: {device_type}")