#!/usr/bin/env python3 """ TCL S-Line Bus Monitor — Real-time IDU↔ODU protocol visualization. Connects to an ESP32 WiFi bridge (raw TCP byte stream at 600 baud) or replays capture_1.jsonl files for offline analysis. Protocol: 19-byte packets, 0xAA sync, 0x55 end, checksum = 0xA9. See TCL_Minisplit_IDU_ODU_Protocol_RE.md for full protocol documentation. Author: TG (dj9kw.de) """ import json import sys import time from collections import deque from dataclasses import dataclass, field from pathlib import Path import numpy as np import pyqtgraph as pg from PyQt6.QtCore import ( QByteArray, QIODevice, QTimer, Qt, pyqtSignal, QObject, QThread, ) from PyQt6.QtGui import QColor, QFont, QPalette, QTextCharFormat from PyQt6.QtNetwork import QAbstractSocket, QTcpSocket from PyQt6.QtWidgets import ( QApplication, QCheckBox, QFileDialog, QGridLayout, QGroupBox, QHBoxLayout, QHeaderView, QLabel, QLineEdit, QMainWindow, QPlainTextEdit, QPushButton, QSizePolicy, QSplitter, QStyle, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget, QSpinBox, ) # ── Protocol constants ────────────────────────────────────────────────────── SYNC = 0xAA END = 0x55 PKT_LEN = 19 CHK_TARGET = 0xA9 # Packet type metadata PKT_TYPES = { 0x01: ("CMD", "Command (IDU→ODU)", QColor("#4FC3F7")), # light blue 0x02: ("SENS", "Sensor (ODU→IDU)", QColor("#81C784")), # green 0x14: ("POLL", "Heartbeat poll", QColor("#90A4AE")), # gray 0x81: ("RCMD", "Response to CMD", QColor("#FFB74D")), # orange 0x82: ("RSENS", "Response to SENS", QColor("#E57373")), # red 0x84: ("RPOLL", "Response to POLL", QColor("#B0BEC5")), # light gray } MODE_NAMES = {0: "STANDBY", 1: "COOL/VENT", 2: "COIL-DRY", 3: "DEHUMIDIFY", 4: "HEAT"} MODE_STATUS_NAMES = {0x70: "STANDBY", 0x71: "COOL/VENT", 0x72: "COIL-DRY", 0x73: "DEHUMIDIFY", 0x74: "HEAT"} VALVE_NAMES = {0: "COOL/OFF", 8: "HEAT"} # ── NTC Steinhart-Hart conversion (water bath calibrated 2026-04-12) ──────── # All bus sensor values are raw 8-bit ADC: RAW = 255 × Rpull / (R_ntc + Rpull) # Convert: R = Rpull × (255/RAW − 1), then T(K) = 1/(A + B·ln(R) + C·ln(R)³) import math _NTC_PARAMS = { # (Rpull, A, B, C) "B7_ODT": (2000, 1.1362532432e-03, 2.0967549577e-04, 1.4371748865e-07), # 20kΩ NTC, CN8 "B8_OPT": (5100, 1.2762540165e-03, 2.1830488166e-04, 3.5380619936e-07), # 5kΩ NTC, CN9 "B9_OAT": (5100, 1.3151700813e-03, 2.1027974713e-04, 4.0091436008e-07), # 5kΩ NTC, CN9 # IDU sensors use a non-standard circuit (implied Rpull varies with RAW). # Converted via direct quadratic fit instead — see idu_raw_to_temp_c() below. } def raw_to_temp_c(raw: int, sensor: str) -> float: """Convert raw 8-bit ADC bus value to temperature in °C using Steinhart-Hart. Returns float('nan') if raw is out of range (open/short circuit).""" if raw <= 2 or raw >= 254: return float('nan') rpull, A, B, C = _NTC_PARAMS[sensor] r_ntc = rpull * (255.0 / raw - 1.0) ln_r = math.log(r_ntc) t_kelvin = 1.0 / (A + B * ln_r + C * ln_r * ln_r * ln_r) return t_kelvin - 273.15 def idu_raw_to_temp_c(raw: int) -> float: """Convert IDU raw 8-bit ADC to temperature in °C. Direct quadratic fit: 1/T(K) = a + b·RAW + c·RAW² Calibrated from 3 points: 9°C/RAW=78, 17°C/RAW=97, 65°C/RAW=218. IDU circuit has non-standard topology (not a simple NTC+pulldown divider), so Steinhart-Hart with fixed Rpull does not apply.""" if raw <= 2 or raw >= 254: return float('nan') inv_t = 4.0048320172e-03 + (-6.5182245701e-06) * raw + 7.8572599935e-09 * raw * raw if inv_t <= 0: return float('nan') return 1.0 / inv_t - 273.15 # ── Data structures ───────────────────────────────────────────────────────── @dataclass class DecodedPacket: ts: float rel_time: float pkt_type: int raw: list # Decoded fields (filled per type) mode: str = "" compressor_freq: int = 0 compressor_rpm: int = 0 valve: str = "" on_off: str = "" setpoint: float = 0.0 discharge_temp: float = 0.0 # B7: ODT, raw 8-bit ADC → Steinhart-Hart → °C (20kΩ NTC + 2kΩ pull) coil_temp: float = 0.0 # B8: OPT, raw 8-bit ADC → Steinhart-Hart → °C (5kΩ NTC + 5.1kΩ pull) ambient_temp: float = 0.0 # B9: OAT, raw 8-bit ADC → Steinhart-Hart → °C (5kΩ NTC + 5.1kΩ pull) b11_val: int = 0 # B11: constant raw=60, not a sensor, purpose unknown idu_coil_temp: float = 0.0 # IDU B11 (0x81/0x82): indoor coil temperature, raw ADC idu_room_temp: float = 0.0 # IDU B12 (0x81/0x82): indoor ambient/room temperature, raw ADC mode_status: str = "" # ODU actuator fields (from type 0x01 — ODU status report) odu_b12: int = -1 # B12: EEV stepper position counter (0-254) odu_fan: int = -1 # B13: ODU outdoor fan speed (0-90, independent PWM) # Raw interesting bytes for display extra: dict = field(default_factory=dict) def decode_packet(raw: list, ts: float, t0: float) -> DecodedPacket: """Decode a validated 19-byte packet into structured fields.""" pkt = DecodedPacket( ts=ts, rel_time=ts - t0, pkt_type=raw[2], raw=raw, ) b = raw # shorthand if b[2] == 0x01: # Command (IDU→ODU) pkt.mode = MODE_NAMES.get(b[4], f"0x{b[4]:02X}") pkt.compressor_freq = b[5] pkt.compressor_rpm = b[5] * 60 pkt.valve = VALVE_NAMES.get(b[6], f"0x{b[6]:02X}") pkt.on_off = "ACTIVE" if b[5] > 0 else ("STANDBY" if b[4] == 0 else "IDLE") # B11: previously assumed on/off but B11=1 observed with compressor running. # B11: on/off flag — toggles during startup/shutdown transitions if b[14] > 0: pkt.ambient_temp = b[14] - 50 # ODU actuator fields pkt.odu_b12 = b[12] pkt.odu_fan = b[13] # ODU outdoor fan speed (independent PWM) pkt.extra = { "B7": b[7], "B8": b[8], "B9": b[9], "B10": b[10], "B16": b[16], } elif b[2] == 0x02: # Sensor (all values are raw 8-bit ADC) pkt.discharge_temp = raw_to_temp_c(b[7], "B7_ODT") # ODT: 20kΩ NTC + 2kΩ pull, CN8 pkt.coil_temp = raw_to_temp_c(b[8], "B8_OPT") # OPT: 5kΩ NTC + 5.1kΩ pull, CN9 pkt.ambient_temp = raw_to_temp_c(b[9], "B9_OAT") # OAT: 5kΩ NTC + 5.1kΩ pull, CN9 pkt.b11_val = b[11] # Constant raw=60, not a sensor pkt.extra = { "CNT": b[4], "VBUS": b[5], "B13": b[13], } elif b[2] in (0x81, 0x82): # IDU → ODU command pkt.mode_status = MODE_STATUS_NAMES.get(b[4], f"0x{b[4]:02X}") # B9 = CRT (Compensated Room Temperature) = setpoint + fan_offset # B6 = turbo flag (0x02 = turbo active) # B10 = IDU fan speed (80-125 range) # B11 = RT room temperature (raw ADC, 10kΩ NTC + ~10kΩ pull) # B12 = IPT indoor pipe temperature (same NTC type) pkt.setpoint = b[9] # CRT raw value, decoded by dashboard using auto-detected offset pkt.idu_coil_temp = idu_raw_to_temp_c(b[11]) # B11: indoor coil pkt.idu_room_temp = idu_raw_to_temp_c(b[12]) # B12: indoor ambient pkt.extra = { "B4": f"0x{b[4]:02X}", "B5": b[5], "B6_turbo": b[6], "B9_CRT": b[9], "B10_fan": b[10], "B11": b[11], "B12": b[12], "B14": b[14], "B16": f"0x{b[16]:02X}", } elif b[2] == 0x14: # Heartbeat poll pkt.extra = {"B7": f"0x{b[7]:02X}"} elif b[2] == 0x84: # Heartbeat response / fan mode fan_mode_names = { 0x09: "Fan 1", 0x0D: "Fan 2", 0x0A: "Fan 3/Auto(lo)", 0x0E: "Fan 4", 0x0B: "Fan 5/Auto(hi)/Turbo", 0x0C: "Silent", } fm = fan_mode_names.get(b[4], f"0x{b[4]:02X}") pkt.extra = {"B4_fan_mode": f"0x{b[4]:02X} ({fm})", "B7": f"0x{b[7]:02X}"} return pkt # ── Packet Parser ─────────────────────────────────────────────────────────── class PacketParser(QObject): """Assembles raw bytes into validated 19-byte packets.""" packet_ready = pyqtSignal(object) # DecodedPacket raw_byte = pyqtSignal(float, int) # (timestamp, byte_value) stats_updated = pyqtSignal(int, int, int) # valid, errors, resyncs def __init__(self): super().__init__() self._buf = [] self._t0 = None self._valid = 0 self._errors = 0 self._resyncs = 0 # Unit variant auto-detection from 0x84 heartbeat response # 0x0B = living room unit (setpoint = B9 - 32) # 0x09 = lab/donor unit (setpoint = B9 raw) self.unit_variant_84_b4 = None # set from first 0x84 packet self.setpoint_offset = 0 # auto-detected: 0 or 32 def reset(self): self._buf.clear() self._t0 = None self._valid = 0 self._errors = 0 self._resyncs = 0 self.unit_variant_84_b4 = None self.setpoint_offset = 0 def feed(self, ts: float, byte_val: int): """Feed one byte with timestamp.""" if self._t0 is None: self._t0 = ts self.raw_byte.emit(ts, byte_val) self._buf.append((ts, byte_val)) # Try to parse when we have enough bytes while len(self._buf) >= PKT_LEN: # Find sync byte if self._buf[0][1] != SYNC: self._buf.pop(0) self._resyncs += 1 continue # Check end marker if self._buf[17][1] != END: self._buf.pop(0) self._resyncs += 1 continue # Extract packet bytes pkt_bytes = [b[1] for b in self._buf[:PKT_LEN]] pkt_ts = self._buf[0][0] # Verify checksum if sum(pkt_bytes) & 0xFF == CHK_TARGET: # Auto-detect fan mode and CRT offset from 0x84 packets if pkt_bytes[2] == 0x84: self.unit_variant_84_b4 = pkt_bytes[4] # CRT offset = fan-mode-dependent compensation # Verified by live testing 2026-04-07 at fixed 24°C setpoint crt_offsets = { 0x09: 0, # Fan 1 0x0D: 16, # Fan 2 0x0A: 16, # Fan 3 / Auto (low demand) 0x0E: 32, # Fan 4 0x0B: 32, # Fan 5 / Auto (high demand) / Turbo 0x0C: 48, # Super silent } self.setpoint_offset = crt_offsets.get(pkt_bytes[4], self.setpoint_offset) pkt = decode_packet(pkt_bytes, pkt_ts, self._t0) self._valid += 1 self.packet_ready.emit(pkt) else: self._errors += 1 self.stats_updated.emit(self._valid, self._errors, self._resyncs) self._buf = self._buf[PKT_LEN:] @property def t0(self): return self._t0 # ── File Replay Worker ────────────────────────────────────────────────────── class ReplayWorker(QObject): """Replays a JSONL capture file, feeding bytes to the parser.""" byte_ready = pyqtSignal(float, int) finished = pyqtSignal() progress = pyqtSignal(int, int) # current, total def __init__(self, filepath: str, speed: float = 100.0): super().__init__() self.filepath = filepath self.speed = speed # multiplier (100 = 100x realtime) self._running = True def stop(self): self._running = False def run(self): records = [] with open(self.filepath) as f: for line in f: records.append(json.loads(line)) total = len(records) if total == 0: self.finished.emit() return t0 = records[0]["ts"] last_real_ts = time.monotonic() last_sim_ts = t0 for i, rec in enumerate(records): if not self._running: break ts = rec["ts"] byte_val = int(rec["hex"], 16) # Pace replay if self.speed < 10000: # not "instant" sim_delta = ts - last_sim_ts real_delta = sim_delta / self.speed if real_delta > 0.0001: elapsed = time.monotonic() - last_real_ts sleep_time = real_delta - elapsed if sleep_time > 0: QThread.msleep(max(1, int(sleep_time * 1000))) self.byte_ready.emit(ts, byte_val) last_real_ts = time.monotonic() last_sim_ts = ts if i % 1000 == 0: self.progress.emit(i, total) self.progress.emit(total, total) self.finished.emit() # ── Dashboard Widget ──────────────────────────────────────────────────────── class DashboardWidget(QWidget): """Shows current decoded state with large labels.""" def __init__(self): super().__init__() layout = QVBoxLayout(self) layout.setContentsMargins(4, 4, 4, 4) # -- Command state (from 0x01) -- cmd_group = QGroupBox("Command State (Type 0x01 — IDU→ODU)") cmd_grid = QGridLayout(cmd_group) self._lbl = {} fields_cmd = [ ("mode", "Mode"), ("mode_flag", "Mode Flag (B16)"), ("freq", "Compressor"), ("valve", "4-Way Valve"), ("onoff", "Status"), ] for row, (key, label) in enumerate(fields_cmd): lbl = QLabel(label + ":") lbl.setStyleSheet("color: #90CAF9; font-weight: bold;") val = QLabel("—") val.setFont(QFont("Menlo", 16, QFont.Weight.Bold)) val.setStyleSheet("color: white;") cmd_grid.addWidget(lbl, row, 0) cmd_grid.addWidget(val, row, 1) self._lbl[key] = val layout.addWidget(cmd_group) # -- ODU actuator fields (from 0x01 — ODU status report) -- odu_group = QGroupBox("ODU Actuators (Type 0x01 — ODU→IDU)") odu_grid = QGridLayout(odu_group) fields_odu = [ ("odu_fan", "ODU Fan Speed (B13)"), ("odu_b12", "EEV Position (B12)"), ] for row, (key, label) in enumerate(fields_odu): lbl = QLabel(label + ":") lbl.setStyleSheet("color: #CE93D8; font-weight: bold;") val = QLabel("—") val.setFont(QFont("Menlo", 16, QFont.Weight.Bold)) val.setStyleSheet("color: white;") odu_grid.addWidget(lbl, row, 0) odu_grid.addWidget(val, row, 1) self._lbl[key] = val # Add info label info = QLabel( "B13: ODU fan speed (DC brushless 310V, 0-90, independent PWM).\n" "B12: EEV stepper position (0-254, fast to hard stop then slow to target).\n" "Direction: ODU → IDU (ODU reports current actuator state)." ) info.setStyleSheet("color: #78909C; font-size: 10px;") info.setWordWrap(True) odu_grid.addWidget(info, len(fields_odu), 0, 1, 2) layout.addWidget(odu_group) # -- Sensor data (from 0x02) -- sens_group = QGroupBox("ODU Sensors (Type 0x02 — ODU→IDU)") sens_grid = QGridLayout(sens_group) fields_sens = [ ("discharge", "Discharge (ODT, B7)"), ("coil", "ODU Coil (OPT, B8)"), ("ambient", "Ambient (OAT, B9)"), ("b11", "B11 (const=60)"), ("vbus", "Vbus (B5)"), ] for row, (key, label) in enumerate(fields_sens): lbl = QLabel(label + ":") lbl.setStyleSheet("color: #A5D6A7; font-weight: bold;") val = QLabel("—") val.setFont(QFont("Menlo", 16, QFont.Weight.Bold)) val.setStyleSheet("color: white;") sens_grid.addWidget(lbl, row, 0) sens_grid.addWidget(val, row, 1) self._lbl[key] = val layout.addWidget(sens_group) # -- IDU command data (from 0x81/0x82) -- resp_group = QGroupBox("IDU Command (Type 0x81/0x82 — IDU→ODU)") resp_grid = QGridLayout(resp_group) fields_resp = [ ("mode_status", "Mode Status"), ("turbo", "Turbo (B6)"), ("setpoint", "CRT / Setpoint (B9)"), ("fan_speed", "IDU Fan Speed (B10)"), ("fan_mode", "Fan Mode (0x84 B4)"), ("idu_coil", "Indoor Coil (B11)"), ("idu_room", "Room Temp (B12)"), ] for row, (key, label) in enumerate(fields_resp): lbl = QLabel(label + ":") lbl.setStyleSheet("color: #FFCC80; font-weight: bold;") val = QLabel("—") val.setFont(QFont("Menlo", 16, QFont.Weight.Bold)) val.setStyleSheet("color: white;") resp_grid.addWidget(lbl, row, 0) resp_grid.addWidget(val, row, 1) self._lbl[key] = val layout.addWidget(resp_group) layout.addStretch() def update_from_packet(self, pkt: DecodedPacket, setpoint_offset: int = 0): if pkt.pkt_type == 0x01: self._lbl["mode"].setText(pkt.mode) self._set_mode_color("mode", pkt.mode) b16 = pkt.extra.get("B16", 0) b16_names = {0x50: "0x50 (normal)", 0x10: "0x10 (vent/fan-only)"} self._lbl["mode_flag"].setText(b16_names.get(b16, f"0x{b16:02X}")) freq_text = f"{pkt.compressor_freq} RPS ({pkt.compressor_rpm} RPM)" if pkt.compressor_freq == 0: freq_text = "OFF" self._lbl["freq"].setText(freq_text) self._lbl["valve"].setText(pkt.valve) self._lbl["onoff"].setText(pkt.on_off) self._set_mode_color("onoff", "ACTIVE" if pkt.on_off == "ACTIVE" else "STANDBY") # ODU actuator fields fan_text = f"{pkt.odu_fan}" if pkt.odu_fan == 0: fan_text += " (off)" self._lbl["odu_fan"].setStyleSheet("color: #90A4AE;") elif pkt.odu_fan > 80: fan_text += " (high)" self._lbl["odu_fan"].setStyleSheet("color: #FF8A65;") else: fan_text += f" (normal)" self._lbl["odu_fan"].setStyleSheet("color: #69F0AE;") self._lbl["odu_fan"].setText(fan_text) self._lbl["odu_b12"].setText(f"EEV pos: {pkt.odu_b12} (0x{pkt.odu_b12:02X})") elif pkt.pkt_type == 0x02: self._lbl["discharge"].setText(f"{pkt.discharge_temp:+.1f} °C (raw={pkt.raw[7]})") self._lbl["coil"].setText(f"{pkt.coil_temp:+.1f} °C (raw={pkt.raw[8]})") self._lbl["ambient"].setText(f"{pkt.ambient_temp:+.1f} °C (raw={pkt.raw[9]})") self._lbl["b11"].setText(f"raw={pkt.b11_val} (constant, not a sensor)") vbus = pkt.extra.get("VBUS", 0) self._lbl["vbus"].setText(f"{vbus} V (raw=0x{vbus:02X})") # Color discharge by temperature dt = pkt.discharge_temp if dt > 90: self._lbl["discharge"].setStyleSheet("color: #FF5252; font-weight: bold;") elif dt > 60: self._lbl["discharge"].setStyleSheet("color: #FFB74D; font-weight: bold;") else: self._lbl["discharge"].setStyleSheet("color: white;") elif pkt.pkt_type in (0x81, 0x82): self._lbl["mode_status"].setText(pkt.mode_status) self._set_mode_color("mode_status", pkt.mode_status) # Turbo flag turbo = pkt.extra.get("B6_turbo", 0) if turbo == 0x02: self._lbl["turbo"].setText("TURBO ACTIVE") self._lbl["turbo"].setStyleSheet("color: #FF5252; font-weight: bold;") else: self._lbl["turbo"].setText("OFF") self._lbl["turbo"].setStyleSheet("color: #90A4AE;") # CRT / setpoint b9_crt = pkt.extra.get("B9_CRT", 0) decoded = b9_crt - setpoint_offset offset_str = f"CRT-{setpoint_offset}" if setpoint_offset else "CRT=raw" self._lbl["setpoint"].setText( f"CRT={b9_crt} setpoint={decoded}°C ({offset_str})" ) self._lbl["setpoint"].setFont(QFont("Menlo", 13, QFont.Weight.Bold)) if 10 <= decoded <= 40: self._lbl["setpoint"].setStyleSheet("color: #69F0AE;") else: self._lbl["setpoint"].setStyleSheet("color: #EF5350;") # Fan speed with context coloring fan_spd = pkt.extra.get("B10_fan", 0) if fan_spd <= 60: fan_label = f"{fan_spd} (pre-start)" self._lbl["fan_speed"].setStyleSheet("color: #90A4AE;") elif fan_spd <= 85: fan_label = f"{fan_spd} (low/dry)" self._lbl["fan_speed"].setStyleSheet("color: white;") elif fan_spd <= 100: fan_label = f"{fan_spd} (normal)" self._lbl["fan_speed"].setStyleSheet("color: white;") elif fan_spd <= 115: fan_label = f"{fan_spd} (high/auto)" self._lbl["fan_speed"].setStyleSheet("color: #69F0AE;") else: fan_label = f"{fan_spd} (turbo)" self._lbl["fan_speed"].setStyleSheet("color: #FF8A65;") self._lbl["fan_speed"].setText(fan_label) # IDU coil and room temperature (raw ADC → estimated °C) coil = pkt.idu_coil_temp room = pkt.idu_room_temp coil_raw = pkt.extra.get("B11", 0) room_raw = pkt.extra.get("B12", 0) if not math.isnan(coil): self._lbl["idu_coil"].setText(f"{coil:+.1f} °C (raw={coil_raw})") else: self._lbl["idu_coil"].setText(f"— (raw={coil_raw})") if not math.isnan(room): self._lbl["idu_room"].setText(f"{room:+.1f} °C (raw={room_raw})") else: self._lbl["idu_room"].setText(f"— (raw={room_raw})") elif pkt.pkt_type == 0x84: fm = pkt.extra.get("B4_fan_mode", "—") self._lbl["fan_mode"].setText(fm) def _set_mode_color(self, key, mode_str): colors = { "HEAT": "#FF8A65", "HEATING": "#FF8A65", "COOL": "#4FC3F7", "COOLING": "#4FC3F7", "DRY": "#FFF176", "VENT": "#CE93D8", "STANDBY": "#90A4AE", "ACTIVE": "#69F0AE", } color = colors.get(mode_str, "white") self._lbl[key].setStyleSheet(f"color: {color}; font-weight: bold;") # ── Main Window ───────────────────────────────────────────────────────────── class BusMonitorWindow(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("TCL S-Line Bus Monitor") self.resize(1400, 900) self._parser = PacketParser() self._parser.packet_ready.connect(self._on_packet) self._parser.raw_byte.connect(self._on_raw_byte) self._parser.stats_updated.connect(self._on_stats) self._socket = None self._recording = False self._record_file = None self._replay_thread = None self._replay_worker = None self._raw_byte_buf = [] # buffer for batched hex view updates self._raw_flush_timer = QTimer() self._raw_flush_timer.setInterval(100) # flush every 100ms self._raw_flush_timer.timeout.connect(self._flush_raw_hex) self._raw_flush_timer.start() self._packet_count_window = deque(maxlen=200) # for rate calc # Plot data buffers self._plot_times = deque(maxlen=6000) # ~10 min at 10 pkt/s self._plot_discharge = deque(maxlen=6000) self._plot_coil = deque(maxlen=6000) self._plot_suction = deque(maxlen=6000) self._plot_ambient = deque(maxlen=6000) self._plot_setpoint = deque(maxlen=6000) self._plot_freq_times = deque(maxlen=6000) self._plot_freq = deque(maxlen=6000) # ODU actuator plot buffers (from type 0x01) self._plot_odu_times = deque(maxlen=6000) self._plot_odu_b12 = deque(maxlen=6000) self._plot_odu_fan = deque(maxlen=6000) self._build_ui() self._apply_dark_theme() def _build_ui(self): central = QWidget() self.setCentralWidget(central) main_layout = QVBoxLayout(central) main_layout.setContentsMargins(4, 4, 4, 4) main_layout.setSpacing(4) # ── Toolbar ── toolbar = QHBoxLayout() toolbar.addWidget(QLabel("Host:")) self._ip_edit = QLineEdit("192.168.1.100") self._ip_edit.setFixedWidth(140) toolbar.addWidget(self._ip_edit) toolbar.addWidget(QLabel("Port:")) self._port_spin = QSpinBox() self._port_spin.setRange(1, 65535) self._port_spin.setValue(333) self._port_spin.setFixedWidth(80) toolbar.addWidget(self._port_spin) self._connect_btn = QPushButton("Connect") self._connect_btn.setCheckable(True) self._connect_btn.clicked.connect(self._toggle_connection) self._connect_btn.setFixedWidth(100) toolbar.addWidget(self._connect_btn) self._status_lbl = QLabel(" Disconnected") self._status_lbl.setStyleSheet("color: #EF5350; font-weight: bold;") toolbar.addWidget(self._status_lbl) toolbar.addSpacing(20) self._record_btn = QPushButton("Record") self._record_btn.setCheckable(True) self._record_btn.clicked.connect(self._toggle_recording) toolbar.addWidget(self._record_btn) load_btn = QPushButton("Load JSONL...") load_btn.clicked.connect(self._load_file) toolbar.addWidget(load_btn) self._speed_spin = QSpinBox() self._speed_spin.setRange(1, 10000) self._speed_spin.setValue(100) self._speed_spin.setSuffix("x") self._speed_spin.setPrefix("Replay ") self._speed_spin.setFixedWidth(130) toolbar.addWidget(self._speed_spin) clear_btn = QPushButton("Clear") clear_btn.clicked.connect(self._clear_all) toolbar.addWidget(clear_btn) toolbar.addStretch() self._stats_lbl = QLabel("Packets: 0 | Errors: 0 | Rate: 0/s") self._stats_lbl.setStyleSheet("color: #B0BEC5;") toolbar.addWidget(self._stats_lbl) main_layout.addLayout(toolbar) # ── Main splitter (left | right) ── splitter = QSplitter(Qt.Orientation.Horizontal) # -- Left side: raw hex + packet table -- left = QWidget() left_layout = QVBoxLayout(left) left_layout.setContentsMargins(0, 0, 0, 0) # Raw hex view hex_group = QGroupBox("Raw Hex Stream") hex_layout = QVBoxLayout(hex_group) self._hex_view = QPlainTextEdit() self._hex_view.setReadOnly(True) self._hex_view.setFont(QFont("Menlo", 10)) self._hex_view.setMaximumBlockCount(5000) self._hex_view.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap) hex_layout.addWidget(self._hex_view) left_layout.addWidget(hex_group, stretch=2) # Packet log table pkt_group = QGroupBox("Packet Log") pkt_layout = QVBoxLayout(pkt_group) # Type filter checkboxes filter_row = QHBoxLayout() self._type_filters = {} for ptype, (short, desc, color) in PKT_TYPES.items(): cb = QCheckBox(f"0x{ptype:02X} {short}") cb.setChecked(True) cb.setStyleSheet(f"color: {color.name()};") self._type_filters[ptype] = cb filter_row.addWidget(cb) filter_row.addStretch() pkt_layout.addLayout(filter_row) self._pkt_table = QTableWidget() self._pkt_table.setColumnCount(5) self._pkt_table.setHorizontalHeaderLabels( ["Time", "Type", "Key Fields", "Extra", "Raw Hex"] ) self._pkt_table.horizontalHeader().setSectionResizeMode( QHeaderView.ResizeMode.ResizeToContents ) self._pkt_table.horizontalHeader().setSectionResizeMode( 4, QHeaderView.ResizeMode.Stretch ) self._pkt_table.setFont(QFont("Menlo", 10)) self._pkt_table.setAlternatingRowColors(True) pkt_layout.addWidget(self._pkt_table) left_layout.addWidget(pkt_group, stretch=3) splitter.addWidget(left) # -- Right side: dashboard + plots -- right = QWidget() right_layout = QVBoxLayout(right) right_layout.setContentsMargins(0, 0, 0, 0) self._dashboard = DashboardWidget() right_layout.addWidget(self._dashboard, stretch=2) # Temperature plot plot_group = QGroupBox("Temperature & Frequency Timeline") plot_layout = QVBoxLayout(plot_group) pg.setConfigOptions(antialias=True) self._plot = pg.PlotWidget() self._plot.setBackground("#1E1E1E") self._plot.showGrid(x=True, y=True, alpha=0.3) self._plot.setLabel("left", "Temperature", units="°C") self._plot.setLabel("bottom", "Time", units="s") self._plot.addLegend(offset=(10, 10)) self._curve_discharge = self._plot.plot( pen=pg.mkPen("#FF5252", width=2), name="Discharge (ODT, B7)" ) self._curve_coil = self._plot.plot( pen=pg.mkPen("#42A5F5", width=2), name="ODU Coil (OPT, B8)" ) self._curve_suction = self._plot.plot( pen=pg.mkPen("#66BB6A", width=2), name="Ambient (OAT, B9)" ) self._curve_ambient = self._plot.plot( pen=pg.mkPen("#78909C", width=1, style=Qt.PenStyle.DotLine), name="B11 (constant)", ) self._curve_setpoint = self._plot.plot( pen=pg.mkPen("#FFF176", width=2, style=Qt.PenStyle.DashLine), name="Setpoint (auto)", ) # Second Y-axis for compressor frequency self._freq_viewbox = pg.ViewBox() self._plot.scene().addItem(self._freq_viewbox) self._plot.getAxis("right").linkToView(self._freq_viewbox) self._freq_viewbox.setXLink(self._plot) self._plot.getAxis("right").setLabel("Frequency", units="RPS") self._plot.getAxis("right").setPen("#FFB74D") self._plot.showAxis("right") self._curve_freq = pg.PlotCurveItem( pen=pg.mkPen("#FFB74D", width=2, style=Qt.PenStyle.DotLine), name="Compressor RPS", ) self._freq_viewbox.addItem(self._curve_freq) # Keep viewboxes in sync on resize self._plot.getViewBox().sigResized.connect(self._sync_freq_viewbox) plot_layout.addWidget(self._plot) right_layout.addWidget(plot_group, stretch=3) # ODU actuator plot odu_plot_group = QGroupBox("ODU Actuators (Type 0x01 — Fan Speed / B12)") odu_plot_layout = QVBoxLayout(odu_plot_group) self._odu_plot = pg.PlotWidget() self._odu_plot.setBackground("#1E1E1E") self._odu_plot.showGrid(x=True, y=True, alpha=0.3) self._odu_plot.setLabel("left", "Fan speed / value") self._odu_plot.setLabel("bottom", "Time", units="s") self._odu_plot.addLegend(offset=(10, 10)) # Link X axis to main temp plot for synchronized scrolling self._odu_plot.setXLink(self._plot) self._curve_odu_fan = self._odu_plot.plot( pen=pg.mkPen("#CE93D8", width=2), name="B13 (ODU fan speed)" ) self._curve_odu_b12 = self._odu_plot.plot( pen=pg.mkPen("#78909C", width=1.5), name="B12 (EEV position)" ) # Add reference lines for typical fan speed values for val, label, color in [ (90, "max observed (90)", "#FF8A65"), (70, "typical steady (70)", "#69F0AE"), ]: line = pg.InfiniteLine( pos=val, angle=0, pen=pg.mkPen(color, width=1, style=Qt.PenStyle.DashLine), label=label, labelOpts={"color": color, "position": 0.05}, ) self._odu_plot.addItem(line) odu_plot_layout.addWidget(self._odu_plot) right_layout.addWidget(odu_plot_group, stretch=2) splitter.addWidget(right) splitter.setSizes([500, 900]) main_layout.addWidget(splitter) # Plot update timer (don't update on every packet — batch at 10 Hz) self._plot_timer = QTimer() self._plot_timer.setInterval(100) self._plot_timer.timeout.connect(self._update_plots) self._plot_timer.start() # Rate calculation timer self._rate_timer = QTimer() self._rate_timer.setInterval(1000) self._rate_timer.timeout.connect(self._update_rate) self._rate_timer.start() def _sync_freq_viewbox(self): self._freq_viewbox.setGeometry(self._plot.getViewBox().sceneBoundingRect()) def _apply_dark_theme(self): app = QApplication.instance() app.setStyle("Fusion") palette = QPalette() palette.setColor(QPalette.ColorRole.Window, QColor(30, 30, 30)) palette.setColor(QPalette.ColorRole.WindowText, QColor(220, 220, 220)) palette.setColor(QPalette.ColorRole.Base, QColor(25, 25, 25)) palette.setColor(QPalette.ColorRole.AlternateBase, QColor(35, 35, 35)) palette.setColor(QPalette.ColorRole.ToolTipBase, QColor(40, 40, 40)) palette.setColor(QPalette.ColorRole.ToolTipText, QColor(220, 220, 220)) palette.setColor(QPalette.ColorRole.Text, QColor(220, 220, 220)) palette.setColor(QPalette.ColorRole.Button, QColor(45, 45, 45)) palette.setColor(QPalette.ColorRole.ButtonText, QColor(220, 220, 220)) palette.setColor(QPalette.ColorRole.BrightText, QColor(255, 50, 50)) palette.setColor(QPalette.ColorRole.Link, QColor(66, 165, 245)) palette.setColor(QPalette.ColorRole.Highlight, QColor(66, 165, 245)) palette.setColor(QPalette.ColorRole.HighlightedText, QColor(0, 0, 0)) app.setPalette(palette) # ── Connection ── def _toggle_connection(self, checked): if checked: self._connect() else: self._disconnect() def _connect(self): host = self._ip_edit.text().strip() port = self._port_spin.value() self._socket = QTcpSocket() self._socket.readyRead.connect(self._on_socket_data) self._socket.connected.connect(self._on_connected) self._socket.disconnected.connect(self._on_disconnected) self._socket.errorOccurred.connect(self._on_socket_error) self._socket.connectToHost(host, port) self._status_lbl.setText(f" Connecting to {host}:{port}...") self._status_lbl.setStyleSheet("color: #FFF176; font-weight: bold;") def _disconnect(self): if self._socket: self._socket.close() self._socket = None self._connect_btn.setChecked(False) self._status_lbl.setText(" Disconnected") self._status_lbl.setStyleSheet("color: #EF5350; font-weight: bold;") def _on_connected(self): self._status_lbl.setText(" Connected") self._status_lbl.setStyleSheet("color: #69F0AE; font-weight: bold;") def _on_disconnected(self): self._status_lbl.setText(" Disconnected") self._status_lbl.setStyleSheet("color: #EF5350; font-weight: bold;") self._connect_btn.setChecked(False) def _on_socket_error(self, error): msg = self._socket.errorString() if self._socket else str(error) self._status_lbl.setText(f" Error: {msg}") self._status_lbl.setStyleSheet("color: #EF5350; font-weight: bold;") self._connect_btn.setChecked(False) def _on_socket_data(self): if not self._socket: return data = self._socket.readAll() ts = time.time() for i in range(data.size()): byte_val = data.at(i) if isinstance(byte_val, bytes): byte_val = byte_val[0] self._parser.feed(ts, byte_val) # Record if enabled if self._recording and self._record_file: rec = {"ts": ts, "src": "rx", "hex": f"{byte_val:02x}", "err": False} self._record_file.write(json.dumps(rec) + "\n") # ── Recording ── def _toggle_recording(self, checked): if checked: ts_str = time.strftime("%Y%m%d_%H%M%S") filepath = Path(f"capture_{ts_str}.jsonl") self._record_file = open(filepath, "w") self._recording = True self._record_btn.setText(f"Recording: {filepath.name}") self._record_btn.setStyleSheet("color: #FF5252; font-weight: bold;") else: self._recording = False if self._record_file: self._record_file.close() self._record_file = None self._record_btn.setText("Record") self._record_btn.setStyleSheet("") # ── File replay ── def _load_file(self): filepath, _ = QFileDialog.getOpenFileName( self, "Open Capture File", ".", "JSONL files (*.jsonl);;All files (*)" ) if not filepath: return self._clear_all() self._status_lbl.setText(f" Replaying: {Path(filepath).name}") self._status_lbl.setStyleSheet("color: #CE93D8; font-weight: bold;") self._replay_worker = ReplayWorker(filepath, self._speed_spin.value()) self._replay_thread = QThread() self._replay_worker.moveToThread(self._replay_thread) self._replay_worker.byte_ready.connect(self._parser.feed) self._replay_worker.finished.connect(self._on_replay_done) self._replay_worker.progress.connect(self._on_replay_progress) self._replay_thread.started.connect(self._replay_worker.run) self._replay_thread.start() def _on_replay_done(self): self._status_lbl.setText(" Replay complete") self._status_lbl.setStyleSheet("color: #69F0AE; font-weight: bold;") if self._replay_thread: self._replay_thread.quit() self._replay_thread.wait() self._replay_thread = None self._replay_worker = None def _on_replay_progress(self, current, total): pct = current * 100 // max(total, 1) self._status_lbl.setText( f" Replaying... {pct}% ({current}/{total} bytes)" ) # ── Packet handling ── def _on_packet(self, pkt: DecodedPacket): self._packet_count_window.append(time.monotonic()) # Update dashboard (pass detected setpoint offset) self._dashboard.update_from_packet(pkt, self._parser.setpoint_offset) # Update plot data if pkt.pkt_type == 0x02: self._plot_times.append(pkt.rel_time) self._plot_discharge.append(pkt.discharge_temp) self._plot_coil.append(pkt.coil_temp) self._plot_suction.append(pkt.ambient_temp) # OAT (Steinhart-Hart from raw ADC) self._plot_ambient.append(pkt.b11_val) # constant B11 raw, plot for reference elif pkt.pkt_type in (0x81, 0x82): # Plot setpoint using auto-detected CRT offset b9_crt = pkt.extra.get("B9_CRT", 0) self._plot_setpoint.append(b9_crt - self._parser.setpoint_offset) # Use last sensor timestamp for setpoint x-axis alignment if self._plot_times: pass # setpoint plotted at sensor times below elif pkt.pkt_type == 0x01: self._plot_freq_times.append(pkt.rel_time) self._plot_freq.append(pkt.compressor_freq) # ODU actuators self._plot_odu_times.append(pkt.rel_time) self._plot_odu_b12.append(pkt.odu_b12) self._plot_odu_fan.append(pkt.odu_fan) # Add to packet log table (check filter) if pkt.pkt_type in self._type_filters: if not self._type_filters[pkt.pkt_type].isChecked(): return self._add_packet_row(pkt) def _add_packet_row(self, pkt: DecodedPacket): table = self._pkt_table row = table.rowCount() # Limit rows if row > 5000: table.removeRow(0) row = table.rowCount() table.insertRow(row) type_info = PKT_TYPES.get(pkt.pkt_type, ("?", "Unknown", QColor("white"))) color = type_info[2] # Time mins = int(pkt.rel_time // 60) secs = pkt.rel_time % 60 time_item = QTableWidgetItem(f"{mins:2d}:{secs:05.2f}") time_item.setForeground(color) table.setItem(row, 0, time_item) # Type type_item = QTableWidgetItem(f"0x{pkt.pkt_type:02X} {type_info[0]}") type_item.setForeground(color) table.setItem(row, 1, type_item) # Key fields key_parts = [] if pkt.pkt_type == 0x01: key_parts = [ f"mode={pkt.mode}", f"freq={pkt.compressor_freq}", f"valve={pkt.valve}", f"{'ON' if pkt.on_off == 'ACTIVE' else 'OFF'}", f"fan={pkt.odu_fan}", f"B12={pkt.odu_b12}", ] elif pkt.pkt_type == 0x02: key_parts = [ f"ODT={pkt.discharge_temp:+.1f}°", f"OPT={pkt.coil_temp:+.1f}°", f"OAT={pkt.ambient_temp:+.1f}°", ] elif pkt.pkt_type in (0x81, 0x82): b9_raw = pkt.extra.get("B9_raw", 0) v16 = pkt.extra.get("B9_v16", b9_raw - 16) key_parts = [ f"status={pkt.mode_status}", f"B9=0x{b9_raw:02X}", f"v-32={b9_raw-32}°", f"v-16={v16}°", ] key_item = QTableWidgetItem(" ".join(key_parts)) key_item.setForeground(color) table.setItem(row, 2, key_item) # Extra fields extra_str = " ".join(f"{k}={v}" for k, v in pkt.extra.items()) extra_item = QTableWidgetItem(extra_str) extra_item.setForeground(QColor("#78909C")) table.setItem(row, 3, extra_item) # Raw hex hex_str = " ".join(f"{b:02X}" for b in pkt.raw) hex_item = QTableWidgetItem(hex_str) hex_item.setForeground(QColor("#616161")) hex_item.setFont(QFont("Menlo", 9)) table.setItem(row, 4, hex_item) table.scrollToBottom() def _on_raw_byte(self, ts: float, byte_val: int): self._raw_byte_buf.append((ts, byte_val)) def _flush_raw_hex(self): """Batch-update the hex view to avoid per-byte GUI updates.""" if not self._raw_byte_buf: return lines = [] # Group into lines of 19 bytes (one packet width) buf = self._raw_byte_buf self._raw_byte_buf = [] # Simple approach: show timestamped hex, highlight sync bytes line_bytes = [] line_ts = None for ts, bv in buf: if bv == SYNC and line_bytes: # Flush previous line if line_ts is not None: t0 = self._parser.t0 or 0 rel = ts - t0 if t0 else 0 mins = int(rel // 60) secs = rel % 60 hex_str = " ".join(f"{b:02X}" for b in line_bytes) lines.append(f"{mins:2d}:{secs:05.2f} {hex_str}") line_bytes = [bv] line_ts = ts else: if line_ts is None: line_ts = ts line_bytes.append(bv) # Flush remaining if line_bytes and line_ts is not None: t0 = self._parser.t0 or 0 rel = line_ts - t0 if t0 else 0 mins = int(rel // 60) secs = rel % 60 hex_str = " ".join(f"{b:02X}" for b in line_bytes) lines.append(f"{mins:2d}:{secs:05.2f} {hex_str}") if lines: self._hex_view.appendPlainText("\n".join(lines)) def _on_stats(self, valid, errors, resyncs): pass # Rate updated by timer def _update_rate(self): now = time.monotonic() # Count packets in last second while self._packet_count_window and self._packet_count_window[0] < now - 1.0: self._packet_count_window.popleft() rate = len(self._packet_count_window) valid = self._parser._valid errors = self._parser._errors variant = self._parser.unit_variant_84_b4 variant_str = "" if variant is not None: offset = self._parser.setpoint_offset variant_str = f" | Unit: 0x84.B4=0x{variant:02X} (setpoint=B9-{offset})" self._stats_lbl.setText( f"Packets: {valid} | Errors: {errors} | Rate: {rate}/s{variant_str}" ) def _update_plots(self): if not self._plot_times: return times = np.array(self._plot_times) self._curve_discharge.setData(times, np.array(self._plot_discharge)) self._curve_coil.setData(times, np.array(self._plot_coil)) self._curve_suction.setData(times, np.array(self._plot_suction)) self._curve_ambient.setData(times, np.array(self._plot_ambient)) # Setpoint: align to sensor times (one setpoint per sensor reading) if self._plot_setpoint: sp = list(self._plot_setpoint) # Pad or trim to match sensor times if len(sp) < len(times): sp.extend([sp[-1]] * (len(times) - len(sp))) elif len(sp) > len(times): sp = sp[:len(times)] self._curve_setpoint.setData(times, np.array(sp)) # Frequency on second axis if self._plot_freq_times: ft = np.array(self._plot_freq_times) fv = np.array(self._plot_freq) self._curve_freq.setData(ft, fv) self._freq_viewbox.setRange( xRange=(times[0], times[-1]), yRange=(0, max(fv.max() + 5, 10)), padding=0, ) # ODU actuator plots if self._plot_odu_times: et = np.array(self._plot_odu_times) self._curve_odu_fan.setData(et, np.array(self._plot_odu_fan)) self._curve_odu_b12.setData(et, np.array(self._plot_odu_b12)) # ── Clear ── def _clear_all(self): # Stop replay if running if self._replay_worker: self._replay_worker.stop() if self._replay_thread: self._replay_thread.quit() self._replay_thread.wait() self._replay_thread = None self._replay_worker = None self._parser.reset() self._hex_view.clear() self._pkt_table.setRowCount(0) self._raw_byte_buf.clear() self._packet_count_window.clear() self._plot_times.clear() self._plot_discharge.clear() self._plot_coil.clear() self._plot_suction.clear() self._plot_ambient.clear() self._plot_setpoint.clear() self._plot_freq_times.clear() self._plot_freq.clear() self._plot_odu_times.clear() self._plot_odu_b12.clear() self._plot_odu_fan.clear() self._curve_discharge.clear() self._curve_coil.clear() self._curve_suction.clear() self._curve_ambient.clear() self._curve_setpoint.clear() self._curve_freq.clear() self._curve_odu_fan.clear() self._curve_odu_b12.clear() def closeEvent(self, event): self._disconnect() if self._recording: self._toggle_recording(False) if self._replay_worker: self._replay_worker.stop() if self._replay_thread: self._replay_thread.quit() self._replay_thread.wait() event.accept() # ── Entry point ───────────────────────────────────────────────────────────── def main(): app = QApplication(sys.argv) app.setApplicationName("TCL S-Line Bus Monitor") window = BusMonitorWindow() window.show() # If capture file passed as argument, load it if len(sys.argv) > 1: filepath = sys.argv[1] if Path(filepath).exists(): QTimer.singleShot(500, lambda: _auto_load(window, filepath)) sys.exit(app.exec()) def _auto_load(window, filepath): """Auto-load a file passed as CLI argument.""" window._parser.reset() window._status_lbl.setText(f" Replaying: {Path(filepath).name}") window._status_lbl.setStyleSheet("color: #CE93D8; font-weight: bold;") worker = ReplayWorker(filepath, window._speed_spin.value()) thread = QThread() worker.moveToThread(thread) worker.byte_ready.connect(window._parser.feed) worker.finished.connect(window._on_replay_done) worker.progress.connect(window._on_replay_progress) thread.started.connect(worker.run) window._replay_worker = worker window._replay_thread = thread thread.start() if __name__ == "__main__": main()