tim
5fa6000c70
Move response decoding into jbd_proto and the data container into jbd_snapshot. Preserve the existing jbd_bms imports and test the transport boundary.
147 lines
6.0 KiB
Python
147 lines
6.0 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",
|
|
"jbd_proto", "jbd_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, jbd_proto
|
|
from bmspy.ecoflow_snapshot import EcoFlowSnapshot
|
|
from bmspy.jbd_snapshot import JBDBMS
|
|
|
|
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 jbd_proto.bytes_to_digits(1, 0) == 256
|
|
assert list(EcoFlowSnapshot().items()) == []
|
|
assert list(JBDBMS().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_jbd_function_imports_remain_compatible():
|
|
from bmspy import ecoflow_bt, jbd_bms, jbd_proto
|
|
from bmspy.jbd_snapshot import JBDBMS
|
|
|
|
assert ecoflow_bt.EcoFlowSnapshot is EcoFlowSnapshot
|
|
assert jbd_bms.JBDBMS is JBDBMS
|
|
for name in (
|
|
"bytes_to_digits", "bytes_to_date", "convert_to_signed",
|
|
"calculate_checksum", "verify_checksum", "parse_03_response", "parse_04_response",
|
|
):
|
|
assert getattr(jbd_bms, name) is getattr(jbd_proto, name)
|
|
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
|