|
21 | 21 | sys.path.insert(0, str(repo)) |
22 | 22 |
|
23 | 23 | from collector.rs485 import Rs485DeviceConfig |
| 24 | +from collector.rs485.eg4_a5_bms import Eg4A5BmsDevice |
24 | 25 | from collector.rs485.jk_bms import JkBmsDevice, STATUS_REQUEST, STATUS_REQUEST_FALLBACK, _append_checksum |
25 | 26 |
|
26 | 27 |
|
| 28 | +def probe_eg4_a5_bms(port: str, baud: int, timeout: float = 1.0) -> bool: |
| 29 | + """Listen for EG4 A5/5A broadcast frames and see if they parse.""" |
| 30 | + try: |
| 31 | + import serial |
| 32 | + except ImportError: |
| 33 | + return False |
| 34 | + |
| 35 | + try: |
| 36 | + dev = serial.Serial(port, baudrate=baud, timeout=timeout) |
| 37 | + except Exception as exc: |
| 38 | + print(f" Could not open {port} at {baud}: {exc}") |
| 39 | + return False |
| 40 | + |
| 41 | + found = False |
| 42 | + try: |
| 43 | + dev.reset_input_buffer() |
| 44 | + # Send a minimal poll probe to wake a quiet BMS. |
| 45 | + dev.write(bytes([0xA5, 0x5A, 0x05, 0x82, 0x21, 0x13, 0x00, 0x00])) |
| 46 | + time.sleep(0.2) |
| 47 | + data = dev.read(4096) |
| 48 | + if data: |
| 49 | + print(f" [{baud}] EG4 A5 probe got {len(data)} bytes: {data[:60].hex()}") |
| 50 | + drv = Eg4A5BmsDevice(Rs485DeviceConfig(port=port, baudrate=baud, timeout=timeout)) |
| 51 | + parsed = drv.read() |
| 52 | + if parsed: |
| 53 | + print(f" [{baud}] EG4 A5 BMS parsed successfully:") |
| 54 | + for key in ("voltage", "current", "power", "temperature_pcb", "cell_count", "cell_delta_voltage"): |
| 55 | + info = parsed.get(key) |
| 56 | + if info: |
| 57 | + print(f" {key}: {info['value']} {info.get('unit', '')}") |
| 58 | + found = True |
| 59 | + else: |
| 60 | + print(f" [{baud}] EG4 A5 BMS parse failed") |
| 61 | + finally: |
| 62 | + dev.close() |
| 63 | + |
| 64 | + return found |
| 65 | + |
| 66 | + |
27 | 67 | def probe_jk_bms(port: str, baud: int, timeout: float = 1.0) -> bool: |
28 | 68 | """Send JK BMS status requests and see if anything parses.""" |
29 | 69 | try: |
@@ -74,6 +114,8 @@ def main(): |
74 | 114 | print(f"Probing {args.port} for RS-485 devices...") |
75 | 115 | for baud in args.baud: |
76 | 116 | print(f"\nBaud {baud}:") |
| 117 | + if probe_eg4_a5_bms(args.port, baud, args.timeout): |
| 118 | + print(f" -> EG4 A5/5A BMS detected at {baud} baud") |
77 | 119 | if probe_jk_bms(args.port, baud, args.timeout): |
78 | 120 | print(f" -> JK BMS detected at {baud} baud") |
79 | 121 |
|
|
0 commit comments