Fix BullTron discharge display and disconnect

bulltron_gui.py

@@ -201,6 +201,7 @@ class Telemetry:
     alarm_words: list[int] = field(default_factory=list)
     time_label: str = "Time"
     time_value: str = "-- h -- m"
+    residue_minutes: int | None = None
 
 
 @dataclass
@@ -255,19 +256,45 @@ def decode_live(payload: bytes, state: DeviceState) -> None:
     if tel.voltage is not None and tel.current is not None:
         current_abs = abs(tel.current)
         tel.power_w = current_abs * tel.voltage
-        tel.charge_current = current_abs if tel.state == 1 else 0.0 if tel.state == 2 else None
+        is_charging = tel.current > 0.05 or (abs(tel.current) <= 0.05 and tel.state == 1)
-        tel.discharge_current = current_abs if tel.state == 2 else 0.0 if tel.state == 1 else None
+        is_discharging = tel.current < -0.05 or (abs(tel.current) <= 0.05 and tel.state == 2)
+        tel.charge_current = current_abs if is_charging else 0.0 if is_discharging else None
+        tel.discharge_current = current_abs if is_discharging else 0.0 if is_charging else None
         tel.charge_power_w = tel.voltage * tel.charge_current if tel.charge_current is not None else None
         tel.discharge_power_w = tel.voltage * tel.discharge_current if tel.discharge_current is not None else None
+    if tel.residue_minutes is not None and tel.residue_minutes != 0xFFFF:
+        if tel.current is not None and tel.current < -0.05:
+            tel.time_label = "Time till empty"
+        elif tel.current is not None and tel.current > 0.05:
+            tel.time_label = "Time till full"
+        elif tel.state == 2:
+            tel.time_label = "Time till empty"
+        elif tel.state == 1:
+            tel.time_label = "Time till full"
+        else:
+            tel.time_label = "Time"
+        tel.time_value = format_duration(tel.residue_minutes)
+        state.alarms = decode_alarm_words(tel.alarm_words)
+        return
     capacity = state.rated_capacity_ah or tel.remaining_ah
     minutes = None
     if capacity and tel.soc is not None and tel.current and abs(tel.current) > 0.05:
-        if tel.state == 1:
+        if tel.current > 0.05:
             tel.time_label = "Time till full"
-            minutes = capacity * ((100 - tel.soc) / 100) / abs(tel.current) * 60
+            remaining = tel.remaining_ah if tel.remaining_ah is not None else capacity * (tel.soc / 100)
+            minutes = max(capacity - remaining, 0) / abs(tel.current) * 60
+        elif tel.current < -0.05:
+            tel.time_label = "Time till empty"
+            remaining = tel.remaining_ah if tel.remaining_ah is not None else capacity * (tel.soc / 100)
+            minutes = remaining / abs(tel.current) * 60
+        elif tel.state == 1:
+            tel.time_label = "Time till full"
+            remaining = tel.remaining_ah if tel.remaining_ah is not None else capacity * (tel.soc / 100)
+            minutes = max(capacity - remaining, 0) / abs(tel.current) * 60
         elif tel.state == 2:
             tel.time_label = "Time till empty"
-            minutes = capacity * (tel.soc / 100) / abs(tel.current) * 60
+            remaining = tel.remaining_ah if tel.remaining_ah is not None else capacity * (tel.soc / 100)
+            minutes = remaining / abs(tel.current) * 60
         else:
             tel.time_label = "Time"
     tel.time_value = format_duration(minutes)
@@ -294,6 +321,8 @@ def decode_response(frame: bytes, state: DeviceState) -> None:
     state.raw_blocks[f"D203 byte_count 0x{byte_count:02X}"] = payload.hex(" ").upper()
     if byte_count == 0x7C:
         decode_live(payload, state)
+    elif byte_count == 0x02:
+        state.telemetry.residue_minutes = int.from_bytes(payload, "big")
     elif byte_count == 0x06:
         text = clean_ascii(payload)
         state.control_pin = text or payload.hex().upper()
@@ -498,6 +527,7 @@ class BullTronGui(tk.Tk):
         self.scan_timeout = tk.StringVar(value="12")
         ttk.Spinbox(scan_row, from_=3, to=60, textvariable=self.scan_timeout, width=4).pack(side=tk.RIGHT)
         ttk.Button(left, text="Connect selected", command=self._connect_selected).pack(fill=tk.X)
+        ttk.Button(left, text="Disconnect", command=self._disconnect).pack(fill=tk.X, pady=(4, 0))
         direct = ttk.LabelFrame(left, text="Direct connect")
         direct.pack(fill=tk.X, pady=(8, 0))
         self.direct_address = tk.StringVar(value=BULLTRON_CAPTURE_ADDRESS)
@@ -609,6 +639,10 @@ class BullTronGui(tk.Tk):
             return
         self.worker.connect(address)
 
+    def _disconnect(self) -> None:
+        self.worker.submit(self.worker.disconnect())
+        self.status.set("Disconnecting...")
+
     def _unlock(self) -> None:
         entered = self.pin_entry.get().strip()
         expected = self.vars.get("read_pin", tk.StringVar(value="000000")).get()

tests/test_bulltron_gui.py

@@ -42,7 +42,7 @@ class BullTronParserTest(unittest.TestCase):
         self.assertTrue(state.telemetry.charge_mos)
         self.assertTrue(state.telemetry.discharge_mos)
         self.assertEqual(state.telemetry.cell_count, 4)
-        self.assertEqual(state.telemetry.time_value, "9 h 21 m")
+        self.assertEqual(state.telemetry.time_value, "14 h 20 m")
 
     def test_charging_time_uses_soc_fraction_and_rated_capacity(self):
         state = DeviceState(rated_capacity_ah=160)
@@ -51,7 +51,7 @@ class BullTronParserTest(unittest.TestCase):
         words[41] = 30152
         words[42] = 450
         words[47] = 1
-        words[48] = 72
+        words[48] = 720
         payload = b"".join(word.to_bytes(2, "big") for word in words)
         body = bytes([0xD2, 0x03, len(payload)]) + payload
         from bulltron_gui import crc16_modbus_swapped
@@ -64,6 +64,50 @@ class BullTronParserTest(unittest.TestCase):
         self.assertEqual(state.telemetry.time_label, "Time till full")
         self.assertEqual(state.telemetry.time_value, "5 h 47 m")
 
+    def test_current_direction_uses_signed_current_for_display(self):
+        state = DeviceState(rated_capacity_ah=160)
+        words = [0] * 62
+        words[40] = 128
+        words[41] = 29922
+        words[42] = 490
+        words[47] = 1
+        words[48] = 784
+        payload = b"".join(word.to_bytes(2, "big") for word in words)
+        body = bytes([0xD2, 0x03, len(payload)]) + payload
+        from bulltron_gui import crc16_modbus_swapped
+
+        frame = body + crc16_modbus_swapped(body).to_bytes(2, "big")
+        decode_response(frame, state)
+        self.assertAlmostEqual(state.telemetry.current, -7.8)
+        self.assertAlmostEqual(state.telemetry.discharge_current, 7.8)
+        self.assertAlmostEqual(state.telemetry.discharge_power_w, 99.84)
+        self.assertAlmostEqual(state.telemetry.charge_current, 0.0)
+        self.assertEqual(state.telemetry.time_label, "Time till empty")
+        self.assertEqual(state.telemetry.time_value, "10 h 03 m")
+
+    def test_bms_residue_time_register_overrides_estimate(self):
+        state = DeviceState(rated_capacity_ah=160)
+        residue_payload = (6 * 60 + 30).to_bytes(2, "big")
+        residue_body = bytes([0xD2, 0x03, len(residue_payload)]) + residue_payload
+        from bulltron_gui import crc16_modbus_swapped
+
+        residue_frame = residue_body + crc16_modbus_swapped(residue_body).to_bytes(2, "big")
+        decode_response(residue_frame, state)
+
+        words = [0] * 62
+        words[40] = 128
+        words[41] = 29999
+        words[42] = 490
+        words[47] = 2
+        words[48] = 784
+        payload = b"".join(word.to_bytes(2, "big") for word in words)
+        body = bytes([0xD2, 0x03, len(payload)]) + payload
+        live_frame = body + crc16_modbus_swapped(body).to_bytes(2, "big")
+        decode_response(live_frame, state)
+
+        self.assertEqual(state.telemetry.time_label, "Time till empty")
+        self.assertEqual(state.telemetry.time_value, "6 h 30 m")
+
     def test_settings_capacity_uses_app_word_zero(self):
         state = DeviceState()
         words = [1600, 3750, 42] + [0] * 38