tim
c2faeb669a
Decode River 3 separately and accumulate partial reports against a field-level checklist. Keep watch mode running after completion. Label fields and nested records from the pinned pr705 schema. Preserve unknown values and document unverified meanings and routing. Add synthetic regressions, the River 3 field reference, schema-generation tooling, and upstream attribution.
136 lines
5.5 KiB
Python
136 lines
5.5 KiB
Python
"""Backend import boundaries and compatibility, using synthetic packets only."""
|
|
|
|
from dataclasses import asdict
|
|
import json
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from bmspy import ecoflow_proto, ecoflow_river2, ecoflow_river3
|
|
from bmspy.ecoflow_snapshot import EcoFlowSnapshot
|
|
|
|
|
|
PD_PACKET = (2, 0x20, 2, bytes(14) + struct.pack("<BHHi", 47, 0, 0, 5999))
|
|
EMS_PACKET = (3, 0x20, 2, bytes(26) + struct.pack("<f", 87.5))
|
|
CELL_TEMP_PACKET = (2, 0xFE, 0x15, b"\x98\x10\x1d") # field 259: 29 C
|
|
COMPOSITE_TEMP_PACKET = (2, 0xFE, 0x15, b"\xb0\x06\x14") # field 102: 20 C
|
|
|
|
|
|
@pytest.mark.parametrize("first", [
|
|
"ecoflow_proto", "ecoflow_river2", "ecoflow_river3", "ecoflow_snapshot",
|
|
])
|
|
def test_decoders_import_in_any_order_without_transport_dependencies(first):
|
|
# A fresh interpreter catches import cycles hidden by pytest's import order.
|
|
script = """
|
|
import importlib
|
|
import importlib.abc
|
|
import sys
|
|
|
|
class NoTransportImports(importlib.abc.MetaPathFinder):
|
|
def find_spec(self, fullname, path=None, target=None):
|
|
if fullname.split('.')[0] in {'serial', 'bleak', 'ecdsa', 'Crypto'}:
|
|
raise AssertionError('Unexpected optional dependency: ' + fullname)
|
|
if fullname in {'bmspy.ecoflow_bt', 'bmspy.jbd_bms'}:
|
|
raise AssertionError('Unexpected transport import: ' + fullname)
|
|
|
|
sys.meta_path.insert(0, NoTransportImports())
|
|
importlib.import_module('bmspy.' + sys.argv[1])
|
|
from bmspy import ecoflow_proto, ecoflow_river2, ecoflow_river3
|
|
from bmspy.ecoflow_snapshot import EcoFlowSnapshot
|
|
|
|
assert ecoflow_proto.format_payload(2, 0xFE, 0x15, b'\\x08\\x00') == [
|
|
'[DisplayPropertyUpload]', ' errcode: 0',
|
|
]
|
|
assert ecoflow_proto.format_payload(2, 0x20, 2, bytes(23))[0] == '[PD heartbeat]'
|
|
assert ecoflow_river2.decode_river2_payload(2, 0x20, 2, bytes(23))['soc'] == 0
|
|
assert ecoflow_river3.decode_river3_payload(2, 0xFE, 0x15, b'\\x08\\x00') == {}
|
|
assert list(EcoFlowSnapshot().items()) == []
|
|
"""
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", script, first],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
|
|
|
|
def test_snapshot_and_decoder_imports_remain_compatible():
|
|
from bmspy import ecoflow_bt
|
|
|
|
assert ecoflow_bt.EcoFlowSnapshot is EcoFlowSnapshot
|
|
assert ecoflow_bt.decode_river2_payload is ecoflow_river2.decode_river2_payload
|
|
assert ecoflow_bt.decode_river3_payload is ecoflow_river3.decode_river3_payload
|
|
|
|
|
|
def test_diagnostic_formatting_aliases_remain_compatible():
|
|
assert ecoflow_proto.decode_payload is ecoflow_proto.format_payload
|
|
assert ecoflow_proto.decode_protobuf is ecoflow_proto.format_protobuf
|
|
assert ecoflow_proto.decode_schema_protobuf is ecoflow_proto.format_schema_protobuf
|
|
assert ecoflow_proto.format_protobuf(b"\x08\x00") == ["field 1: 0"]
|
|
assert ecoflow_proto.format_schema_protobuf(b"\x08\x00", "DisplayPropertyUpload") == [
|
|
"errcode: 0",
|
|
]
|
|
assert ecoflow_proto.format_payload(2, 0xFE, 0x15, b"") is None
|
|
assert ecoflow_proto.format_payload(0x35, 0x35, 0x20, b"\x00") is None
|
|
assert ecoflow_proto.describe_message(2, 0x20, 2) == "PD heartbeat"
|
|
assert ecoflow_proto.describe_message(2, 0xFE, 0x15) == "DisplayPropertyUpload"
|
|
assert ecoflow_proto.describe_message(0x35, 0x35, 0x20) is None
|
|
|
|
|
|
@pytest.mark.parametrize("model,packets", [
|
|
(ecoflow_river2, [PD_PACKET, EMS_PACKET, CELL_TEMP_PACKET, (2, 0x20, 2, b"")]),
|
|
(ecoflow_river3, [CELL_TEMP_PACKET, COMPOSITE_TEMP_PACKET, PD_PACKET, (2, 0xFE, 0x15, b"\x80")]),
|
|
])
|
|
def test_legacy_numeric_decoder_entry_points(model, packets):
|
|
name = "decode_" + model.__name__.rsplit("_", 1)[1] + "_payload"
|
|
legacy = getattr(ecoflow_proto, name)
|
|
direct = getattr(model, name)
|
|
for packet in packets:
|
|
assert legacy(*packet) == direct(*packet)
|
|
|
|
|
|
@pytest.mark.parametrize("decoder_type,high_packet,low_packet,field,high,low", [
|
|
(ecoflow_river2.River2Decoder, EMS_PACKET, PD_PACKET, "bms_capacity_charge_ratio", 0.875, 0.47),
|
|
(ecoflow_river3.River3Decoder, CELL_TEMP_PACKET, COMPOSITE_TEMP_PACKET, "bms_temperature_celcius", 29, 20),
|
|
], ids=["river2_soc", "river3_temperature"])
|
|
def test_measurement_priorities_are_per_instance_and_reset_per_connection(
|
|
decoder_type, high_packet, low_packet, field, high, low,
|
|
):
|
|
def value(snapshot):
|
|
measurement = getattr(snapshot, field)
|
|
return measurement.raw_values[1] if hasattr(measurement, "raw_values") else measurement.raw_value
|
|
|
|
decoder = decoder_type()
|
|
snapshot = EcoFlowSnapshot()
|
|
decoder.update(snapshot, *high_packet)
|
|
decoder.update(snapshot, *low_packet)
|
|
assert value(snapshot) == high
|
|
|
|
other = decoder_type()
|
|
other_snapshot = EcoFlowSnapshot()
|
|
other.update(other_snapshot, *low_packet)
|
|
assert value(other_snapshot) == low
|
|
assert value(snapshot) == high
|
|
|
|
before = asdict(snapshot)
|
|
decoder.reset()
|
|
assert asdict(snapshot) == before
|
|
decoder.update(snapshot, *low_packet)
|
|
assert value(snapshot) == low
|
|
|
|
|
|
def test_snapshot_keeps_diagnostic_packets_and_json_round_trip():
|
|
from bmspy.classes import UPS
|
|
|
|
snapshot = EcoFlowSnapshot()
|
|
ecoflow_river2.River2Decoder().update(snapshot, *PD_PACKET)
|
|
snapshot.record_packet(*PD_PACKET)
|
|
items = dict(snapshot.items())
|
|
assert items["bms_capacity_charge_ratio"].raw_value == 0.47
|
|
assert items["ef_pkt_02_20_02"].info == PD_PACKET[3].hex()
|
|
assert "ef_remaining_time_minutes" not in items
|
|
wire_data = {name: asdict(field) for name, field in snapshot.items()}
|
|
restored = UPS.from_dict(json.loads(json.dumps(wire_data)))
|
|
assert dict(restored.items()) == items
|