From 161ae7ee65b9b8ca5896ce31680a7a453019fa82 Mon Sep 17 00:00:00 2001 From: Timothy Allen Date: Sat, 10 Oct 2026 09:03:35 +0200 Subject: [PATCH] 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 | 6 +++ bmspy/ecoflow_bt.py | 59 +++++++++++++++++++-- bmspy/ecoflow_river2.py | 17 ++++++ bmspy/ecoflow_river3.py | 18 +++++++ bmspy/ecoflow_snapshot.py | 2 + docs/ecoflow-river2-fields.md | 6 +++ tests/test_backend_modules.py | 44 ++++++++++++++++ tests/test_ecoflow_bt.py | 99 +++++++++++++++++++++++++++++++++++ tools/ecoflow-bt.py | 9 ++-- 9 files changed, 249 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 1a3e675..8bf7af9 100644 --- a/README.md +++ b/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: diff --git a/bmspy/ecoflow_bt.py b/bmspy/ecoflow_bt.py index be14f1b..af56bf0 100644 --- a/bmspy/ecoflow_bt.py +++ b/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() diff --git a/bmspy/ecoflow_river2.py b/bmspy/ecoflow_river2.py index 02ae93e..6350dc7 100644 --- a/bmspy/ecoflow_river2.py +++ b/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( diff --git a/bmspy/ecoflow_river3.py b/bmspy/ecoflow_river3.py index f182ffd..9760b86 100644 --- a/bmspy/ecoflow_river3.py +++ b/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( diff --git a/bmspy/ecoflow_snapshot.py b/bmspy/ecoflow_snapshot.py index a7bfc0b..4362ffc 100644 --- a/bmspy/ecoflow_snapshot.py +++ b/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 diff --git a/docs/ecoflow-river2-fields.md b/docs/ecoflow-river2-fields.md index 407ae09..fb04505 100644 --- a/docs/ecoflow-river2-fields.md +++ b/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 diff --git a/tests/test_backend_modules.py b/tests/test_backend_modules.py index fe14b73..7993251 100644 --- a/tests/test_backend_modules.py +++ b/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(" 0x7F else 0)) + part = part >> 7 or None + return bytes(encoded) + (struct.pack("= 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" diff --git a/tools/ecoflow-bt.py b/tools/ecoflow-bt.py index fd47078..9c4b3d4 100644 --- a/tools/ecoflow-bt.py +++ b/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",