From 5fa6000c70f2770cdc1f04758a49f3c575268bad Mon Sep 17 00:00:00 2001 From: Timothy Allen Date: Thu, 8 Oct 2026 14:09:05 +0200 Subject: [PATCH] Separate JBD decoding from serial transport Move response decoding into jbd_proto and the data container into jbd_snapshot. Preserve the existing jbd_bms imports and test the transport boundary. --- bmspy/jbd_bms.py | 431 ++-------------------------------- bmspy/jbd_proto.py | 376 +++++++++++++++++++++++++++++ bmspy/jbd_snapshot.py | 44 ++++ tests/test_backend_modules.py | 17 +- tests/test_jbd_bms.py | 6 +- 5 files changed, 454 insertions(+), 420 deletions(-) create mode 100644 bmspy/jbd_proto.py create mode 100644 bmspy/jbd_snapshot.py diff --git a/bmspy/jbd_bms.py b/bmspy/jbd_bms.py index 36627b9..455d497 100644 --- a/bmspy/jbd_bms.py +++ b/bmspy/jbd_bms.py @@ -1,54 +1,26 @@ #!/usr/bin/env python3 -# -# Communicate with a JBD/SZLLT BMS and return basic information -# +"""Serial transport and collection for JBD/SZLLT BMS devices. + +Response decoding lives in jbd_proto and measurement data in jbd_snapshot. +Their public names remain available here for existing callers. +""" + import atexit import serial import serial.rs485 import time -from collections.abc import Iterator -from dataclasses import dataclass, fields as dataclass_fields +from bmspy.jbd_proto import ( + bytes_to_date, + bytes_to_digits, + calculate_checksum, + convert_to_signed, + parse_03_response, + parse_04_response, + verify_checksum, +) +from bmspy.jbd_snapshot import JBDBMS from bmspy.utilities import debugger -from bmspy.classes import BMSScalarField, BMSMultiField, BMSInfoField, BMSField, UPS - - -@dataclass -class JBDBMS(UPS): - bms_voltage_total_volts: BMSScalarField | None = None - bms_current_amps: BMSScalarField | None = None - bms_capacity_remaining_ah: BMSScalarField | None = None - bms_capacity_nominal_ah: BMSScalarField | None = None - bms_charge_cycles: BMSScalarField | None = None - bms_manufacture_date: BMSInfoField | None = None - bms_protection_sop_bool: BMSScalarField | None = None - bms_protection_sup_bool: BMSScalarField | None = None - bms_protection_wgop_bool: BMSScalarField | None = None - bms_protection_wgup_bool: BMSScalarField | None = None - bms_protection_cotp_bool: BMSScalarField | None = None - bms_protection_cutp_bool: BMSScalarField | None = None - bms_protection_dotp_bool: BMSScalarField | None = None - bms_protection_dutp_bool: BMSScalarField | None = None - bms_protection_cocp_bool: BMSScalarField | None = None - bms_protection_docp_bool: BMSScalarField | None = None - bms_protection_scp_bool: BMSScalarField | None = None - bms_protection_fdic_bool: BMSScalarField | None = None - bms_protection_slmos_bool: BMSScalarField | None = None - bms_capacity_charge_ratio: BMSScalarField | None = None - bms_charge_is_charging: BMSScalarField | None = None - bms_charge_is_discharging: BMSScalarField | None = None - bms_cell_number: BMSScalarField | None = None - bms_cells_balancing: BMSMultiField | None = None - bms_temperature_sensor_num: BMSScalarField | None = None - bms_temperature_celcius: BMSMultiField | None = None - bms_voltage_cells_volts: BMSMultiField | None = None - - def items(self) -> Iterator[tuple[str, BMSField]]: - """Yield (name, field) for all non-None BMS field attributes.""" - for f in dataclass_fields(self): - v = getattr(self, f.name) - if v is not None: - yield f.name, v def serial_cleanup(ser: serial.Serial, debug: int = 0) -> None: @@ -74,51 +46,6 @@ def initialise_serial(device: str, debug: int = 0) -> serial.Serial: return ser -def calculate_checksum(msg: bytes | bytearray) -> str: - checksum = "" - return checksum - - -def verify_checksum(data: bytes | bytearray, checksum: bytes) -> bool: - """Verify a JBD BMS packet checksum (sum complement + 1, big-endian).""" - # (data + length + command code) checksum, then complement, then add 1, high bit first, low bit last - # data should have start/rw stripped - s = 0 - for i in data: - s += i - s = (s ^ 0xFFFF) + 1 - chk = bytes_to_digits(checksum[0], checksum[1]) - return s == chk - - -def convert_to_signed(x: int) -> int: - """Convert a JBD current value (possibly encoded) to a signed integer.""" - # For values below 1024, these seem to be actual results - # For values above 1024, these seem to be encoded to account for high and negative floats - max_uint = 1024 - if x >= max_uint: - return (x - 2**9) % 2**10 - 2**9 - else: - return x - - -def bytes_to_digits(high: int, low: int) -> int: - """Combine two bytes (high, low) into a single 16-bit unsigned integer.""" - result = high - result <<= 8 - result = result | low - return result - - -def bytes_to_date(high: int, low: int) -> str: - """Decode a JBD-encoded manufacture date from two bytes into 'YYYY-MM-DD' format.""" - result = bytes_to_digits(high, low) - day = result & 0x1F - mon = (result >> 5) & 0x0F - year = 2000 + (result >> 9) - return "{:04d}-{:02d}-{:02d}".format(year, mon, day) - - def requestMessage(ser: serial.Serial, reqmsg: bytearray, debug: int = 0) -> bytes | str | bool: """Send a request message to the BMS over serial and return the raw response bytes.""" if debug > 2: @@ -179,332 +106,6 @@ def requestMessage(ser: serial.Serial, reqmsg: bytearray, debug: int = 0) -> byt debugger("serial: cannot open port") -def parse_03_response(response: bytearray, debug: int = 0) -> JBDBMS | bool: - """Parse a JBD BMS 0x03 (basic info) response packet into a JBDBMS dataclass.""" - # Response is 34 bytes: - # 00 begin: \xDD - # 01 r/w: \xA5 - # 02 status: \x00 = correct; \x80 = incorrect - # 03 length (usually 27) - # 04 data (size of length) - # ... - # length+4 checksum - # length+5 checksum - # length+6 end \x77 - if bytes([response[0]]) != b"\xdd": - debugger( - "parse_03_response ERROR: first byte not found: {0}".format(response[0]) - ) - return False - - if bytes([response[2]]) == b"\x80": - debugger( - "parse_03_response ERROR: error byte returned from BMS: {0}".format( - response[2] - ) - ) - return False - - data_len = response[3] - if debug > 2: - debugger( - "parse_03_response: data length (trimming 4 bytes): {0}".format(data_len) - ) - - # The checksum is two bytes, offset by data_len + 4 - # Five bytes at the front of data: begin; rw; status, command; length - # The checksum should check command, length, and data: [3] to [3+data_len+1] - first = data_len + 4 - second = data_len + 5 - if second > len(response): - debugger("parse_03_response ERROR: primary response checksum not found") - return False - checksum = bytes([response[first], response[second]]) - if not verify_checksum(response[3:first], checksum): - debugger("parse_03_response ERROR: failed to validate received checksum") - return False - - if data_len == 0: # pragma: no cover - return False - - result = JBDBMS() - - vtot = bytes_to_digits(response[4], response[5]) * 0.01 - result.bms_voltage_total_volts = BMSScalarField( - help="Total Voltage", raw_value=vtot, value="{:.2f}".format(vtot), units="V" - ) - if debug > 1: - debugger(" Total voltage: {:.2f}V".format(vtot)) - - current = convert_to_signed(bytes_to_digits(response[6], response[7])) * 0.01 - result.bms_current_amps = BMSScalarField( - help="Current", raw_value=current, value="{:.2f}".format(current), units="A" - ) - if debug > 1: - debugger(" Current: {:.2f}A".format(current)) - - res_cap = bytes_to_digits(response[8], response[9]) * 0.01 - nom_cap = bytes_to_digits(response[10], response[11]) * 0.01 - result.bms_capacity_remaining_ah = BMSScalarField( - help="Remaining Capacity", - raw_value=res_cap, - value="{:.2f}".format(res_cap), - units="Ah", - ) - result.bms_capacity_nominal_ah = BMSScalarField( - help="Nominal Capacity", - raw_value=nom_cap, - value="{:.2f}".format(nom_cap), - units="Ah", - ) - if debug > 1: - debugger(" Remaining capacity: {:.2f}Ah".format(res_cap)) - debugger(" Nominal capacity: {:.2f}Ah".format(nom_cap)) - - cycle_times = bytes_to_digits(response[12], response[13]) - result.bms_charge_cycles = BMSScalarField( - help="Charge Cycles", raw_value=cycle_times, value="{0}".format(cycle_times) - ) - if debug > 1: - debugger(" Cycle times: {0}".format(cycle_times)) - - man_date = bytes_to_date(response[14], response[15]) - result.bms_manufacture_date = BMSInfoField( - help="Date of Manufacture", info="{0}".format(man_date) - ) - if debug > 1: - debugger(" Manufacturing date: {0}".format(man_date)) - - cells = response[25] - result.bms_cell_number = BMSScalarField( - help="Cells", raw_value=cells, value="{0}".format(cells) - ) - if debug > 1: - debugger(" Cells: {0}S".format(cells)) - - balance_state_high = bytes_to_digits(response[16], response[17]) # 1S to 16S - balance_state_low = bytes_to_digits(response[18], response[19]) # 17S to 32S - # 1 bit per 4S (2 bytes = 16S); in 4S, we should expect: - # 0x0 (no cells balancing) 0 - # 0x1 (cell 1 balancing) 1 - # 0x2 (cell 2 balancing) 2 - # 0x3 (cells 1 + 2 balancing) 3 - # 0x4 (cell 3 balancing) 4 - # 0x5 (cells 1 + 3 balancing) 5 - # 0x6 (cells 2 + 3 balancing) 6 - # 0x7 (cells 1 + 2 + 3 balancing) 7 - # 0x8 (cell 4 balancing) 8 - # 0x9 (cells 1 + 4 balancing) 9 - # 0xA (cells 2 + 4 balancing) 10 - # 0xB (cells 1 + 2 + 4 balancing) 11 - # 0xC (cells 3 + 4 balancing) 12 - # 0xD (cells 1 + 3 + 4 balancing) 13 - # 0xE (cells 2 + 3 + 4 balancing) 14 - # 0xF (cells 1 + 2 + 3 + 4 balancing) 15 - raw_balancing = {} - str_balancing = {} - for cell in range(cells): - # Cells 1-16 are recorded in balance_state_low, 17-32 in balance_state_high; - # hilo_cell records the offset relative to the state group - if cell >= 16: - state = balance_state_high - hilo_cell = cell - 16 - else: - state = balance_state_low - hilo_cell = cell - # Cells are recorded as groups of 4 bits (0x0-0xF) per 4 cells - g = int(hilo_cell / 4) - b = 2 ** (hilo_cell - (g * 4)) - balancing = bool((state >> g) & b) - raw_balancing[cell + 1] = balancing - str_balancing[cell + 1] = "{0}".format(int(balancing)) - if debug > 1: - debugger(" Balancing cell {0}: {1}".format(cell, balancing)) - result.bms_cells_balancing = BMSMultiField( - help="Cells balancing", - label="cell", - raw_values=raw_balancing, - values=str_balancing, - ) - - protection_state = bytes_to_digits(response[20], response[21]) - sop = protection_state & 1 - sup = protection_state & 2 - gop = protection_state & 4 - gup = protection_state & 8 - cotp = protection_state & 16 - cutp = protection_state & 32 - dotp = protection_state & 64 - dutp = protection_state & 128 - cocp = protection_state & 256 - docp = protection_state & 512 - scp = protection_state & 1024 - fdic = protection_state & 2048 - slm = protection_state & 4096 - - def _prot(help, flag): - return BMSScalarField( - help=help, raw_value=bool(flag), value="{0}".format(int(bool(flag))) - ) - - result.bms_protection_sop_bool = _prot("Single overvoltage protection", sop) - result.bms_protection_sup_bool = _prot("Single undervoltage protection", sup) - result.bms_protection_wgop_bool = _prot("Whole group overvoltage protection", gop) - result.bms_protection_wgup_bool = _prot("Whole group undervoltage protection", gup) - result.bms_protection_cotp_bool = _prot( - "Charging over-temperature protection", cotp - ) - result.bms_protection_cutp_bool = _prot( - "Charging under-temperature protection", cutp - ) - result.bms_protection_dotp_bool = _prot( - "Discharging over-temperature protection", dotp - ) - result.bms_protection_dutp_bool = _prot("Discharging under-protection", dutp) - result.bms_protection_cocp_bool = _prot("Charging over-current protection", cocp) - result.bms_protection_docp_bool = _prot("Discharging over-current protection", docp) - result.bms_protection_scp_bool = _prot("Short-circuit protection", scp) - result.bms_protection_fdic_bool = _prot("Front detection IC error", fdic) - result.bms_protection_slmos_bool = _prot("Software lock MOS", slm) - - if debug > 2: - debugger(" Protection state: {0}".format(protection_state)) - for attr in ( - "sop", - "sup", - "gop", - "gup", - "cotp", - "cutp", - "dotp", - "dutp", - "cocp", - "docp", - "scp", - "fdic", - "slm", - ): - val = locals()[attr] - debugger(" {}: {}".format(attr, bool(val))) - - rsoc = response[23] * 0.01 - result.bms_capacity_charge_ratio = BMSScalarField( - help="Percent Charge", raw_value=rsoc, value="{0}".format(rsoc), units="‰" - ) - if debug > 1: - debugger(" Capacity remaining: {0}%".format(rsoc * 100)) - - # bit0 = charging; bit1 = discharging; 0 = MOS closing; 1 = MOS opening - control_status = response[24] - result.bms_charge_is_charging = BMSScalarField( - help="MOSFET charging", - raw_value=bool(control_status & 1), - value="{0}".format(int(bool(control_status & 1))), - ) - result.bms_charge_is_discharging = BMSScalarField( - help="MOSFET discharging", - raw_value=bool((control_status >> 1) & 1), - value="{0}".format(int(bool((control_status >> 1) & 1))), - ) - if debug > 1: - debugger( - " MOSFET charging: {0}".format("yes" if (control_status & 1) else "no") - ) - debugger( - " MOSFET discharging: {0}".format( - "yes" if ((control_status >> 1) & 1) else "no" - ) - ) - - ntc_num = response[26] - temperatures = [ - (bytes_to_digits(response[27 + (2 * i)], response[28 + (2 * i)]) - 2731) * 0.1 - for i in range(ntc_num) - ] - result.bms_temperature_sensor_num = BMSScalarField( - help="Temperature Sensors", raw_value=ntc_num, value="{0}".format(ntc_num) - ) - result.bms_temperature_celcius = BMSMultiField( - help="Temperature", - label="sensor", - raw_values={i + 1: temp for i, temp in enumerate(temperatures)}, - values={i + 1: "{:.2f}".format(temp) for i, temp in enumerate(temperatures)}, - units="°C", - ) - if debug > 1: - debugger(" Number of temperature sensors: {0}".format(ntc_num)) - for i, temp in enumerate(temperatures): - debugger(" Temperature sensor {:d}: {:.2f}°C".format(i + 1, temp)) - - return result - - -def parse_04_response(response: bytearray, debug: int = 0) -> BMSMultiField | bool: - """Parse a JBD BMS 0x04 (cell voltages) response packet into a BMSMultiField.""" - # Response is 7 + cells * 2 bytes: - # 00 begin: \xDD - # 01 r/w: \xA5 - # 02 status: \x00 = correct; \x80 = incorrect - # 03 length (usually 8) - # 04 data (size of length) - # ... - # length+4 checksum - # length+5 checksum - # length+6 end \x77 - if bytes([response[0]]) != b"\xdd": - debugger( - "parse_04_response ERROR: first byte not found: {0}".format(response[0]) - ) - return False - - if bytes([response[2]]) == b"\x80": - debugger( - "parse_04_response ERROR: error byte returned from BMS: {0}".format( - response[2] - ) - ) - return False - - data_len = response[3] - if debug > 2: - debugger(" Data length (trimming 4 bytes): {0}".format(data_len)) - - # The checksum is two bytes, offset by data_len + 4 - # Five bytes at the front of data: begin; rw; status, command; length - # The checksum should check command, length, and data: [3] to [3+data_len+1] - first = data_len + 4 - second = data_len + 5 - if second > len(response): - debugger("parse_04_response ERROR: cell voltage checksum not found") - return False - checksum = bytes([response[first], response[second]]) - if not verify_checksum(response[3:first], checksum): - debugger("parse_04_response ERROR: failed to validate received checksum") - return False - - if data_len == 0: # pragma: no cover - return False - - raw_values = {} - str_values = {} - for cell in range(int(data_len / 2)): - lo = (cell * 2) + 4 - hi = (cell * 2) + 5 - cellv = bytes_to_digits(response[lo], response[hi]) * 0.001 - raw_values[cell + 1] = cellv - str_values[cell + 1] = "{:.3f}".format(cellv) - if debug > 1: - debugger(" Cell {:.0f}: {:.3f}V".format(cell + 1, cellv)) - - return BMSMultiField( - help="Cell Voltages", - label="cell", - raw_values=raw_values, - values=str_values, - units="V", - ) - - def collect_data(ser: serial.Serial, debug: int = 0) -> JBDBMS | bool: """Request both 0x03 and 0x04 packets from the BMS and return a populated JBDBMS.""" # Request is 7 bytes: diff --git a/bmspy/jbd_proto.py b/bmspy/jbd_proto.py new file mode 100644 index 0000000..c17534f --- /dev/null +++ b/bmspy/jbd_proto.py @@ -0,0 +1,376 @@ +"""JBD checksums, byte conversions, and response decoding. No serial I/O.""" + +from bmspy.classes import BMSInfoField, BMSMultiField, BMSScalarField +from bmspy.jbd_snapshot import JBDBMS +from bmspy.utilities import debugger + + +def calculate_checksum(msg: bytes | bytearray) -> str: + checksum = "" + return checksum + + +def verify_checksum(data: bytes | bytearray, checksum: bytes) -> bool: + """Verify a JBD BMS packet checksum (sum complement + 1, big-endian).""" + # (data + length + command code) checksum, then complement, then add 1, high bit first, low bit last + # data should have start/rw stripped + s = 0 + for i in data: + s += i + s = (s ^ 0xFFFF) + 1 + chk = bytes_to_digits(checksum[0], checksum[1]) + return s == chk + + +def convert_to_signed(x: int) -> int: + """Convert a JBD current value (possibly encoded) to a signed integer.""" + # For values below 1024, these seem to be actual results + # For values above 1024, these seem to be encoded to account for high and negative floats + max_uint = 1024 + if x >= max_uint: + return (x - 2**9) % 2**10 - 2**9 + else: + return x + + +def bytes_to_digits(high: int, low: int) -> int: + """Combine two bytes (high, low) into a single 16-bit unsigned integer.""" + result = high + result <<= 8 + result = result | low + return result + + +def bytes_to_date(high: int, low: int) -> str: + """Decode a JBD-encoded manufacture date from two bytes into 'YYYY-MM-DD' format.""" + result = bytes_to_digits(high, low) + day = result & 0x1F + mon = (result >> 5) & 0x0F + year = 2000 + (result >> 9) + return "{:04d}-{:02d}-{:02d}".format(year, mon, day) + + +def parse_03_response(response: bytearray, debug: int = 0) -> JBDBMS | bool: + """Parse a JBD BMS 0x03 (basic info) response packet into a JBDBMS dataclass.""" + # Response is 34 bytes: + # 00 begin: \xDD + # 01 r/w: \xA5 + # 02 status: \x00 = correct; \x80 = incorrect + # 03 length (usually 27) + # 04 data (size of length) + # ... + # length+4 checksum + # length+5 checksum + # length+6 end \x77 + if bytes([response[0]]) != b"\xdd": + debugger( + "parse_03_response ERROR: first byte not found: {0}".format(response[0]) + ) + return False + + if bytes([response[2]]) == b"\x80": + debugger( + "parse_03_response ERROR: error byte returned from BMS: {0}".format( + response[2] + ) + ) + return False + + data_len = response[3] + if debug > 2: + debugger( + "parse_03_response: data length (trimming 4 bytes): {0}".format(data_len) + ) + + # The checksum is two bytes, offset by data_len + 4 + # Five bytes at the front of data: begin; rw; status, command; length + # The checksum should check command, length, and data: [3] to [3+data_len+1] + first = data_len + 4 + second = data_len + 5 + if second > len(response): + debugger("parse_03_response ERROR: primary response checksum not found") + return False + checksum = bytes([response[first], response[second]]) + if not verify_checksum(response[3:first], checksum): + debugger("parse_03_response ERROR: failed to validate received checksum") + return False + + if data_len == 0: # pragma: no cover + return False + + result = JBDBMS() + + vtot = bytes_to_digits(response[4], response[5]) * 0.01 + result.bms_voltage_total_volts = BMSScalarField( + help="Total Voltage", raw_value=vtot, value="{:.2f}".format(vtot), units="V" + ) + if debug > 1: + debugger(" Total voltage: {:.2f}V".format(vtot)) + + current = convert_to_signed(bytes_to_digits(response[6], response[7])) * 0.01 + result.bms_current_amps = BMSScalarField( + help="Current", raw_value=current, value="{:.2f}".format(current), units="A" + ) + if debug > 1: + debugger(" Current: {:.2f}A".format(current)) + + res_cap = bytes_to_digits(response[8], response[9]) * 0.01 + nom_cap = bytes_to_digits(response[10], response[11]) * 0.01 + result.bms_capacity_remaining_ah = BMSScalarField( + help="Remaining Capacity", + raw_value=res_cap, + value="{:.2f}".format(res_cap), + units="Ah", + ) + result.bms_capacity_nominal_ah = BMSScalarField( + help="Nominal Capacity", + raw_value=nom_cap, + value="{:.2f}".format(nom_cap), + units="Ah", + ) + if debug > 1: + debugger(" Remaining capacity: {:.2f}Ah".format(res_cap)) + debugger(" Nominal capacity: {:.2f}Ah".format(nom_cap)) + + cycle_times = bytes_to_digits(response[12], response[13]) + result.bms_charge_cycles = BMSScalarField( + help="Charge Cycles", raw_value=cycle_times, value="{0}".format(cycle_times) + ) + if debug > 1: + debugger(" Cycle times: {0}".format(cycle_times)) + + man_date = bytes_to_date(response[14], response[15]) + result.bms_manufacture_date = BMSInfoField( + help="Date of Manufacture", info="{0}".format(man_date) + ) + if debug > 1: + debugger(" Manufacturing date: {0}".format(man_date)) + + cells = response[25] + result.bms_cell_number = BMSScalarField( + help="Cells", raw_value=cells, value="{0}".format(cells) + ) + if debug > 1: + debugger(" Cells: {0}S".format(cells)) + + balance_state_high = bytes_to_digits(response[16], response[17]) # 1S to 16S + balance_state_low = bytes_to_digits(response[18], response[19]) # 17S to 32S + # 1 bit per 4S (2 bytes = 16S); in 4S, we should expect: + # 0x0 (no cells balancing) 0 + # 0x1 (cell 1 balancing) 1 + # 0x2 (cell 2 balancing) 2 + # 0x3 (cells 1 + 2 balancing) 3 + # 0x4 (cell 3 balancing) 4 + # 0x5 (cells 1 + 3 balancing) 5 + # 0x6 (cells 2 + 3 balancing) 6 + # 0x7 (cells 1 + 2 + 3 balancing) 7 + # 0x8 (cell 4 balancing) 8 + # 0x9 (cells 1 + 4 balancing) 9 + # 0xA (cells 2 + 4 balancing) 10 + # 0xB (cells 1 + 2 + 4 balancing) 11 + # 0xC (cells 3 + 4 balancing) 12 + # 0xD (cells 1 + 3 + 4 balancing) 13 + # 0xE (cells 2 + 3 + 4 balancing) 14 + # 0xF (cells 1 + 2 + 3 + 4 balancing) 15 + raw_balancing = {} + str_balancing = {} + for cell in range(cells): + # Cells 1-16 are recorded in balance_state_low, 17-32 in balance_state_high; + # hilo_cell records the offset relative to the state group + if cell >= 16: + state = balance_state_high + hilo_cell = cell - 16 + else: + state = balance_state_low + hilo_cell = cell + # Cells are recorded as groups of 4 bits (0x0-0xF) per 4 cells + g = int(hilo_cell / 4) + b = 2 ** (hilo_cell - (g * 4)) + balancing = bool((state >> g) & b) + raw_balancing[cell + 1] = balancing + str_balancing[cell + 1] = "{0}".format(int(balancing)) + if debug > 1: + debugger(" Balancing cell {0}: {1}".format(cell, balancing)) + result.bms_cells_balancing = BMSMultiField( + help="Cells balancing", + label="cell", + raw_values=raw_balancing, + values=str_balancing, + ) + + protection_state = bytes_to_digits(response[20], response[21]) + sop = protection_state & 1 + sup = protection_state & 2 + gop = protection_state & 4 + gup = protection_state & 8 + cotp = protection_state & 16 + cutp = protection_state & 32 + dotp = protection_state & 64 + dutp = protection_state & 128 + cocp = protection_state & 256 + docp = protection_state & 512 + scp = protection_state & 1024 + fdic = protection_state & 2048 + slm = protection_state & 4096 + + def _prot(help, flag): + return BMSScalarField( + help=help, raw_value=bool(flag), value="{0}".format(int(bool(flag))) + ) + + result.bms_protection_sop_bool = _prot("Single overvoltage protection", sop) + result.bms_protection_sup_bool = _prot("Single undervoltage protection", sup) + result.bms_protection_wgop_bool = _prot("Whole group overvoltage protection", gop) + result.bms_protection_wgup_bool = _prot("Whole group undervoltage protection", gup) + result.bms_protection_cotp_bool = _prot( + "Charging over-temperature protection", cotp + ) + result.bms_protection_cutp_bool = _prot( + "Charging under-temperature protection", cutp + ) + result.bms_protection_dotp_bool = _prot( + "Discharging over-temperature protection", dotp + ) + result.bms_protection_dutp_bool = _prot("Discharging under-protection", dutp) + result.bms_protection_cocp_bool = _prot("Charging over-current protection", cocp) + result.bms_protection_docp_bool = _prot("Discharging over-current protection", docp) + result.bms_protection_scp_bool = _prot("Short-circuit protection", scp) + result.bms_protection_fdic_bool = _prot("Front detection IC error", fdic) + result.bms_protection_slmos_bool = _prot("Software lock MOS", slm) + + if debug > 2: + debugger(" Protection state: {0}".format(protection_state)) + for attr in ( + "sop", + "sup", + "gop", + "gup", + "cotp", + "cutp", + "dotp", + "dutp", + "cocp", + "docp", + "scp", + "fdic", + "slm", + ): + val = locals()[attr] + debugger(" {}: {}".format(attr, bool(val))) + + rsoc = response[23] * 0.01 + result.bms_capacity_charge_ratio = BMSScalarField( + help="Percent Charge", raw_value=rsoc, value="{0}".format(rsoc), units="‰" + ) + if debug > 1: + debugger(" Capacity remaining: {0}%".format(rsoc * 100)) + + # bit0 = charging; bit1 = discharging; 0 = MOS closing; 1 = MOS opening + control_status = response[24] + result.bms_charge_is_charging = BMSScalarField( + help="MOSFET charging", + raw_value=bool(control_status & 1), + value="{0}".format(int(bool(control_status & 1))), + ) + result.bms_charge_is_discharging = BMSScalarField( + help="MOSFET discharging", + raw_value=bool((control_status >> 1) & 1), + value="{0}".format(int(bool((control_status >> 1) & 1))), + ) + if debug > 1: + debugger( + " MOSFET charging: {0}".format("yes" if (control_status & 1) else "no") + ) + debugger( + " MOSFET discharging: {0}".format( + "yes" if ((control_status >> 1) & 1) else "no" + ) + ) + + ntc_num = response[26] + temperatures = [ + (bytes_to_digits(response[27 + (2 * i)], response[28 + (2 * i)]) - 2731) * 0.1 + for i in range(ntc_num) + ] + result.bms_temperature_sensor_num = BMSScalarField( + help="Temperature Sensors", raw_value=ntc_num, value="{0}".format(ntc_num) + ) + result.bms_temperature_celcius = BMSMultiField( + help="Temperature", + label="sensor", + raw_values={i + 1: temp for i, temp in enumerate(temperatures)}, + values={i + 1: "{:.2f}".format(temp) for i, temp in enumerate(temperatures)}, + units="°C", + ) + if debug > 1: + debugger(" Number of temperature sensors: {0}".format(ntc_num)) + for i, temp in enumerate(temperatures): + debugger(" Temperature sensor {:d}: {:.2f}°C".format(i + 1, temp)) + + return result + + +def parse_04_response(response: bytearray, debug: int = 0) -> BMSMultiField | bool: + """Parse a JBD BMS 0x04 (cell voltages) response packet into a BMSMultiField.""" + # Response is 7 + cells * 2 bytes: + # 00 begin: \xDD + # 01 r/w: \xA5 + # 02 status: \x00 = correct; \x80 = incorrect + # 03 length (usually 8) + # 04 data (size of length) + # ... + # length+4 checksum + # length+5 checksum + # length+6 end \x77 + if bytes([response[0]]) != b"\xdd": + debugger( + "parse_04_response ERROR: first byte not found: {0}".format(response[0]) + ) + return False + + if bytes([response[2]]) == b"\x80": + debugger( + "parse_04_response ERROR: error byte returned from BMS: {0}".format( + response[2] + ) + ) + return False + + data_len = response[3] + if debug > 2: + debugger(" Data length (trimming 4 bytes): {0}".format(data_len)) + + # The checksum is two bytes, offset by data_len + 4 + # Five bytes at the front of data: begin; rw; status, command; length + # The checksum should check command, length, and data: [3] to [3+data_len+1] + first = data_len + 4 + second = data_len + 5 + if second > len(response): + debugger("parse_04_response ERROR: cell voltage checksum not found") + return False + checksum = bytes([response[first], response[second]]) + if not verify_checksum(response[3:first], checksum): + debugger("parse_04_response ERROR: failed to validate received checksum") + return False + + if data_len == 0: # pragma: no cover + return False + + raw_values = {} + str_values = {} + for cell in range(int(data_len / 2)): + lo = (cell * 2) + 4 + hi = (cell * 2) + 5 + cellv = bytes_to_digits(response[lo], response[hi]) * 0.001 + raw_values[cell + 1] = cellv + str_values[cell + 1] = "{:.3f}".format(cellv) + if debug > 1: + debugger(" Cell {:.0f}: {:.3f}V".format(cell + 1, cellv)) + + return BMSMultiField( + help="Cell Voltages", + label="cell", + raw_values=raw_values, + values=str_values, + units="V", + ) diff --git a/bmspy/jbd_snapshot.py b/bmspy/jbd_snapshot.py new file mode 100644 index 0000000..3c090cf --- /dev/null +++ b/bmspy/jbd_snapshot.py @@ -0,0 +1,44 @@ +"""JBD measurement snapshot, independent of serial I/O.""" + +from collections.abc import Iterator +from dataclasses import dataclass, fields as dataclass_fields + +from bmspy.classes import BMSField, BMSInfoField, BMSMultiField, BMSScalarField, UPS + + +@dataclass +class JBDBMS(UPS): + bms_voltage_total_volts: BMSScalarField | None = None + bms_current_amps: BMSScalarField | None = None + bms_capacity_remaining_ah: BMSScalarField | None = None + bms_capacity_nominal_ah: BMSScalarField | None = None + bms_charge_cycles: BMSScalarField | None = None + bms_manufacture_date: BMSInfoField | None = None + bms_protection_sop_bool: BMSScalarField | None = None + bms_protection_sup_bool: BMSScalarField | None = None + bms_protection_wgop_bool: BMSScalarField | None = None + bms_protection_wgup_bool: BMSScalarField | None = None + bms_protection_cotp_bool: BMSScalarField | None = None + bms_protection_cutp_bool: BMSScalarField | None = None + bms_protection_dotp_bool: BMSScalarField | None = None + bms_protection_dutp_bool: BMSScalarField | None = None + bms_protection_cocp_bool: BMSScalarField | None = None + bms_protection_docp_bool: BMSScalarField | None = None + bms_protection_scp_bool: BMSScalarField | None = None + bms_protection_fdic_bool: BMSScalarField | None = None + bms_protection_slmos_bool: BMSScalarField | None = None + bms_capacity_charge_ratio: BMSScalarField | None = None + bms_charge_is_charging: BMSScalarField | None = None + bms_charge_is_discharging: BMSScalarField | None = None + bms_cell_number: BMSScalarField | None = None + bms_cells_balancing: BMSMultiField | None = None + bms_temperature_sensor_num: BMSScalarField | None = None + bms_temperature_celcius: BMSMultiField | None = None + bms_voltage_cells_volts: BMSMultiField | None = None + + def items(self) -> Iterator[tuple[str, BMSField]]: + """Yield (name, field) for all non-None BMS field attributes.""" + for f in dataclass_fields(self): + v = getattr(self, f.name) + if v is not None: + yield f.name, v diff --git a/tests/test_backend_modules.py b/tests/test_backend_modules.py index a5c83fd..86b6443 100644 --- a/tests/test_backend_modules.py +++ b/tests/test_backend_modules.py @@ -20,6 +20,7 @@ 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. @@ -37,8 +38,9 @@ class NoTransportImports(importlib.abc.MetaPathFinder): sys.meta_path.insert(0, NoTransportImports()) importlib.import_module('bmspy.' + sys.argv[1]) -from bmspy import ecoflow_proto, ecoflow_river2, ecoflow_river3 +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', @@ -46,7 +48,9 @@ assert ecoflow_proto.format_payload(2, 0xFE, 0x15, b'\\x08\\x00') == [ 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], @@ -55,10 +59,17 @@ assert list(EcoFlowSnapshot().items()) == [] assert result.returncode == 0, result.stderr -def test_snapshot_and_decoder_imports_remain_compatible(): - from bmspy import ecoflow_bt +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 diff --git a/tests/test_jbd_bms.py b/tests/test_jbd_bms.py index 2d428c3..65d5cb1 100644 --- a/tests/test_jbd_bms.py +++ b/tests/test_jbd_bms.py @@ -1,13 +1,15 @@ import pytest from unittest.mock import MagicMock, patch -from bmspy.jbd_bms import ( - JBDBMS, +from bmspy.jbd_snapshot import JBDBMS +from bmspy.jbd_proto import ( bytes_to_digits, bytes_to_date, convert_to_signed, verify_checksum, parse_03_response, parse_04_response, +) +from bmspy.jbd_bms import ( requestMessage, serial_cleanup, collect_data,