tim
886991caf3
Decode River 2 charge, temperature, power flow, and time estimates. Preserve precise charge readings across packet reordering and omit unavailable time estimates. Require valid PD, EMS, and BMS core heartbeats for diagnostic completion. Report optional packets separately and support raw watch output with reliable disconnect cleanup. Use synthetic regression tests and keep captures, credentials, and external keydata private. Device details come from command-line arguments or discovery. This commit retains Poetry; the separate packaging commit migrates to uv.
242 lines
11 KiB
Python
242 lines
11 KiB
Python
"""Synthetic River 2 diagnostics. No device captures or private identifiers."""
|
|
|
|
import asyncio
|
|
import importlib.util
|
|
import struct
|
|
from argparse import Namespace
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, Mock
|
|
|
|
import pytest
|
|
|
|
from bmspy.ecoflow_bt import EcoFlowBTDevice, EcoFlowPacket, _State
|
|
|
|
ADDRESS = "02:00:00:00:00:03"
|
|
|
|
@pytest.fixture
|
|
def script():
|
|
path = Path(__file__).resolve().parents[1] / "tools" / "ecoflow-bt.py"
|
|
spec = importlib.util.spec_from_file_location("ecoflow_live_diagnostics", path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def device(prefix="R601"):
|
|
return EcoFlowBTDevice(ADDRESS, prefix + "TEST00000001", "test-user")
|
|
|
|
|
|
def attach(script, dev, watch=False):
|
|
collector = script.PacketCollector(ADDRESS, script.MODEL_PROFILES[dev.serial_number[:4]], watch)
|
|
dev.on_packet(collector.receive)
|
|
return collector
|
|
|
|
|
|
def river2_bypass_packets():
|
|
"""Synthetic first burst: three core heartbeats, MPPT, and an empty reply."""
|
|
pd = bytearray(121)
|
|
struct.pack_into("<BHHi", pd, 14, 88, 35, 35, 5218)
|
|
ems = bytearray(46)
|
|
struct.pack_into("<II", ems, 17, 5999, 5999)
|
|
struct.pack_into("<f", ems, 26, 87.5)
|
|
bms = bytearray(69)
|
|
bms[11], bms[20] = 88, 28
|
|
struct.pack_into("<I", bms, 12, 13)
|
|
struct.pack_into("<fIII", bms, 53, 87.75, 0, 0, 0)
|
|
return [
|
|
EcoFlowPacket(2, 0x21, 0x20, 2, bytes(pd), version=2),
|
|
EcoFlowPacket(0x35, 0x21, 1, 0x52, b"", version=2),
|
|
EcoFlowPacket(3, 0x21, 0x20, 2, bytes(ems), version=2),
|
|
EcoFlowPacket(3, 0x21, 0x20, 0x32, bytes(bms), version=2),
|
|
EcoFlowPacket(5, 0x21, 0x20, 2, bytes(94), version=2),
|
|
]
|
|
|
|
|
|
def test_river2_finishes_on_first_heartbeat_set_with_same_snapshot_format(script, monkeypatch, capsys):
|
|
dev = device("R601")
|
|
# Include optional INV/MPPT packets in the initial burst as well as the three
|
|
# core heartbeats. Normal River 2 output and required packet types stay intact.
|
|
for src, cmd_id in [(2, 2), (3, 2), (3, 0x32), (4, 2), (5, 2)]:
|
|
dev._record(EcoFlowPacket(src, 0x21, 0x20, cmd_id, bytes(128), version=2))
|
|
expected = script.MODEL_PROFILES["R601"]["expected_fields"]
|
|
sleep = AsyncMock(side_effect=AssertionError("must not wait"))
|
|
monkeypatch.setattr(script.asyncio, "sleep", sleep)
|
|
asyncio.run(script.collect([(ADDRESS, dev, expected)], max_wait=120))
|
|
script.print_snapshot("River 2", ADDRESS, dev.snapshot, [2, 3, 4, 5])
|
|
sleep.assert_not_awaited()
|
|
out = capsys.readouterr().out
|
|
assert "3/3 required heartbeats received (PD, EMS, BMS)" in out
|
|
assert "Optional heartbeat types seen: INV, MPPT. Not seen: none (not required)" in out
|
|
for message in ("PD heartbeat", "EMS heartbeat", "BMS heartbeat", "INV heartbeat", "MPPT heartbeat"):
|
|
assert message in out
|
|
assert "latest fields collected" not in out
|
|
|
|
|
|
@pytest.mark.parametrize("prefix", ["R601", "R603"])
|
|
def test_river2_bypass_burst_without_inverter_finishes_immediately(script, monkeypatch, capsys, prefix):
|
|
dev = device(prefix)
|
|
collector = attach(script, dev)
|
|
for packet in river2_bypass_packets():
|
|
dev._record(packet)
|
|
expected = script.MODEL_PROFILES[prefix]["expected_fields"]
|
|
sleep = AsyncMock(side_effect=AssertionError("must not wait for optional INV"))
|
|
monkeypatch.setattr(script.asyncio, "sleep", sleep)
|
|
asyncio.run(script.collect([(ADDRESS, dev, expected)], max_wait=120, collectors={ADDRESS: collector}))
|
|
script.print_snapshot("River 2", ADDRESS, dev.snapshot, [2, 3, 4, 5])
|
|
sleep.assert_not_awaited()
|
|
out = capsys.readouterr().out
|
|
assert "3/3 required heartbeats received (PD, EMS, BMS)" in out
|
|
assert "Optional heartbeat types seen: MPPT. Not seen: INV (not required)" in out
|
|
assert "Other packet types seen: ef_pkt_35_01_52 (empty payload)" in out
|
|
assert "5 packet type(s)" not in out
|
|
assert "[INV heartbeat]" not in out
|
|
for message in ("PD heartbeat", "EMS heartbeat", "BMS heartbeat", "MPPT heartbeat"):
|
|
assert message in out
|
|
snap = dev.snapshot
|
|
assert snap.ef_packets_received.raw_value == 5
|
|
assert snap.bms_capacity_charge_ratio.raw_value == pytest.approx(0.8775)
|
|
assert snap.bms_temperature_celcius.raw_values == {1: 28}
|
|
assert snap.bms_charge_is_charging.raw_value is False
|
|
assert snap.bms_charge_is_discharging.raw_value is False
|
|
assert snap.ef_input_power_watts.raw_value == snap.ef_output_power_watts.raw_value == 35
|
|
assert snap.ef_remaining_time_minutes.raw_value == 5218
|
|
assert snap.ef_charge_remaining_time_minutes is None
|
|
assert snap.ef_discharge_remaining_time_minutes is None
|
|
assert "chg_remain_time: 5999 min" in out
|
|
assert "dsg_remain_time: 5999 min" in out
|
|
|
|
|
|
@pytest.mark.parametrize("index,minimum,label", [(0, 23, "PD"), (2, 25, "EMS"), (3, 21, "BMS")])
|
|
@pytest.mark.parametrize("empty", [False, True])
|
|
@pytest.mark.parametrize("listen", [False, True])
|
|
def test_invalid_river2_heartbeat_cannot_complete_collection(
|
|
script, capsys, index, minimum, label, empty, listen,
|
|
):
|
|
dev = device("R601")
|
|
collectors = {ADDRESS: attach(script, dev)} if listen else {}
|
|
packets = river2_bypass_packets()
|
|
valid = packets[index]
|
|
packets[index] = EcoFlowPacket(valid.src, valid.dst, valid.cmd_set, valid.cmd_id,
|
|
b"" if empty else valid.payload[:minimum - 1], version=2)
|
|
for packet in packets:
|
|
dev._record(packet)
|
|
expected = script.MODEL_PROFILES["R601"]["expected_fields"]
|
|
asyncio.run(script.collect([(ADDRESS, dev, expected)], max_wait=0, collectors=collectors))
|
|
out = capsys.readouterr().out
|
|
assert "timed out, incomplete" in out
|
|
assert f"{label} (ef_pkt_" in out
|
|
assert "3/3 required heartbeats received" not in out
|
|
dev._record(valid)
|
|
asyncio.run(script.collect([(ADDRESS, dev, expected)], max_wait=0, collectors=collectors))
|
|
assert "3/3 required heartbeats received" in capsys.readouterr().out
|
|
|
|
|
|
def test_river2_valid_heartbeat_stays_received_after_a_later_bad_packet(script, capsys):
|
|
dev = device("R601")
|
|
collector = attach(script, dev)
|
|
packets = river2_bypass_packets()
|
|
dev._record(packets[0])
|
|
dev._record(EcoFlowPacket(2, 0x21, 0x20, 2, b"", version=2))
|
|
for packet in packets[1:]:
|
|
dev._record(packet)
|
|
expected = script.MODEL_PROFILES["R601"]["expected_fields"]
|
|
asyncio.run(script.collect([(ADDRESS, dev, expected)], max_wait=0, collectors={ADDRESS: collector}))
|
|
assert "3/3 required heartbeats received" in capsys.readouterr().out
|
|
|
|
|
|
def test_river2_accepts_zero_values_and_minimum_valid_core_lengths(script, capsys):
|
|
dev = device("R601")
|
|
for src, cmd_id, size in [(2, 2, 23), (3, 2, 25), (3, 0x32, 21)]:
|
|
dev._record(EcoFlowPacket(src, 0x21, 0x20, cmd_id, bytes(size), version=2))
|
|
asyncio.run(script.collect([(ADDRESS, dev, script.MODEL_PROFILES["R601"]["expected_fields"])], max_wait=0))
|
|
out = capsys.readouterr().out
|
|
assert "3/3 required heartbeats received" in out
|
|
assert "Not seen: INV, MPPT (not required)" in out
|
|
|
|
|
|
def test_packet_listeners_receive_duplicates_after_snapshot_updates():
|
|
dev = device()
|
|
observed = []
|
|
dev.on_packet(Mock(side_effect=ValueError("bad listener")))
|
|
dev.on_packet(lambda pkt: observed.append((pkt, dev.snapshot.ef_input_power_watts.raw_value)))
|
|
updated = Mock()
|
|
dev.on_update(updated)
|
|
pkt = river2_bypass_packets()[0]
|
|
dev._record(pkt)
|
|
dev._record(pkt)
|
|
assert observed == [(pkt, 35), (pkt, 35)]
|
|
assert updated.call_count == 2
|
|
|
|
|
|
def test_river2_watch_keeps_late_unknown_packets_after_completion(script, monkeypatch, capsys):
|
|
dev = device()
|
|
dev._state = _State.AUTHENTICATED
|
|
collector = attach(script, dev, watch=True)
|
|
for packet in river2_bypass_packets():
|
|
dev._record(packet)
|
|
calls = 0
|
|
|
|
async def receive_later(_):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1:
|
|
dev._record(EcoFlowPacket(0x35, 0x21, 0x35, 0x20, b"synthetic-unknown"))
|
|
else:
|
|
raise asyncio.CancelledError()
|
|
|
|
monkeypatch.setattr(script.asyncio, "sleep", receive_later)
|
|
expected = script.MODEL_PROFILES["R601"]["expected_fields"]
|
|
with pytest.raises(asyncio.CancelledError):
|
|
asyncio.run(script.collect([(ADDRESS, dev, expected)], max_wait=0,
|
|
collectors={ADDRESS: collector}, watch=True))
|
|
out = capsys.readouterr().out
|
|
assert calls == 2
|
|
assert "3/3 required heartbeats received" in out
|
|
assert "continuing watch for optional or unknown information" in out
|
|
assert "raw_payload: " + b"synthetic-unknown".hex() in out
|
|
assert "timed out" not in out
|
|
|
|
|
|
@pytest.mark.parametrize("failure", [None, "cancel", "output", "disconnect"])
|
|
def test_river2_run_listens_before_connect_and_always_disconnects(script, monkeypatch, capsys, failure):
|
|
dev = device()
|
|
|
|
async def connect(**_):
|
|
dev._state = _State.AUTHENTICATED
|
|
packets = river2_bypass_packets()
|
|
dev._record(packets[0])
|
|
# Valid receipt must survive a later bad packet even before collect starts.
|
|
dev._record(EcoFlowPacket(2, 0x21, 0x20, 2, b"", version=2))
|
|
for packet in packets[1:]:
|
|
dev._record(packet)
|
|
if failure == "disconnect":
|
|
dev._state = _State.DISCONNECTED
|
|
|
|
manager = Mock(add_device=Mock(return_value=dev), connect_all=AsyncMock(side_effect=connect),
|
|
disconnect_all=AsyncMock())
|
|
monkeypatch.setattr(script, "EcoFlowBTManager", Mock(return_value=manager))
|
|
if failure == "cancel":
|
|
monkeypatch.setattr(script.asyncio, "sleep", AsyncMock(side_effect=asyncio.CancelledError()))
|
|
if failure == "output":
|
|
monkeypatch.setattr(script.PacketCollector, "_receive", Mock(side_effect=OSError("output failed")))
|
|
args = Namespace(scan=False, device=ADDRESS, serial_number=dev.serial_number,
|
|
debug=0, timeout=0, watch=bool(failure))
|
|
if failure:
|
|
error = {"cancel": asyncio.CancelledError, "output": OSError, "disconnect": ConnectionError}[failure]
|
|
with pytest.raises(error):
|
|
asyncio.run(script.run("test-user", args))
|
|
else:
|
|
asyncio.run(script.run("test-user", args))
|
|
assert "3/3 required heartbeats received" in capsys.readouterr().out
|
|
manager.disconnect_all.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["", "not-hex", None, 123])
|
|
def test_river2_snapshot_validation_rejects_invalid_hex(script, value):
|
|
collector = script.PacketCollector(ADDRESS, script.MODEL_PROFILES["R601"])
|
|
snapshot = Mock()
|
|
snapshot.items.return_value = [(key, Mock(info=value)) for key in script.RIVER2_REQUIRED_HEARTBEATS]
|
|
collector.observe_snapshot(snapshot)
|
|
assert len(collector.missing()) == 3
|
|
|