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.
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_celsius", 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
|