tim
f36bb2510c
Correct the spelling in the JBD and EcoFlow snapshots, the diagnostic tool, and the tests. UPS.from_dict maps the old name to the new name, so a client still reads data from an older server during an upgrade. InfluxDB and Prometheus now receive this measurement under the new name.
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_celsius.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
|
|
|