Wait for the required EcoFlow reports in each collection

A Bluetooth collection listened for a fixed 10 seconds and returned the data that arrived. If a required report was late, the snapshot missed shared measurements without notice.

Each decoder now tracks the reports behind the shared measurements. A collection listens for 10 seconds, then until those reports arrive, for at most 30 seconds in total. The new ef_missing_measurements field names the reports that did not arrive. The server log shows one line each time that list changes.

The diagnostic tool now uses the River 2 heartbeat list from the decoder.

README.md

@@ -101,6 +101,12 @@ failed one. It reads a serial BMS every 5 seconds and retries a failed serial
 read after 1 second. Unavailable devices are omitted from the response. Only the
 first request after startup waits, until each device completes one read attempt.
 
+Each Bluetooth read listens for 10 seconds. If a required report is still
+missing, the read continues until the report arrives, for at most 30 seconds in
+total. A read that reaches this limit returns a partial snapshot. Its
+`ef_missing_measurements` field names the missing reports, and the server log
+shows one line each time that list changes.
+
 #### Diagnostics
 
 Stop the server before connecting directly. Scan and collect reports from both Rivers:

bmspy/ecoflow_bt.py

@@ -36,6 +36,11 @@ _LOGGER = logging.getLogger(__name__)
 
 MANUFACTURER_KEY = 0xB5B5
 
+# How often a collection checks for the required reports.
+_COMPLETION_POLL_SECONDS = 0.5
+# Missing reports of the last collection per address, to log changes only.
+_last_missing: dict[str, set[str]] = {}
+
 # Packet version and encryption type are independent. R601/R603 use V2
 # packets even when the BLE advertisement selects Type 7 encryption.
 def _default_packet_version(serial_number: str) -> int:
@@ -502,6 +507,26 @@ class EcoFlowBTDevice:
         except asyncio.TimeoutError:
             return False
 
+    async def wait_for_measurements(self, minimum: float, timeout: float) -> set[str]:
+        """Listen for *minimum* seconds, then until the required reports arrive.
+
+        Stop after *timeout* seconds in total, or when the connection ends, and
+        return the names of the reports still missing. A model without a
+        decoder has no known requirements, so only *minimum* applies.
+        """
+        loop = asyncio.get_running_loop()
+        start = loop.time()
+        await asyncio.sleep(minimum)
+        decoder = self._measurement_decoder
+        if decoder is None:
+            return set()
+        while decoder.missing() and self._state != _State.DISCONNECTED:
+            remaining = start + timeout - loop.time()
+            if remaining <= 0:
+                break
+            await asyncio.sleep(min(_COMPLETION_POLL_SECONDS, remaining))
+        return decoder.missing()
+
     async def disconnect(self) -> None:
         """Stop notifications and close the BLE connection."""
         if self._auth_fallback_task is not None:
@@ -978,7 +1003,8 @@ class EcoFlowBTConfig:
     user_id: str
     encrypt_type: int = 7
     packet_version: int | None = None
-    listen_duration: float = 10.0
+    listen_duration: float = 10.0  # Minimum listening time, for optional reports
+    listen_timeout: float = 30.0   # Limit for the required reports
 
 
 def initialise_device(
@@ -988,6 +1014,7 @@ def initialise_device(
     encrypt_type: int = 7,
     packet_version: int | None = None,
     listen_duration: float = 10.0,
+    listen_timeout: float = 30.0,
 ) -> EcoFlowBTConfig:
     """Return a device config for use with :func:`collect_data`.
 
@@ -1002,17 +1029,36 @@ def initialise_device(
         encrypt_type=encrypt_type,
         packet_version=packet_version,
         listen_duration=listen_duration,
+        listen_timeout=listen_timeout,
     )
 
 
+def _record_missing(address: str, snapshot: EcoFlowSnapshot, missing: set[str]) -> None:
+    """Expose the reports that a collection did not receive, and log changes."""
+    snapshot.ef_missing_measurements = BMSInfoField(
+        help="Required reports not received in this collection",
+        info=", ".join(sorted(missing)),
+    )
+    if missing != _last_missing.get(address, set()):
+        _last_missing[address] = missing
+        debugger(f"ecoflow_bt [{address}]: " + (
+            f"collection incomplete; missing {', '.join(sorted(missing))}" if missing
+            else "collection complete again"
+        ))
+
+
 def collect_data(cfg: EcoFlowBTConfig, debug: int = 0) -> EcoFlowSnapshot | bool:
     """Connect to an EcoFlow device, collect push data, disconnect, return snapshot.
 
     Analogous to :func:`bmspy.jbd_bms.collect_data`.  Each call opens a fresh
-    BLE connection, authenticates, waits *listen_duration* seconds for the
-    device to push telemetry via notifications, then disconnects.
+    BLE connection, authenticates, and listens to the telemetry that the device
+    pushes via notifications. It listens for *listen_duration* seconds, then
+    until the reports behind the shared measurements arrive, for at most
+    *listen_timeout* seconds in total. Then it disconnects.
 
     Returns the populated :class:`EcoFlowSnapshot`, or ``False`` on failure.
+    A snapshot from a collection that reached the limit is partial, and its
+    ``ef_missing_measurements`` field names the reports that did not arrive.
     """
     async def _run() -> EcoFlowSnapshot | bool:
         device = EcoFlowBTDevice(
@@ -1028,9 +1074,12 @@ def collect_data(cfg: EcoFlowBTConfig, debug: int = 0) -> EcoFlowSnapshot | bool
                 return False
             if not await device.wait_authenticated(timeout=30.0):
                 return False
-            await asyncio.sleep(cfg.listen_duration)
+            missing = await device.wait_for_measurements(cfg.listen_duration, cfg.listen_timeout)
             # Metadata alone is not a successful telemetry collection.
-            return device.snapshot if device._pkt_count else False
+            if not device._pkt_count:
+                return False
+            _record_missing(cfg.address, device.snapshot, missing)
+            return device.snapshot
         finally:
             await device.disconnect()
 

bmspy/ecoflow_river2.py

@@ -217,6 +217,13 @@ _CORE_MINIMUM_LENGTHS = {
     (0x03, 0x20, 0x02): 25,  # EMS through dsg_remain_time
     (0x03, 0x20, 0x32): 21,  # BMS through temp
 }
+# Heartbeats behind the shared measurements, by raw packet key.
+# PD: total power; EMS: time estimates; BMS: battery temperature and direction.
+REQUIRED_HEARTBEATS = {
+    "ef_pkt_02_20_02": "PD",
+    "ef_pkt_03_20_02": "EMS",
+    "ef_pkt_03_20_32": "BMS",
+}
 _PROTOCOL_NAMESPACES = {
     (0x02, 0x20, 0x02): "ef_river2_pd",
     (0x03, 0x20, 0x02): "ef_river2_ems",
@@ -304,6 +311,14 @@ class River2Decoder:
     def reset(self) -> None:
         """Reset source priorities before a new connection, not the snapshot."""
         self._soc_priority = 0
+        self._received: set[str] = set()
+
+    def missing(self) -> set[str]:
+        """Name the required heartbeats that this connection has not received."""
+        return {
+            f"{label} heartbeat" for key, label in REQUIRED_HEARTBEATS.items()
+            if key not in self._received
+        }
 
     def update(
         self, snapshot: EcoFlowSnapshot, src: int, cmd_set: int, cmd_id: int, payload: bytes,
@@ -312,6 +327,8 @@ class River2Decoder:
         values = decode_river2_payload(src, cmd_set, cmd_id, payload)
         if values is None:
             return
+        # Receipt counts, not the values: an unavailable estimate is still a report.
+        self._received.add(f"ef_pkt_{src:02x}_{cmd_set:02x}_{cmd_id:02x}")
 
         def scalar(name: str, help: str, value: int | float | bool, units: str | None = None):
             setattr(snapshot, name, BMSScalarField(

bmspy/ecoflow_river3.py

@@ -191,6 +191,14 @@ _RIVER3_TELEMETRY_WIRE_TYPES = {
 }
 
 
+# Reported fields behind the shared measurements. Reports are partial, so a
+# collection is complete when each group has one field. Either temperature will do.
+REQUIRED_FIELDS = (
+    ("cms_batt_soc",), ("pow_in_sum_w",), ("pow_out_sum_w",), ("pow_get_bms",),
+    ("bms_max_cell_temp", "cms_batt_temp"), ("cms_chg_rem_time",), ("cms_dsg_rem_time",),
+)
+
+
 def decode_river3_payload(
     src: int,
     cmd_set: int,
@@ -274,6 +282,14 @@ class River3Decoder:
     def reset(self) -> None:
         """Reset source priorities before a new connection, not the snapshot."""
         self._river3_temp_priority = 0
+        self._received: set[str] = set()
+
+    def missing(self) -> set[str]:
+        """Name the required fields that this connection has not received."""
+        return {
+            " or ".join(names) for names in REQUIRED_FIELDS
+            if not self._received.intersection(names)
+        }
 
     def update(
         self, snapshot: EcoFlowSnapshot, src: int, cmd_set: int, cmd_id: int, payload: bytes,
@@ -283,6 +299,8 @@ class River3Decoder:
         values = decode_river3_payload(src, cmd_set, cmd_id, payload)
         if values is None:
             return
+        # Receipt counts, not the values: an unavailable estimate is still a report.
+        self._received.update(values)
 
         def scalar(name: str, help: str, value: int | float | bool, units: str | None = None):
             setattr(snapshot, name, BMSScalarField(

bmspy/ecoflow_snapshot.py

@@ -21,6 +21,8 @@ class EcoFlowSnapshot(UPS):
     ef_bt_address:       BMSInfoField | None = None
     ef_connection_state: BMSInfoField | None = None
     ef_packets_received: BMSScalarField | None = None
+    # Empty text when the collection received every required report.
+    ef_missing_measurements: BMSInfoField | None = None
 
     bms_capacity_charge_ratio: BMSScalarField | None = None
     bms_temperature_celsius: BMSMultiField | None = None

docs/ecoflow-river2-fields.md

@@ -27,6 +27,12 @@ rules for interpreting the device reports.
 | `bms_charge_is_charging` | Whether battery-side power flows into the battery. |
 | `bms_charge_is_discharging` | Whether battery-side power flows out of the battery. |
 | `ef_charge_state` | Numeric device state code. Observed R601 values are 0 for idle, 1 for charging, and 2 for discharging. |
+| `ef_missing_measurements` | Text. Required reports that the server collection did not receive, or empty text for a complete collection. |
+
+The server collection requires the PD, EMS, and BMS heartbeats from River 2.
+From River 3 it requires battery charge, total input and output power, battery
+power, one battery temperature, and both time estimates. Receipt counts, not the
+value, so an unavailable estimate still completes the collection.
 
 Input power alone does not indicate battery charging. The device can supply an
 AC load without charging its battery. Do not apply River 2 state-code meanings

tests/test_backend_modules.py

@@ -152,6 +152,50 @@ def test_shared_power_and_temperature_measurements_are_floats(decoder_type, pack
     assert [type(value) for value in values] == [float, float, float]
 
 
+def test_river2_collection_is_complete_after_the_three_core_heartbeats():
+    decoder = ecoflow_river2.River2Decoder()
+    snapshot = EcoFlowSnapshot()
+    assert decoder.missing() == {"PD heartbeat", "EMS heartbeat", "BMS heartbeat"}
+    decoder.update(snapshot, *PD_PACKET)
+    decoder.update(snapshot, 4, 0x20, 2, bytes(40))  # INV is optional.
+    decoder.update(snapshot, 3, 0x20, 0x32, bytes(5))  # A short BMS heartbeat is rejected.
+    assert decoder.missing() == {"EMS heartbeat", "BMS heartbeat"}
+    # 5999 marks an unavailable estimate. Its heartbeat still counts as received.
+    decoder.update(snapshot, 3, 0x20, 2, bytes(17) + struct.pack("<II", 5999, 5999))
+    decoder.update(snapshot, 3, 0x20, 0x32, bytes(20) + b"\x1c")
+    assert snapshot.ef_charge_remaining_time_minutes is None
+    assert decoder.missing() == set()
+    decoder.reset()
+    assert len(decoder.missing()) == 3
+
+
+def test_river3_collection_is_complete_across_partial_reports():
+    def field(number, value):
+        tag = number << 3 | (5 if isinstance(value, float) else 0)
+        encoded = bytearray()
+        for part in (tag, None if isinstance(value, float) else value):
+            while part is not None:
+                encoded.append(part & 0x7F | (0x80 if part > 0x7F else 0))
+                part = part >> 7 or None
+        return bytes(encoded) + (struct.pack("<f", value) if isinstance(value, float) else b"")
+
+    decoder = ecoflow_river3.River3Decoder()
+    snapshot = EcoFlowSnapshot()
+    assert len(decoder.missing()) == len(ecoflow_river3.REQUIRED_FIELDS)
+    first = field(262, 50.0) + field(3, 244.0) + field(4, 39.0) + field(158, -39.0)
+    decoder.update(snapshot, 2, 0xFE, 0x15, first)
+    assert decoder.missing() == {
+        "bms_max_cell_temp or cms_batt_temp", "cms_chg_rem_time", "cms_dsg_rem_time",
+    }
+    # Either temperature will do, and an unavailable estimate still counts as received.
+    second = field(102, 20) + field(268, 0xFFFFFFFF) + field(269, 0xFFFFFFFF)
+    decoder.update(snapshot, 2, 0xFE, 0x15, second)
+    assert snapshot.ef_charge_remaining_time_minutes is None
+    assert decoder.missing() == set()
+    decoder.reset()
+    assert len(decoder.missing()) == len(ecoflow_river3.REQUIRED_FIELDS)
+
+
 def test_snapshot_keeps_diagnostic_packets_and_json_round_trip():
     from bmspy.classes import UPS
 

tests/test_ecoflow_bt.py

@@ -494,6 +494,7 @@ def test_collection_always_disconnects(monkeypatch, connect_ok, authenticated, p
     fake = Mock(
         connect=AsyncMock(return_value=connect_ok),
         wait_authenticated=AsyncMock(return_value=authenticated),
+        wait_for_measurements=AsyncMock(return_value=set()),
         disconnect=AsyncMock(), _pkt_count=packets,
     )
     monkeypatch.setattr("bmspy.ecoflow_bt.EcoFlowBTDevice", Mock(return_value=fake))
@@ -502,6 +503,104 @@ def test_collection_always_disconnects(monkeypatch, connect_ok, authenticated, p
     fake.disconnect.assert_awaited_once()
 
 
+def test_wait_ends_when_the_required_heartbeats_arrive(monkeypatch):
+    monkeypatch.setattr("bmspy.ecoflow_bt._COMPLETION_POLL_SECONDS", 0.01)
+
+    async def run():
+        dev = device()
+        dev._record(heartbeat(2, pd_payload()))
+        dev._record(heartbeat(3, ems_payload()))
+
+        async def late_bms():
+            await asyncio.sleep(0.03)
+            dev._record(heartbeat(3, bms_payload(), 0x32))
+
+        task = asyncio.create_task(late_bms())
+        loop = asyncio.get_running_loop()
+        start = loop.time()
+        # The late heartbeat extends the wait past the minimum, far below the limit.
+        missing = await dev.wait_for_measurements(0, 30)
+        await task
+        return missing, loop.time() - start
+
+    missing, elapsed = asyncio.run(run())
+    assert missing == set()
+    assert 0.03 <= elapsed < 5
+
+
+def test_wait_keeps_the_minimum_for_optional_reports(monkeypatch):
+    monkeypatch.setattr("bmspy.ecoflow_bt._COMPLETION_POLL_SECONDS", 0.01)
+
+    async def run():
+        dev = device()
+        for packet in (heartbeat(2, pd_payload()), heartbeat(3, ems_payload()), heartbeat(3, bms_payload(), 0x32)):
+            dev._record(packet)
+        loop = asyncio.get_running_loop()
+        start = loop.time()
+        missing = await dev.wait_for_measurements(0.05, 30)
+        return missing, loop.time() - start
+
+    missing, elapsed = asyncio.run(run())
+    assert missing == set()
+    assert elapsed >= 0.05
+
+
+@pytest.mark.parametrize("disconnected", [False, True])
+def test_wait_stops_at_the_limit_or_a_lost_connection(monkeypatch, disconnected):
+    monkeypatch.setattr("bmspy.ecoflow_bt._COMPLETION_POLL_SECONDS", 0.01)
+
+    async def run():
+        dev = device()
+        dev._record(heartbeat(2, pd_payload()))
+        if disconnected:
+            dev._on_ble_disconnect(None)
+        loop = asyncio.get_running_loop()
+        start = loop.time()
+        missing = await dev.wait_for_measurements(0, 30 if disconnected else 0.05)
+        return missing, loop.time() - start
+
+    missing, elapsed = asyncio.run(run())
+    assert missing == {"EMS heartbeat", "BMS heartbeat"}
+    assert elapsed < 5
+
+
+def test_wait_uses_only_the_minimum_for_a_model_without_a_decoder():
+    async def run():
+        return await device(serial="XXXXTEST00000001").wait_for_measurements(0, 30)
+
+    assert asyncio.run(run()) == set()
+
+
+def test_collection_reports_missing_reports_and_logs_changes(monkeypatch, capsys):
+    from bmspy import ecoflow_bt
+
+    monkeypatch.setattr(ecoflow_bt, "_last_missing", {})
+    results = [{"EMS heartbeat", "BMS heartbeat"}, {"EMS heartbeat", "BMS heartbeat"}, set()]
+    snapshots = []
+
+    def fake_device(**kwargs):
+        snapshots.append(ecoflow_bt.EcoFlowSnapshot())
+        return Mock(
+            connect=AsyncMock(return_value=True), wait_authenticated=AsyncMock(return_value=True),
+            wait_for_measurements=AsyncMock(return_value=results[len(snapshots) - 1]),
+            disconnect=AsyncMock(), _pkt_count=1, snapshot=snapshots[-1],
+        )
+
+    monkeypatch.setattr(ecoflow_bt, "EcoFlowBTDevice", fake_device)
+    config = initialise_device(ADDRESS, SERIAL, "test-user", listen_duration=7, listen_timeout=20)
+    logs = []
+    for _ in results:
+        assert collect_data(config) is snapshots[-1]
+        logs.append(capsys.readouterr().out)
+    assert [snap.ef_missing_measurements.info for snap in snapshots] == [
+        "BMS heartbeat, EMS heartbeat", "BMS heartbeat, EMS heartbeat", "",
+    ]
+    # One line for each change, not one for each collection.
+    assert "collection incomplete; missing BMS heartbeat, EMS heartbeat" in logs[0]
+    assert logs[1] == ""
+    assert "collection complete again" in logs[2]
+
+
 @pytest.fixture
 def diagnostic_script():
     path = Path(__file__).resolve().parents[1] / "tools" / "ecoflow-bt.py"

tools/ecoflow-bt.py

@@ -28,7 +28,9 @@ try:
     from bmspy.ecoflow_proto import (
         format_payload, format_protobuf, describe_protobuf_field, parse_protobuf_fields,
     )
-    from bmspy.ecoflow_river2 import decode_river2_payload
+    from bmspy.ecoflow_river2 import (
+        REQUIRED_HEARTBEATS as RIVER2_REQUIRED_HEARTBEATS, decode_river2_payload,
+    )
 except ImportError as exc:
     sys.exit(f"Missing dependency: {exc}\n"
              "Run: uv sync --extra ecoflow-bt")
@@ -52,11 +54,6 @@ RIVER3_REQUIRED_GROUPS = {
     ),
 }
 
-RIVER2_REQUIRED_HEARTBEATS = {
-    "ef_pkt_02_20_02": "PD",
-    "ef_pkt_03_20_02": "EMS",
-    "ef_pkt_03_20_32": "BMS",
-}
 RIVER2_OPTIONAL_HEARTBEATS = {
     "ef_pkt_04_20_02": "INV",
     "ef_pkt_05_20_02": "MPPT",