"""MPPT Testbench GUI -- live measurements + full instrument control. Tkinter app with embedded matplotlib graphs and threaded instrument I/O. """ from __future__ import annotations import csv import math import os import queue import subprocess import sys import threading import time import tkinter as tk from tkinter import ttk, messagebox, filedialog from collections import deque import matplotlib matplotlib.use("TkAgg") from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg, NavigationToolbar2Tk from matplotlib.figure import Figure from it6500.driver import IT6500 from prodigit3366g.driver import Prodigit3366G from hioki3193.driver import Hioki3193 from testbench.bench import MPPTTestbench from testbench.gui_workers import InstrumentWorker, Cmd, STM32Worker from testbench.stm32_link import ( FLAG_NAMES, FLAG_INFO_MASK, build_ping, build_clear_flags, ) ERROR_THRESHOLD = 1e90 POLL_MS = 200 # GUI poll interval for checking worker data GRAPH_WINDOW_S = 20.0 # all live graphs show the last 20 seconds TELEM_DECIMATE = 4 # telemetry display decimation (full rate still recorded) MIN_REDRAW_S = 0.4 # throttle canvas redraws STM_EMA_TAU_S = 2.0 # EMA time constant for displayed telemetry values # HV supply capability guard: the PSU can source at most this much current. # Load steps whose estimated input draw exceeds it are rejected, regardless # of what the sweep asks for. The estimate is conservative (low assumed # efficiency) so rejection kicks in before the PSU actually current-limits. PSU_MAX_CURRENT_A = 20.0 VOUT_NOM_V = 48.0 # nominal converter output, fallback when unmeasured EFF_ASSUMED = 0.90 # conservative efficiency for input-current estimates # Thermal sweep guard: firmware trips at ETEMP_MAX 60 C (heatsink, readout # ceiling 61.6 C) and BTEMP_MAX 80 C (board) -- code64 main.c. The sweep # pauses shortly before that, holds the load at ~1 A until both temps have # cooled below pause - hysteresis, then resumes where it left off. ETEMP_PAUSE_C = 57.0 BTEMP_PAUSE_C = 77.0 TEMP_RESUME_HYST_C = 5.0 THERMAL_HOLD_LOAD_A = 1.0 TELEM_STALE_S = 2.0 # ignore STM32 telemetry older than this def _est_input_current(load_mode: str, setpoint: float, vin: float, vout: float) -> float: """Conservative PSU input-current estimate for a converter load step.""" p_out = setpoint * vout if load_mode == "CC" else setpoint return p_out / (EFF_ASSUMED * max(vin, 1.0)) # Compact flag names for the 2-column checkbox list (bit -> short label); # full names (FLAG_NAMES) are still used for console messages. FLAG_SHORT = { 0: "STARTUP wait", 1: "PRECHG TIMEOUT", 2: "GUARD VIN_MAX", 3: "GUARD VIN str: """Format a measurement value, showing '---' for error codes.""" if abs(val) >= ERROR_THRESHOLD: return "---" return f"{val:+.{decimals}f}" def _fmt_eng(val: float) -> str: """Format in engineering notation, showing '---' for error codes.""" if abs(val) >= ERROR_THRESHOLD: return "---" return f"{val:+.4E}" def _clean(val: float) -> float: """Replace HIOKI error codes with NaN for plotting.""" return float("nan") if abs(val) >= ERROR_THRESHOLD else val class _ConsoleRedirector: """Redirects stdout writes to the GUI console widget.""" def __init__(self, gui: "TestbenchGUI") -> None: self._gui = gui self._orig = sys.stdout self._buf = "" def write(self, text: str) -> None: if self._orig: self._orig.write(text) # Buffer partial lines, flush on newline self._buf += text while "\n" in self._buf: line, self._buf = self._buf.split("\n", 1) line = line.strip() if line: self._gui._console(line) def flush(self) -> None: if self._orig: self._orig.flush() class TestbenchGUI(tk.Tk): """Main GUI window.""" def __init__(self) -> None: super().__init__() self.title("MPPT Testbench Control Panel") self.geometry("1860x1020") self.minsize(1400, 900) self.bench: MPPTTestbench | None = None self.worker: InstrumentWorker | None = None self._log_file = None self._log_writer = None self._log_count = 0 self._last_log_path: str | None = None self._latest_data: dict = {} self._point_count = 0 self._t0 = time.time() self._t0_mono = time.monotonic() # single session timebase for all graphs # STM32 telemetry link self.stm32: STM32Worker | None = None self._telem_log_path: str | None = None self._stm_labels: dict[str, tk.Widget] = {} self._stm_ema: dict[str, float] = {} self._stm_ema_t: float | None = None self._last_flags = 0 self._flags_shown = -1 self._last_pong: int | None = None self._rate_buckets: deque = deque() self._last_draw = 0.0 # Graph data self._history = 300 self._timestamps: deque[float] = deque(maxlen=self._history) self._series: dict[str, deque[float]] = { k: deque(maxlen=self._history) for k in [ "supply_P", "load_P", "meter_P5", "meter_P6", "meter_EFF1", "meter_U5", "meter_U6", "meter_I5", "meter_I6", ] } self._meter_fmt_sci = True # True=scientific, False=normal self._sweep_data_queue: queue.Queue = queue.Queue(maxsize=100) self._build_ui() self.protocol("WM_DELETE_WINDOW", self._on_close) # Logging is on by default: auto-start into logs/data_.csv # (Stop Log / Start Log still work normally for custom paths). logdir = os.path.join(os.getcwd(), "logs") try: os.makedirs(logdir, exist_ok=True) self._start_log(os.path.join( logdir, time.strftime("data_%Y%m%d_%H%M%S.csv"))) except OSError as e: self._console(f"Auto-log failed ({e}) - use Start Log manually", "error") # ── UI Construction ─────────────────────────────────────────────── def _build_ui(self) -> None: # Top bar self._build_connection_bar() # Main content: left controls + right graphs content = ttk.PanedWindow(self, orient=tk.HORIZONTAL) content.pack(fill=tk.BOTH, expand=True, padx=4, pady=4) # Left panel: two columns of controls left_outer = ttk.Frame(content) left_outer.columnconfigure(0, weight=1) left_outer.columnconfigure(1, weight=1) content.add(left_outer, weight=0) # Column 0: Supply + Load left_col0 = ttk.Frame(left_outer) left_col0.grid(row=0, column=0, sticky="nsew", padx=(0, 2)) self._build_supply_controls(left_col0) self._build_load_controls(left_col0) self._build_logging_controls(left_col0) self._build_profile_controls(left_col0) # Column 1: Meter left_col1 = ttk.Frame(left_outer) left_col1.grid(row=0, column=1, sticky="nsew", padx=(2, 0)) self._build_meter_readout(left_col1) self._build_sweep_vi_controls(left_col1) # Right panel: graphs + console right_pane = ttk.PanedWindow(content, orient=tk.VERTICAL) content.add(right_pane, weight=1) graph_frame = ttk.Frame(right_pane) right_pane.add(graph_frame, weight=3) self._build_graphs(graph_frame) console_frame = ttk.Frame(right_pane) right_pane.add(console_frame, weight=1) self._build_console(console_frame) # Rightmost panel: STM32 telemetry telem_outer = ttk.Frame(content) content.add(telem_outer, weight=0) self._build_stm32_panel(telem_outer) # Status bar self._build_status_bar() def _build_connection_bar(self) -> None: bar = ttk.Frame(self) bar.pack(fill=tk.X, padx=4, pady=(4, 0)) # Connection fields ttk.Label(bar, text="Supply:").pack(side=tk.LEFT) self._supply_addr = ttk.Entry(bar, width=20) self._supply_addr.insert(0, "auto") self._supply_addr.pack(side=tk.LEFT, padx=(2, 8)) ttk.Label(bar, text="Load:").pack(side=tk.LEFT) self._load_port = ttk.Entry(bar, width=8) self._load_port.insert(0, "COM11") self._load_port.pack(side=tk.LEFT, padx=(2, 4)) ttk.Label(bar, text="@").pack(side=tk.LEFT) self._load_baud = ttk.Entry(bar, width=7) self._load_baud.insert(0, "115200") self._load_baud.pack(side=tk.LEFT, padx=(2, 8)) ttk.Label(bar, text="Meter:").pack(side=tk.LEFT) self._meter_addr = ttk.Entry(bar, width=20) self._meter_addr.insert(0, "auto") self._meter_addr.pack(side=tk.LEFT, padx=(2, 8)) ttk.Label(bar, text="STM32:").pack(side=tk.LEFT) self._stm32_port = ttk.Entry(bar, width=7) self._stm32_port.insert(0, "COM4") self._stm32_port.pack(side=tk.LEFT, padx=(2, 4)) self._btn_stm32_connect = ttk.Button(bar, text="Link STM32", command=self._connect_stm32) self._btn_stm32_connect.pack(side=tk.LEFT, padx=2) self._btn_stm32_disconnect = ttk.Button( bar, text="Unlink", command=self._disconnect_stm32, state=tk.DISABLED) self._btn_stm32_disconnect.pack(side=tk.LEFT, padx=(2, 8)) self._btn_connect = ttk.Button(bar, text="Connect", command=self._connect) self._btn_connect.pack(side=tk.LEFT, padx=2) self._btn_setup = ttk.Button(bar, text="Setup All", command=self._setup_all, state=tk.DISABLED) self._btn_setup.pack(side=tk.LEFT, padx=2) self._btn_disconnect = ttk.Button(bar, text="Disconnect", command=self._disconnect, state=tk.DISABLED) self._btn_disconnect.pack(side=tk.LEFT, padx=2) # SAFE OFF - always visible, right side self._btn_safe_off = tk.Button( bar, text="SAFE OFF", bg="#cc0000", fg="white", font=("TkDefaultFont", 13, "bold"), width=10, command=self._safe_off, activebackground="#ff0000", ) self._btn_safe_off.pack(side=tk.RIGHT, padx=4) self.bind("", lambda e: self._safe_off()) def _build_supply_controls(self, parent) -> None: frame = ttk.LabelFrame(parent, text="DC Supply (IT6500D)", padding=8) frame.pack(fill=tk.X, padx=4, pady=4) # Setpoints row = ttk.Frame(frame) row.pack(fill=tk.X, pady=2) ttk.Label(row, text="Voltage (V):", width=12).pack(side=tk.LEFT) self._sup_voltage = ttk.Entry(row, width=10) self._sup_voltage.insert(0, "75.0") self._sup_voltage.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set V", width=6, command=self._set_supply_voltage).pack(side=tk.LEFT, padx=2) row = ttk.Frame(frame) row.pack(fill=tk.X, pady=2) ttk.Label(row, text="Current (A):", width=12).pack(side=tk.LEFT) self._sup_current = ttk.Entry(row, width=10) self._sup_current.insert(0, "10.0") self._sup_current.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set I", width=6, command=self._set_supply_current).pack(side=tk.LEFT, padx=2) row = ttk.Frame(frame) row.pack(fill=tk.X, pady=2) ttk.Button(row, text="Apply V+I", command=self._apply_supply).pack(side=tk.LEFT, padx=2) self._btn_sup_on = ttk.Button(row, text="Output ON", command=lambda: self._send(Cmd.OUTPUT_ON)) self._btn_sup_on.pack(side=tk.LEFT, padx=2) self._btn_sup_off = ttk.Button(row, text="Output OFF", command=lambda: self._send(Cmd.OUTPUT_OFF)) self._btn_sup_off.pack(side=tk.LEFT, padx=2) # OVP / Slew adv = ttk.LabelFrame(frame, text="Protection / Slew", padding=4) adv.pack(fill=tk.X, pady=4) row = ttk.Frame(adv) row.pack(fill=tk.X, pady=1) ttk.Label(row, text="OVP (V):", width=12).pack(side=tk.LEFT) self._sup_ovp = ttk.Entry(row, width=10) self._sup_ovp.insert(0, "120.0") self._sup_ovp.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set", width=4, command=self._set_ovp).pack(side=tk.LEFT, padx=2) row = ttk.Frame(adv) row.pack(fill=tk.X, pady=1) ttk.Label(row, text="Rise (s):", width=12).pack(side=tk.LEFT) self._sup_rise = ttk.Entry(row, width=10) self._sup_rise.insert(0, "0.1") self._sup_rise.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set", width=4, command=lambda: self._send_float(Cmd.SET_RISE_TIME, self._sup_rise)).pack(side=tk.LEFT, padx=2) row = ttk.Frame(adv) row.pack(fill=tk.X, pady=1) ttk.Label(row, text="Fall (s):", width=12).pack(side=tk.LEFT) self._sup_fall = ttk.Entry(row, width=10) self._sup_fall.insert(0, "0.1") self._sup_fall.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set", width=4, command=lambda: self._send_float(Cmd.SET_FALL_TIME, self._sup_fall)).pack(side=tk.LEFT, padx=2) # Live readout readout = ttk.LabelFrame(frame, text="Measured", padding=4) readout.pack(fill=tk.X, pady=4) self._sup_v_label = ttk.Label(readout, text="V: ---", font=("Consolas", 11)) self._sup_v_label.pack(anchor=tk.W) self._sup_i_label = ttk.Label(readout, text="I: ---", font=("Consolas", 11)) self._sup_i_label.pack(anchor=tk.W) self._sup_p_label = ttk.Label(readout, text="P: ---", font=("Consolas", 11)) self._sup_p_label.pack(anchor=tk.W) self._sup_state_label = tk.Label( readout, text="OUTPUT: ---", font=("Consolas", 11, "bold"), fg="#888888", ) self._sup_state_label.pack(anchor=tk.W, pady=(4, 0)) def _build_load_controls(self, parent) -> None: frame = ttk.LabelFrame(parent, text="DC Load (Prodigit 3366G)", padding=8) frame.pack(fill=tk.X, padx=4, pady=4) # Mode row = ttk.Frame(frame) row.pack(fill=tk.X, pady=2) ttk.Label(row, text="Mode:", width=12).pack(side=tk.LEFT) self._load_mode = ttk.Combobox(row, values=["CC", "CR", "CV", "CP"], width=6, state="readonly") self._load_mode.set("CC") self._load_mode.pack(side=tk.LEFT, padx=2) self._load_mode.bind("<>", self._on_mode_change) ttk.Button(row, text="Set Mode", command=self._set_load_mode).pack(side=tk.LEFT, padx=2) # Value row = ttk.Frame(frame) row.pack(fill=tk.X, pady=2) self._load_unit_label = ttk.Label(row, text="Value (A):", width=12) self._load_unit_label.pack(side=tk.LEFT) self._load_value = ttk.Entry(row, width=10) self._load_value.insert(0, "5.0") self._load_value.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set", width=6, command=self._set_load_value).pack(side=tk.LEFT, padx=2) # On/Off row = ttk.Frame(frame) row.pack(fill=tk.X, pady=2) ttk.Button(row, text="Load ON", command=lambda: self._send(Cmd.LOAD_ON)).pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Load OFF", command=lambda: self._send(Cmd.LOAD_OFF)).pack(side=tk.LEFT, padx=2) # Slew rate adv = ttk.LabelFrame(frame, text="Slew Rate", padding=4) adv.pack(fill=tk.X, pady=4) row = ttk.Frame(adv) row.pack(fill=tk.X, pady=1) ttk.Label(row, text="Rise (A/us):", width=12).pack(side=tk.LEFT) self._load_rise = ttk.Entry(row, width=10) self._load_rise.insert(0, "1.0") self._load_rise.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set", width=4, command=lambda: self._send_float(Cmd.SET_SLEW_RISE, self._load_rise)).pack(side=tk.LEFT, padx=2) row = ttk.Frame(adv) row.pack(fill=tk.X, pady=1) ttk.Label(row, text="Fall (A/us):", width=12).pack(side=tk.LEFT) self._load_fall = ttk.Entry(row, width=10) self._load_fall.insert(0, "1.0") self._load_fall.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set", width=4, command=lambda: self._send_float(Cmd.SET_SLEW_FALL, self._load_fall)).pack(side=tk.LEFT, padx=2) # Live readout readout = ttk.LabelFrame(frame, text="Measured", padding=4) readout.pack(fill=tk.X, pady=4) self._load_v_label = ttk.Label(readout, text="V: ---", font=("Consolas", 11)) self._load_v_label.pack(anchor=tk.W) self._load_i_label = ttk.Label(readout, text="I: ---", font=("Consolas", 11)) self._load_i_label.pack(anchor=tk.W) self._load_p_label = ttk.Label(readout, text="P: ---", font=("Consolas", 11)) self._load_p_label.pack(anchor=tk.W) self._load_state_label = tk.Label( readout, text="LOAD: ---", font=("Consolas", 11, "bold"), fg="#888888", ) self._load_state_label.pack(anchor=tk.W, pady=(4, 0)) def _build_meter_readout(self, parent) -> None: frame = ttk.LabelFrame(parent, text="Power Analyzer (HIOKI 3193-10)", padding=8) frame.pack(fill=tk.X, padx=4, pady=4) # Meter controls ctrl = ttk.LabelFrame(frame, text="Settings", padding=4) ctrl.pack(fill=tk.X, pady=4) row = ttk.Frame(ctrl) row.pack(fill=tk.X, pady=1) ttk.Label(row, text="Wiring:", width=10).pack(side=tk.LEFT) self._meter_wiring = ttk.Combobox(row, values=["1P2W", "1P3W", "3P3W", "3V3A", "3P4W"], width=6, state="readonly") self._meter_wiring.set("1P2W") self._meter_wiring.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set", width=4, command=lambda: self._send(Cmd.SET_WIRING, self._meter_wiring.get())).pack(side=tk.LEFT, padx=2) row = ttk.Frame(ctrl) row.pack(fill=tk.X, pady=1) ttk.Label(row, text="Speed:", width=10).pack(side=tk.LEFT) self._meter_speed = ttk.Combobox(row, values=["FAST", "MID", "SLOW"], width=6, state="readonly") self._meter_speed.set("FAST") self._meter_speed.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set", width=4, command=lambda: self._send(Cmd.SET_RESPONSE_SPEED, self._meter_speed.get())).pack(side=tk.LEFT, padx=2) # Ch5 coupling row = ttk.Frame(ctrl) row.pack(fill=tk.X, pady=1) ttk.Label(row, text="Ch5 coup:", width=10).pack(side=tk.LEFT) self._ch5_coupling = ttk.Combobox(row, values=["DC", "AC", "ACDC"], width=6, state="readonly") self._ch5_coupling.set("DC") self._ch5_coupling.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set", width=4, command=lambda: self._send(Cmd.SET_COUPLING, 5, self._ch5_coupling.get())).pack(side=tk.LEFT, padx=2) # Ch6 coupling row = ttk.Frame(ctrl) row.pack(fill=tk.X, pady=1) ttk.Label(row, text="Ch6 coup:", width=10).pack(side=tk.LEFT) self._ch6_coupling = ttk.Combobox(row, values=["DC", "AC", "ACDC"], width=6, state="readonly") self._ch6_coupling.set("DC") self._ch6_coupling.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set", width=4, command=lambda: self._send(Cmd.SET_COUPLING, 6, self._ch6_coupling.get())).pack(side=tk.LEFT, padx=2) # Format selector fmt_row = ttk.Frame(frame) fmt_row.pack(fill=tk.X, pady=2) ttk.Label(fmt_row, text="Format:", width=10).pack(side=tk.LEFT) self._meter_fmt_combo = ttk.Combobox( fmt_row, values=["Scientific", "Normal"], width=10, state="readonly" ) self._meter_fmt_combo.set("Scientific") self._meter_fmt_combo.pack(side=tk.LEFT, padx=2) self._meter_fmt_combo.bind("<>", self._on_meter_fmt_change) # Voltage ranges: 6,15,30,60,150,300,600,1000 v_ranges = ["AUTO", "6", "15", "30", "60", "150", "300", "600", "1000"] # Current ranges depend on clamp sensor; common values i_ranges = ["AUTO", "0.5", "1", "2", "5", "10", "20", "50", "100", "200", "500", "1000"] # Input side input_frame = ttk.LabelFrame(frame, text="Input (Ch5 - Solar)", padding=4) input_frame.pack(fill=tk.X, pady=2) self._m_u5 = ttk.Label(input_frame, text="U5: ---", font=("Consolas", 11)) self._m_u5.pack(anchor=tk.W) self._m_i5 = ttk.Label(input_frame, text="I5: ---", font=("Consolas", 11)) self._m_i5.pack(anchor=tk.W) self._m_p5 = ttk.Label(input_frame, text="P5: ---", font=("Consolas", 11)) self._m_p5.pack(anchor=tk.W) # Ch5 ranges rng5 = ttk.Frame(input_frame) rng5.pack(fill=tk.X, pady=(4, 0)) ttk.Label(rng5, text="V range:").pack(side=tk.LEFT) self._ch5_v_range = ttk.Combobox(rng5, values=v_ranges, width=6, state="readonly") self._ch5_v_range.set("AUTO") self._ch5_v_range.pack(side=tk.LEFT, padx=2) ttk.Button(rng5, text="Set", width=3, command=lambda: self._set_range(5, "V", self._ch5_v_range)).pack(side=tk.LEFT) ttk.Label(rng5, text=" I range:").pack(side=tk.LEFT) self._ch5_i_range = ttk.Combobox(rng5, values=i_ranges, width=6, state="readonly") self._ch5_i_range.set("AUTO") self._ch5_i_range.pack(side=tk.LEFT, padx=2) ttk.Button(rng5, text="Set", width=3, command=lambda: self._set_range(5, "I", self._ch5_i_range)).pack(side=tk.LEFT) ttk.Button(rng5, text="Degauss", width=7, command=lambda: self._send(Cmd.DEGAUSS, [5])).pack(side=tk.LEFT, padx=(8, 0)) self._ch5_range_label = ttk.Label(input_frame, text="V: --- / I: ---", font=("Consolas", 9)) self._ch5_range_label.pack(anchor=tk.W) # Output side output_frame = ttk.LabelFrame(frame, text="Output (Ch6 - MPPT)", padding=4) output_frame.pack(fill=tk.X, pady=2) self._m_u6 = ttk.Label(output_frame, text="U6: ---", font=("Consolas", 11)) self._m_u6.pack(anchor=tk.W) self._m_i6 = ttk.Label(output_frame, text="I6: ---", font=("Consolas", 11)) self._m_i6.pack(anchor=tk.W) self._m_p6 = ttk.Label(output_frame, text="P6: ---", font=("Consolas", 11)) self._m_p6.pack(anchor=tk.W) # Ch6 ranges rng6 = ttk.Frame(output_frame) rng6.pack(fill=tk.X, pady=(4, 0)) ttk.Label(rng6, text="V range:").pack(side=tk.LEFT) self._ch6_v_range = ttk.Combobox(rng6, values=v_ranges, width=6, state="readonly") self._ch6_v_range.set("AUTO") self._ch6_v_range.pack(side=tk.LEFT, padx=2) ttk.Button(rng6, text="Set", width=3, command=lambda: self._set_range(6, "V", self._ch6_v_range)).pack(side=tk.LEFT) ttk.Label(rng6, text=" I range:").pack(side=tk.LEFT) self._ch6_i_range = ttk.Combobox(rng6, values=i_ranges, width=6, state="readonly") self._ch6_i_range.set("AUTO") self._ch6_i_range.pack(side=tk.LEFT, padx=2) ttk.Button(rng6, text="Set", width=3, command=lambda: self._set_range(6, "I", self._ch6_i_range)).pack(side=tk.LEFT) ttk.Button(rng6, text="Degauss", width=7, command=lambda: self._send(Cmd.DEGAUSS, [6])).pack(side=tk.LEFT, padx=(8, 0)) self._ch6_range_label = ttk.Label(output_frame, text="V: --- / I: ---", font=("Consolas", 9)) self._ch6_range_label.pack(anchor=tk.W) # Efficiency - big and bold eff_frame = ttk.Frame(frame) eff_frame.pack(fill=tk.X, pady=4) self._m_eff = ttk.Label(eff_frame, text="EFF1: --- %", font=("Consolas", 16, "bold")) self._m_eff.pack(anchor=tk.CENTER) def _build_logging_controls(self, parent) -> None: frame = ttk.LabelFrame(parent, text="Data Logging", padding=8) frame.pack(fill=tk.X, padx=4, pady=4) row = ttk.Frame(frame) row.pack(fill=tk.X, pady=2) ttk.Label(row, text="Interval (s):", width=12).pack(side=tk.LEFT) self._poll_interval = ttk.Entry(row, width=6) self._poll_interval.insert(0, "1.0") self._poll_interval.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Set", width=4, command=self._set_interval).pack(side=tk.LEFT, padx=2) row = ttk.Frame(frame) row.pack(fill=tk.X, pady=2) self._btn_log_start = ttk.Button(row, text="Start Log", command=self._start_log) self._btn_log_start.pack(side=tk.LEFT, padx=2) self._btn_log_stop = ttk.Button(row, text="Stop Log", command=self._stop_log, state=tk.DISABLED) self._btn_log_stop.pack(side=tk.LEFT, padx=2) ttk.Button(row, text="Plot Eff...", command=self._plot_eff).pack(side=tk.LEFT, padx=2) self._log_status = ttk.Label(frame, text="Not logging", font=("Consolas", 9)) self._log_status.pack(anchor=tk.W, pady=2) def _build_profile_controls(self, parent) -> None: frame = ttk.LabelFrame(parent, text="Shade Profile", padding=8) frame.pack(fill=tk.X, padx=4, pady=4) row = ttk.Frame(frame) row.pack(fill=tk.X, pady=2) self._profile_path_label = ttk.Label(row, text="No file selected", font=("Consolas", 9)) self._profile_path_label.pack(side=tk.LEFT, fill=tk.X, expand=True) ttk.Button(row, text="Browse", command=self._browse_profile).pack(side=tk.RIGHT, padx=2) row = ttk.Frame(frame) row.pack(fill=tk.X, pady=2) ttk.Label(row, text="Settle (s):", width=12).pack(side=tk.LEFT) self._profile_settle = ttk.Entry(row, width=6) self._profile_settle.insert(0, "5.0") self._profile_settle.pack(side=tk.LEFT, padx=2) row = ttk.Frame(frame) row.pack(fill=tk.X, pady=2) self._btn_profile_run = ttk.Button(row, text="Run Profile", command=self._run_profile) self._btn_profile_run.pack(side=tk.LEFT, padx=2) self._btn_profile_stop = ttk.Button(row, text="Stop", command=self._stop_profile, state=tk.DISABLED) self._btn_profile_stop.pack(side=tk.LEFT, padx=2) self._profile_status = ttk.Label(frame, text="Idle", font=("Consolas", 9)) self._profile_status.pack(anchor=tk.W, pady=2) self._profile_steps: list[dict] = [] self._profile_index = 0 self._profile_running = False self._profile_t0 = 0.0 self._profile_after_id = None def _build_sweep_vi_controls(self, parent) -> None: frame = ttk.LabelFrame(parent, text="2D Sweep (V × Load)", padding=8) frame.pack(fill=tk.X, padx=4, pady=4) # Voltage range row = ttk.Frame(frame) row.pack(fill=tk.X, pady=1) ttk.Label(row, text="V start:", width=10).pack(side=tk.LEFT) self._svi_v_start = ttk.Entry(row, width=7) self._svi_v_start.insert(0, "35") self._svi_v_start.pack(side=tk.LEFT, padx=2) ttk.Label(row, text="stop:").pack(side=tk.LEFT) self._svi_v_stop = ttk.Entry(row, width=7) self._svi_v_stop.insert(0, "100") self._svi_v_stop.pack(side=tk.LEFT, padx=2) ttk.Label(row, text="step:").pack(side=tk.LEFT) self._svi_v_step = ttk.Entry(row, width=5) self._svi_v_step.insert(0, "5") self._svi_v_step.pack(side=tk.LEFT, padx=2) # Load mode selector row = ttk.Frame(frame) row.pack(fill=tk.X, pady=1) ttk.Label(row, text="Load mode:", width=10).pack(side=tk.LEFT) self._svi_load_mode = ttk.Combobox(row, values=["CC", "CP"], width=4, state="readonly") self._svi_load_mode.set("CC") self._svi_load_mode.pack(side=tk.LEFT, padx=2) self._svi_load_mode.bind("<>", self._on_svi_mode_change) # Load setpoint range row = ttk.Frame(frame) row.pack(fill=tk.X, pady=1) self._svi_l_label = ttk.Label(row, text="I start:", width=10) self._svi_l_label.pack(side=tk.LEFT) self._svi_l_start = ttk.Entry(row, width=7) self._svi_l_start.insert(0, "0.5") self._svi_l_start.pack(side=tk.LEFT, padx=2) ttk.Label(row, text="stop:").pack(side=tk.LEFT) self._svi_l_stop = ttk.Entry(row, width=7) self._svi_l_stop.insert(0, "30") self._svi_l_stop.pack(side=tk.LEFT, padx=2) ttk.Label(row, text="step:").pack(side=tk.LEFT) self._svi_l_step = ttk.Entry(row, width=5) self._svi_l_step.insert(0, "1") self._svi_l_step.pack(side=tk.LEFT, padx=2) # Supply current limit + settle row = ttk.Frame(frame) row.pack(fill=tk.X, pady=1) ttk.Label(row, text="I limit:", width=10).pack(side=tk.LEFT) self._svi_ilimit = ttk.Entry(row, width=7) self._svi_ilimit.insert(0, f"{PSU_MAX_CURRENT_A:g}") self._svi_ilimit.pack(side=tk.LEFT, padx=2) ttk.Label(row, text="settle:").pack(side=tk.LEFT) self._svi_settle = ttk.Entry(row, width=5) self._svi_settle.insert(0, "5.0") self._svi_settle.pack(side=tk.LEFT, padx=2) ttk.Label(row, text="s").pack(side=tk.LEFT) # Run / Stop row = ttk.Frame(frame) row.pack(fill=tk.X, pady=2) self._btn_svi_run = ttk.Button(row, text="Run Sweep", command=self._run_sweep_vi) self._btn_svi_run.pack(side=tk.LEFT, padx=2) self._btn_svi_stop = ttk.Button(row, text="Stop", command=self._stop_sweep_vi, state=tk.DISABLED) self._btn_svi_stop.pack(side=tk.LEFT, padx=2) self._svi_status = ttk.Label(frame, text="Idle", font=("Consolas", 9)) self._svi_status.pack(anchor=tk.W, pady=2) self._svi_estimate = ttk.Label(frame, text="", font=("Consolas", 9)) self._svi_estimate.pack(anchor=tk.W) # Bind entries to recalculate time estimate on change for entry in (self._svi_v_start, self._svi_v_stop, self._svi_v_step, self._svi_l_start, self._svi_l_stop, self._svi_l_step, self._svi_settle): entry.bind("", lambda _e: self._update_svi_estimate()) self._update_svi_estimate() self._svi_thread = None self._svi_stop_event = None def _build_graphs(self, parent) -> None: # Selector for the phase-offset graph signal sel_row = ttk.Frame(parent) sel_row.pack(fill=tk.X) ttk.Label(sel_row, text="Phase graph:").pack(side=tk.LEFT, padx=(4, 2)) self._phase_sel = ttk.Combobox( sel_row, values=["vfly_correction", "vfly_ofs_applied"], width=16, state="readonly") self._phase_sel.set("vfly_correction") self._phase_sel.pack(side=tk.LEFT) self._phase_sel.bind("<>", self._on_phase_sel_change) self._fig = Figure(figsize=(10, 11), dpi=100) self._fig.suptitle("MPPT Testbench Live", fontsize=12, fontweight="bold") self._ax_power = self._fig.add_subplot(6, 1, 1) self._ax_eff = self._fig.add_subplot(6, 1, 2) self._ax_volt = self._fig.add_subplot(6, 1, 3) self._ax_curr = self._fig.add_subplot(6, 1, 4) self._ax_vfly = self._fig.add_subplot(6, 1, 5) self._ax_phase = self._fig.add_subplot(6, 1, 6) # Power self._ax_power.set_ylabel("Power (W)") self._ax_power.set_title("Power", fontsize=10) self._ax_power.grid(True, alpha=0.3) self._ln_p5, = self._ax_power.plot([], [], label="P_in (meter)", linewidth=1.5) self._ln_p6, = self._ax_power.plot([], [], label="P_out (meter)", linewidth=1.5) self._ln_sp, = self._ax_power.plot([], [], label="Supply P", linewidth=1, ls="--", alpha=0.6) self._ln_lp, = self._ax_power.plot([], [], label="Load P", linewidth=1, ls="--", alpha=0.6) self._ax_power.legend(loc="upper left", fontsize=8) # Efficiency self._ax_eff.set_ylabel("Efficiency (%)") self._ax_eff.set_title("Efficiency", fontsize=10) self._ax_eff.grid(True, alpha=0.3) self._ln_eff, = self._ax_eff.plot([], [], label="EFF1", linewidth=1.5, color="green") self._ax_eff.legend(loc="upper left", fontsize=8) # Voltage self._ax_volt.set_ylabel("Voltage (V)") self._ax_volt.set_title("Voltage", fontsize=10) self._ax_volt.grid(True, alpha=0.3) self._ln_u5, = self._ax_volt.plot([], [], label="U5 (input)", linewidth=1.5) self._ln_u6, = self._ax_volt.plot([], [], label="U6 (output)", linewidth=1.5) self._ax_volt.legend(loc="upper left", fontsize=8) # Current self._ax_curr.set_ylabel("Current (A)") self._ax_curr.set_title("Current", fontsize=10) self._ax_curr.grid(True, alpha=0.3) self._ln_i5, = self._ax_curr.plot([], [], label="I5 (input)", linewidth=1.5) self._ln_i6, = self._ax_curr.plot([], [], label="I6 (output)", linewidth=1.5) self._ax_curr.legend(loc="upper left", fontsize=8) # Vfly (STM32 telemetry, 100 Hz) self._ax_vfly.set_ylabel("Vfly (V)") self._ax_vfly.set_title("Flying Capacitor", fontsize=10) self._ax_vfly.grid(True, alpha=0.3) self._ln_vfly, = self._ax_vfly.plot( [], [], label="Vfly", linewidth=1.0, color="tab:purple") self._ax_vfly.legend(loc="upper left", fontsize=8) # Vfly correction / phase offset (selectable, STM32 telemetry) self._ax_phase.set_ylabel("ticks") self._ax_phase.set_xlabel("Time (s)") self._ax_phase.set_title("Vfly correction / phase offset", fontsize=10) self._ax_phase.grid(True, alpha=0.3) self._ln_phase, = self._ax_phase.plot( [], [], label="vfly_correction", linewidth=1.0, color="tab:orange") self._ax_phase.legend(loc="upper left", fontsize=8) self._all_axes = [self._ax_power, self._ax_eff, self._ax_volt, self._ax_curr, self._ax_vfly, self._ax_phase] self._fig.tight_layout() self._canvas = FigureCanvasTkAgg(self._fig, master=parent) self._canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) toolbar = NavigationToolbar2Tk(self._canvas, parent) toolbar.update() def _build_console(self, parent) -> None: frame = ttk.LabelFrame(parent, text="Console", padding=4) frame.pack(fill=tk.BOTH, expand=True, padx=0, pady=0) self._console_text = tk.Text( frame, height=8, font=("Consolas", 9), wrap=tk.WORD, state=tk.DISABLED, bg="#1e1e1e", fg="#cccccc", insertbackground="#cccccc", ) scrollbar = ttk.Scrollbar(frame, orient=tk.VERTICAL, command=self._console_text.yview) self._console_text.config(yscrollcommand=scrollbar.set) scrollbar.pack(side=tk.RIGHT, fill=tk.Y) self._console_text.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) # Tag for error messages self._console_text.tag_configure("error", foreground="#ff6b6b") self._console_text.tag_configure("success", foreground="#69db7c") self._console_text.tag_configure("warn", foreground="#ffd43b") # Redirect stdout to console self._orig_stdout = sys.stdout sys.stdout = _ConsoleRedirector(self) def _console(self, msg: str, tag: str = "") -> None: """Append a message to the console log. Thread-safe.""" ts = time.strftime("%H:%M:%S") line = f"[{ts}] {msg}\n" def _append(): self._console_text.config(state=tk.NORMAL) if tag: self._console_text.insert(tk.END, line, tag) else: self._console_text.insert(tk.END, line) self._console_text.see(tk.END) self._console_text.config(state=tk.DISABLED) # If called from a non-main thread, schedule on main thread try: if threading.current_thread() is threading.main_thread(): _append() else: self.after(0, _append) except RuntimeError: pass def _build_status_bar(self) -> None: bar = ttk.Frame(self) bar.pack(fill=tk.X, padx=4, pady=(0, 4)) self._status_label = ttk.Label(bar, text="Disconnected", font=("Consolas", 9)) self._status_label.pack(side=tk.LEFT) # ── STM32 Telemetry ─────────────────────────────────────────────── def _build_stm32_panel(self, parent) -> None: frame = ttk.LabelFrame(parent, text="STM32 Telemetry", padding=4) frame.pack(fill=tk.BOTH, expand=True, padx=2) F = ("Consolas", 9) def group(title: str) -> ttk.LabelFrame: g = ttk.LabelFrame(frame, text=title, padding=(4, 2)) g.pack(fill=tk.X, pady=2) g.columnconfigure(1, weight=1) g.columnconfigure(3, weight=1) return g def fld(g, row: int, col: int, caption: str, key: str) -> None: ttk.Label(g, text=caption, font=F).grid(row=row, column=col * 2, sticky="w") lbl = ttk.Label(g, text="---", font=F) lbl.grid(row=row, column=col * 2 + 1, sticky="e", padx=(2, 8)) self._stm_labels[key] = lbl g = group("Link") lbl = ttk.Label(g, text="NOT CONNECTED", font=("Consolas", 10, "bold"), foreground="#cc0000") lbl.grid(row=0, column=0, columnspan=4, sticky="w") self._stm_labels["link_state"] = lbl fld(g, 1, 0, "rate", "rate") fld(g, 1, 1, "loss", "loss") fld(g, 2, 0, "count", "counter") fld(g, 2, 1, "pong", "pong") btns = ttk.Frame(g) btns.grid(row=3, column=0, columnspan=4, sticky="w", pady=(2, 0)) ttk.Button(btns, text="Ping", width=6, command=self._stm32_ping).pack(side=tk.LEFT, padx=(0, 4)) ttk.Button(btns, text="Clear Flags", width=11, command=self._stm32_clear_flags).pack(side=tk.LEFT) g = group("Power") fld(g, 0, 0, "Vin", "vin") fld(g, 0, 1, "Vout", "vout") fld(g, 1, 0, "Iin", "iin") fld(g, 1, 1, "Iavg", "iin_avg") fld(g, 2, 0, "Iout", "iout") fld(g, 2, 1, "Islw", "iout_slow") fld(g, 3, 0, "Isys", "sys_i") fld(g, 4, 0, "Pin", "p_in") fld(g, 4, 1, "Pout", "p_out") fld(g, 5, 0, "Psys", "p_sys") eff = ttk.Label(g, text="EFF net: --- %", font=("Consolas", 13, "bold")) eff.grid(row=6, column=0, columnspan=4, pady=(2, 0)) self._stm_labels["eff_net"] = eff g = group("Temperature") fld(g, 0, 0, "Hsink", "etemp") fld(g, 0, 1, "Board", "btemp") g = group("Vfly") fld(g, 0, 0, "Vfly", "vfly") fld(g, 0, 1, "mode", "vfly_active") fld(g, 1, 0, "corr", "corr") fld(g, 1, 1, "ofs", "ofs") fld(g, 2, 0, "integ", "integ") fld(g, 2, 1, "avg", "avg_dbg") g = group("Control") fld(g, 0, 0, "mode", "mode") fld(g, 0, 1, "VREF", "vref") fld(g, 1, 0, "cc_out", "cc_out") fld(g, 1, 1, "iref", "iref") fld(g, 2, 0, "m.vin", "m_vin") fld(g, 2, 1, "m.iin", "m_iin") g = group("HRTIM") fld(g, 0, 0, "cmp_o", "cmp_o") fld(g, 0, 1, "cmp_i", "cmp_i") fld(g, 1, 0, "tmp", "tmp") fld(g, 1, 1, "diff", "cmp_d") g = group("Status Flags") fld(g, 0, 0, "hex", "flags_hex") # All 30 flags as checkbox-style indicators, 2 columns x 15 rows self._stm_flag_labels: dict[int, ttk.Label] = {} for bit in FLAG_SHORT: col, row = divmod(bit, 15) lbl = ttk.Label(g, text="☐ " + FLAG_SHORT[bit], font=("Consolas", 8), foreground="#777777") lbl.grid(row=1 + row, column=col * 2, columnspan=2, sticky="w") self._stm_flag_labels[bit] = lbl g = group("Fault Registers") fld(g, 0, 0, "FMAC", "fmac") fld(g, 0, 1, "PC", "pc") fld(g, 1, 0, "CFSR", "cfsr") def _connect_stm32(self) -> None: if self.stm32: return port = self._stm32_port.get().strip() self.stm32 = STM32Worker(port) self.stm32.start() self._last_flags = 0 self._flags_shown = -1 self._last_pong = None self._rate_buckets.clear() self._stm_ema.clear() self._stm_ema_t = None self._btn_stm32_connect.config(state=tk.DISABLED) self._btn_stm32_disconnect.config(state=tk.NORMAL) self._stm32_port.config(state=tk.DISABLED) self._console(f"STM32 link started on {port} (460800 8-O-1)", "success") # If a data log is already running, start the telemetry CSV alongside if self._log_file and self._telem_log_path: self.stm32.start_csv(self._telem_log_path) self._console(f"Telemetry CSV: {self._telem_log_path}") self._poll_stm32() def _disconnect_stm32(self) -> None: if not self.stm32: return w = self.stm32 self.stm32 = None # ends the _poll_stm32 loop w.stop() self._btn_stm32_connect.config(state=tk.NORMAL) self._btn_stm32_disconnect.config(state=tk.DISABLED) self._stm32_port.config(state=tk.NORMAL) self._stm_labels["link_state"].config(text="NOT CONNECTED", foreground="#cc0000") self._console("STM32 link stopped") def _stm32_ping(self) -> None: if self.stm32: self.stm32.send_frame(build_ping()) self._console("STM32 PING sent") def _stm32_clear_flags(self) -> None: if self.stm32: self.stm32.send_frame(build_clear_flags()) self._last_flags = 0 self._console("STM32 CLEAR FLAGS sent") def _poll_stm32(self) -> None: w = self.stm32 if not w: return b, wall = w.get_latest() live = b is not None and (time.time() - wall) < 2.0 state_lbl = self._stm_labels["link_state"] if live: state_lbl.config(text="RECEIVING", foreground="#00aa00") elif w.connected: state_lbl.config(text="PORT OPEN, NO DATA", foreground="#cc8800") else: state_lbl.config(text="NOT CONNECTED", foreground="#cc0000") if live: self._update_stm32_panel(b) # Latched status flags -> console the newly-set ones if b.status_flags != self._last_flags: newly = b.status_flags & ~self._last_flags self._last_flags = b.status_flags for bit, name in FLAG_NAMES.items(): if newly & (1 << bit): tag = "warn" if (1 << bit) & FLAG_INFO_MASK else "error" self._console(f"STM32 flag: {name}", tag) # Pong increments on the MCU when it processes CMD_PING if self._last_pong is None: self._last_pong = b.pong elif b.pong != self._last_pong: self._last_pong = b.pong self._console("STM32 PONG received", "success") # Publish rate + loss over a 3 s sliding window. The oldest bucket # only anchors the time span -- its counts accumulated BEFORE it, so # summing it too overestimates the rate by ~1 bucket. fresh, sent = w.get_rates() tm = time.monotonic() self._rate_buckets.append((tm, fresh, sent)) while self._rate_buckets and self._rate_buckets[0][0] < tm - 3.0: self._rate_buckets.popleft() span = tm - self._rate_buckets[0][0] if span > 0.5: buckets = list(self._rate_buckets)[1:] fsum = sum(x[1] for x in buckets) ssum = sum(x[2] for x in buckets) self._stm_labels["rate"].config(text=f"{fsum / span:5.1f} Hz") loss = (1.0 - fsum / ssum) * 100.0 if ssum > 0 else 0.0 self._stm_labels["loss"].config(text=f"{max(loss, 0.0):4.1f} %") # Telemetry graphs (decimated for display; full rate recorded) t, vfly, corr, ofs = w.get_graph_snapshot(TELEM_DECIMATE) ts = [x - self._t0_mono for x in t] self._ln_vfly.set_data(ts, [v / 1000.0 for v in vfly]) if self._phase_sel.get() == "vfly_ofs_applied": self._ln_phase.set_data(ts, ofs) else: self._ln_phase.set_data(ts, corr) self._redraw_axes() self.after(POLL_MS, self._poll_stm32) def _update_stm32_panel(self, b) -> None: """Refresh all telemetry labels from the latest broadcast sample.""" # Efficiency computed raw per sample, only the results EMA'd raw_p_in = b.vin * (-b.iin) / 1e6 raw_p_out = b.vout * b.iout_slow / 1e6 raw_p_sys = b.vout * b.sys_current_ma / 1e6 raw_eff = ((raw_p_out - raw_p_sys) / raw_p_in * 100.0) if raw_p_in > 0.1 else 0.0 e = self._stm_ema_update({ "vin": b.vin, "vout": b.vout, "iin": b.iin, "iin_avg": float(b.iin_avg_ma), "iout": b.iout, "iout_slow": b.iout_slow, "sys_i": float(b.sys_current_ma), "etemp": b.etemp, "btemp": b.btemp, "p_in": raw_p_in, "p_out": raw_p_out, "p_sys": raw_p_sys, "eff_net": raw_eff, }) L = self._stm_labels L["vin"].config(text=f"{e['vin'] / 1000:7.3f} V") L["vout"].config(text=f"{e['vout'] / 1000:7.3f} V") L["iin"].config(text=f"{e['iin'] / 1000:+7.3f} A") L["iin_avg"].config(text=f"{e['iin_avg']:+6.0f} mA") L["iout"].config(text=f"{e['iout'] / 1000:7.3f} A") L["iout_slow"].config(text=f"{e['iout_slow'] / 1000:7.3f} A") L["sys_i"].config(text=f"{e['sys_i']:5.0f} mA") L["p_in"].config(text=f"{e['p_in']:7.2f} W") L["p_out"].config(text=f"{e['p_out']:7.2f} W") L["p_sys"].config(text=f"{e['p_sys']:6.2f} W") # Efficiency is meaningless at (near) no load if e["p_out"] < 5.0: L["eff_net"].config(text="EFF net: --.-- %") else: L["eff_net"].config(text=f"EFF net: {e['eff_net']:.2f} %") L["etemp"].config(text=f"{e['etemp']:5.1f} °C") L["btemp"].config(text=f"{e['btemp']:5.1f} °C") L["counter"].config(text=str(b.counter)) L["pong"].config(text=str(b.pong)) L["vfly"].config(text=f"{b.vfly / 1000:7.3f} V") L["vfly_active"].config(text=str(b.vfly_active)) L["corr"].config(text=f"{b.vfly_correction:+d}") L["ofs"].config(text=f"{b.vfly_ofs_applied:+d}") L["integ"].config(text=f"{b.vfly_integral:.1f}") L["avg_dbg"].config(text=f"{b.vfly_avg_debug:.1f}") modes = ("OFF", "MPPT", "CV", "CC") L["mode"].config(text=modes[b.ctrl_mode] if 0 <= b.ctrl_mode < len(modes) else str(b.ctrl_mode)) L["vref"].config(text=str(b.VREF)) L["cc_out"].config(text=f"{b.cc_output_f:.1f}") L["iref"].config(text=f"{b.mppt_iref:.1f}") L["m_vin"].config(text=f"{b.mppt_last_vin:.0f}") L["m_iin"].config(text=f"{b.mppt_last_iin:.0f}") L["cmp_o"].config(text=str(b.cmp_outer)) L["cmp_i"].config(text=str(b.cmp_inner)) s = 2 * b.last_tmp - 7158 L["tmp"].config(text=f"{b.last_tmp} ({'D>0.5' if s > 0 else 'D<0.5'})") L["cmp_d"].config(text=f"{b.cmp_outer - b.cmp_inner:+d}") L["flags_hex"].config(text=f"0x{b.status_flags:08X}") if b.status_flags != self._flags_shown: self._flags_shown = b.status_flags for bit, lbl in self._stm_flag_labels.items(): m = 1 << bit if b.status_flags & m: color = "#c8a400" if m & FLAG_INFO_MASK else "#cc0000" lbl.config(text="☑ " + FLAG_SHORT[bit], foreground=color) else: lbl.config(text="☐ " + FLAG_SHORT[bit], foreground="#777777") for key, val in (("fmac", b.fmac_sr), ("pc", b.fault_pc), ("cfsr", b.cfsr)): L[key].config(text=f"0x{val:08X}", foreground="#888888" if val == 0 else "#cc0000") def _stm_ema_update(self, values: dict) -> dict: """Rate-independent EMA for displayed values (tau = STM_EMA_TAU_S).""" now = time.monotonic() if self._stm_ema_t is None: a = 1.0 else: a = 1.0 - math.exp(-(now - self._stm_ema_t) / STM_EMA_TAU_S) self._stm_ema_t = now for k, v in values.items(): prev = self._stm_ema.get(k) self._stm_ema[k] = v if prev is None else prev + a * (v - prev) return self._stm_ema def _on_phase_sel_change(self, _event=None) -> None: sel = self._phase_sel.get() self._ln_phase.set_label(sel) self._ax_phase.legend(loc="upper left", fontsize=8) self._last_draw = 0.0 # force an immediate redraw self._redraw_axes() def _redraw_axes(self) -> None: """Scroll all axes to the last GRAPH_WINDOW_S seconds; throttled redraw.""" xmax = time.monotonic() - self._t0_mono for ax in self._all_axes: ax.set_xlim(xmax - GRAPH_WINDOW_S, xmax) ax.relim() ax.autoscale_view(scalex=False, scaley=True) now = time.monotonic() if now - self._last_draw >= MIN_REDRAW_S: self._last_draw = now self._canvas.draw_idle() # ── Connection ──────────────────────────────────────────────────── def _connect(self) -> None: try: supply_addr = self._supply_addr.get().strip() if supply_addr == "auto": supply_addr = MPPTTestbench.find_supply(None) meter_addr = self._meter_addr.get().strip() if meter_addr == "auto": meter_addr = MPPTTestbench.find_meter(None) load_port = self._load_port.get().strip() load_baud = int(self._load_baud.get().strip()) supply = IT6500(supply_addr) load = Prodigit3366G(load_port, baudrate=load_baud) meter = Hioki3193(meter_addr) self.bench = MPPTTestbench(supply, load, meter) self.bench.supply.remote() self.bench.load.remote() self.worker = InstrumentWorker(self.bench, interval=1.0) self.worker.start() self._t0 = time.time() self._point_count = 0 # Fresh instrument history (shared timebase is NOT reset) self._timestamps.clear() for s in self._series.values(): s.clear() # Update UI state self._btn_connect.config(state=tk.DISABLED) self._btn_setup.config(state=tk.NORMAL) self._btn_disconnect.config(state=tk.NORMAL) self._supply_addr.config(state=tk.DISABLED) self._load_port.config(state=tk.DISABLED) self._load_baud.config(state=tk.DISABLED) self._meter_addr.config(state=tk.DISABLED) self._status_label.config(text="Connected") self._console(f"Connected: supply={supply_addr}, load={load_port}, meter={meter_addr}", "success") self._poll() # find_supply/find_meter sys.exit(1) when auto-detect fails (CLI # helpers) -- catch SystemExit too or the whole GUI dies except (Exception, SystemExit) as e: msg = str(e) if isinstance(e, SystemExit) or not msg: msg = "Instrument auto-detect failed (see console for details)." self._console(f"Connection failed: {msg}", "error") messagebox.showerror("Connection Error", msg) def _disconnect(self) -> None: self._stop_log() if self.worker: self.worker.stop() self.worker.join(timeout=5) self.worker = None if self.bench: try: self.bench.close() except Exception: pass self.bench = None self._btn_connect.config(state=tk.NORMAL) self._btn_setup.config(state=tk.DISABLED) self._btn_disconnect.config(state=tk.DISABLED) self._supply_addr.config(state=tk.NORMAL) self._load_port.config(state=tk.NORMAL) self._load_baud.config(state=tk.NORMAL) self._meter_addr.config(state=tk.NORMAL) self._status_label.config(text="Disconnected") self._console("Disconnected") def _setup_all(self) -> None: self._send(Cmd.SETUP_ALL) self._console("Setup All sent") def _safe_off(self) -> None: """Emergency stop -- bypasses command queue for minimum latency.""" self._console("SAFE OFF triggered", "error") if self.bench: try: self.bench.safe_off() except Exception: pass if self.worker: self.worker.send(Cmd.SAFE_OFF) def _on_close(self) -> None: self._stop_log() self._disconnect_stm32() if self.bench: try: self.bench.safe_off() except Exception: pass self._disconnect() sys.stdout = self._orig_stdout self.destroy() # ── Polling / Data Update ───────────────────────────────────────── def _poll(self) -> None: """Called periodically to pull data from the worker and update UI.""" data = None if self.worker: data = self.worker.get_data() else: # During 2D sweep the worker is stopped; drain sweep queue latest = None while True: try: latest = self._sweep_data_queue.get_nowait() except queue.Empty: break data = latest if data: error = data.get("_error") if error: self._console(f"Error: {error}", "error") self._status_label.config(text=f"Error: {error}") else: self._latest_data = data self._update_readouts(data) self._update_graphs(data) self._log_data(data) self._point_count += 1 elapsed = time.time() - self._t0 mins, secs = divmod(int(elapsed), 60) hrs, mins = divmod(mins, 60) self._status_label.config( text=f"Connected | {self._point_count} pts | {hrs:02d}:{mins:02d}:{secs:02d}" ) # Keep polling as long as connected (bench exists) if self.worker or self.bench: self.after(POLL_MS, self._poll) def _update_readouts(self, data: dict) -> None: """Update all numeric labels from measurement data.""" # Supply self._sup_v_label.config(text=f"V: {_fmt(data['supply_V'])} V") self._sup_i_label.config(text=f"I: {_fmt(data['supply_I'])} A") self._sup_p_label.config(text=f"P: {_fmt(data['supply_P'])} W") # Load self._load_v_label.config(text=f"V: {_fmt(data['load_V'])} V") self._load_i_label.config(text=f"I: {_fmt(data['load_I'])} A") self._load_p_label.config(text=f"P: {_fmt(data['load_P'])} W") # Meter fm = self._fmt_meter self._m_u5.config(text=f"U5: {fm(data['meter_U5'])} V") self._m_i5.config(text=f"I5: {fm(data['meter_I5'])} A") self._m_p5.config(text=f"P5: {fm(data['meter_P5'])} W") self._m_u6.config(text=f"U6: {fm(data['meter_U6'])} V") self._m_i6.config(text=f"I6: {fm(data['meter_I6'])} A") self._m_p6.config(text=f"P6: {fm(data['meter_P6'])} W") eff = data['meter_EFF1'] if abs(eff) >= ERROR_THRESHOLD: self._m_eff.config(text="EFF1: --- %") else: self._m_eff.config(text=f"EFF1: {eff:.2f} %") # ON/OFF indicators supply_on = data.get("supply_on") if supply_on is True: self._sup_state_label.config(text="OUTPUT: ON", fg="#00aa00") elif supply_on is False: self._sup_state_label.config(text="OUTPUT: OFF", fg="#cc0000") else: self._sup_state_label.config(text="OUTPUT: ---", fg="#888888") load_on = data.get("load_on") if load_on is True: self._load_state_label.config(text="LOAD: ON", fg="#00aa00") elif load_on is False: self._load_state_label.config(text="LOAD: OFF", fg="#cc0000") else: self._load_state_label.config(text="LOAD: ---", fg="#888888") # Meter channel ranges vr5 = data.get("v_range_5") ir5 = data.get("i_range_5") self._ch5_range_label.config( text=f"V: {vr5 or '---'} / I: {ir5 or '---'}" ) vr6 = data.get("v_range_6") ir6 = data.get("i_range_6") self._ch6_range_label.config( text=f"V: {vr6 or '---'} / I: {ir6 or '---'}" ) def _update_graphs(self, data: dict) -> None: """Append data to series and redraw graphs (last GRAPH_WINDOW_S s).""" now = time.monotonic() - self._t0_mono self._timestamps.append(now) for key in self._series: self._series[key].append(_clean(data.get(key, 0.0))) # Time-trim beyond the display window (deques append in lockstep) cutoff = now - GRAPH_WINDOW_S - 2.0 while self._timestamps and self._timestamps[0] < cutoff: self._timestamps.popleft() for key in self._series: self._series[key].popleft() t = list(self._timestamps) self._ln_p5.set_data(t, list(self._series["meter_P5"])) self._ln_p6.set_data(t, list(self._series["meter_P6"])) self._ln_sp.set_data(t, list(self._series["supply_P"])) self._ln_lp.set_data(t, list(self._series["load_P"])) self._ln_eff.set_data(t, list(self._series["meter_EFF1"])) self._ln_u5.set_data(t, list(self._series["meter_U5"])) self._ln_u6.set_data(t, list(self._series["meter_U6"])) self._ln_i5.set_data(t, list(self._series["meter_I5"])) self._ln_i6.set_data(t, list(self._series["meter_I6"])) self._redraw_axes() # ── Format Helpers ───────────────────────────────────────────────── def _on_meter_fmt_change(self, _event=None) -> None: self._meter_fmt_sci = self._meter_fmt_combo.get() == "Scientific" def _fmt_meter(self, val: float) -> str: """Format a meter value based on the current format selection.""" if self._meter_fmt_sci: return _fmt_eng(val) return _fmt(val, decimals=4) def _on_svi_mode_change(self, _event=None) -> None: mode = self._svi_load_mode.get() if mode == "CC": self._svi_l_label.config(text="I start:") else: self._svi_l_label.config(text="P start:") self._update_svi_estimate() def _update_svi_estimate(self) -> None: """Recalculate and display estimated sweep duration.""" try: v_start = float(self._svi_v_start.get()) v_stop = float(self._svi_v_stop.get()) v_step = abs(float(self._svi_v_step.get())) l_start = float(self._svi_l_start.get()) l_stop = float(self._svi_l_stop.get()) l_step = abs(float(self._svi_l_step.get())) settle = float(self._svi_settle.get()) if v_step <= 0 or l_step <= 0: raise ValueError v_count = int(abs(v_stop - v_start) / v_step) + 1 l_count = int(abs(l_stop - l_start) / l_step) + 1 total = v_count * l_count secs = total * settle mins, s = divmod(int(secs), 60) hrs, m = divmod(mins, 60) if hrs: time_str = f"{hrs}h {m:02d}m {s:02d}s" elif m: time_str = f"{m}m {s:02d}s" else: time_str = f"{s}s" self._svi_estimate.config( text=f"{total} points, ~{time_str} (settle only)" ) except (ValueError, ZeroDivisionError): self._svi_estimate.config(text="") def _set_range(self, channel: int, vi: str, combo: ttk.Combobox) -> None: """Set voltage or current range for a HIOKI channel.""" val = combo.get() if val == "AUTO": if vi == "V": self._send(Cmd.SET_VOLTAGE_AUTO, channel, True) else: self._send(Cmd.SET_CURRENT_AUTO, channel, True) else: if vi == "V": self._send(Cmd.SET_VOLTAGE_RANGE, channel, int(val)) else: self._send(Cmd.SET_CURRENT_RANGE, channel, float(val)) # ── Command Helpers ─────────────────────────────────────────────── def _send(self, cmd: Cmd, *args) -> None: """Send a command to the worker thread.""" if self.worker: self.worker.send(cmd, *args) if args: self._console(f"{cmd.name} {' '.join(str(a) for a in args)}") else: self._console(cmd.name) def _send_float(self, cmd: Cmd, entry: ttk.Entry) -> None: """Parse an entry as float and send to worker.""" try: val = float(entry.get()) self._send(cmd, val) except ValueError: entry.config(foreground="red") self.after(1000, lambda: entry.config(foreground="")) def _set_supply_voltage(self) -> None: self._send_float(Cmd.SET_VOLTAGE, self._sup_voltage) def _set_supply_current(self) -> None: self._clamp_supply_current_entry() self._send_float(Cmd.SET_CURRENT, self._sup_current) def _clamp_supply_current_entry(self) -> None: """Keep the programmed supply current within PSU capability.""" try: val = float(self._sup_current.get()) except ValueError: return if val > PSU_MAX_CURRENT_A: self._console( f"Supply I {val:g}A clamped to PSU max " f"{PSU_MAX_CURRENT_A:g}A", "warn") self._sup_current.delete(0, tk.END) self._sup_current.insert(0, f"{PSU_MAX_CURRENT_A:g}") def _apply_supply(self) -> None: self._clamp_supply_current_entry() try: v = float(self._sup_voltage.get()) i = float(self._sup_current.get()) self._send(Cmd.APPLY, v, i) except ValueError: pass def _set_ovp(self) -> None: self._send_float(Cmd.SET_OVP, self._sup_ovp) def _set_load_mode(self) -> None: self._send(Cmd.SET_MODE, self._load_mode.get()) def _set_load_value(self) -> None: try: val = float(self._load_value.get()) mode = self._load_mode.get() except ValueError: self._load_value.config(foreground="red") self.after(1000, lambda: self._load_value.config(foreground="")) return # PSU capability gate (CC/CP; live readings when available) if mode in ("CC", "CP"): vin = self._latest_data.get("supply_V", 0.0) vout = self._latest_data.get("load_V", 0.0) if vout < 5.0: vout = VOUT_NOM_V if vin > 5.0: iin_est = _est_input_current(mode, val, vin, vout) if iin_est > PSU_MAX_CURRENT_A: self._console( f"Load {mode}={val:g} rejected: est. input current " f"{iin_est:.1f}A > PSU max {PSU_MAX_CURRENT_A:g}A " f"at Vin={vin:.1f}V / Vout={vout:.1f}V", "error") self._load_value.config(foreground="red") self.after(1000, lambda: self._load_value.config(foreground="")) return self._send(Cmd.SET_MODE_VALUE, mode, val) def _on_mode_change(self, _event=None) -> None: """Update the value label units when load mode changes.""" units = {"CC": "A", "CR": "\u03a9", "CV": "V", "CP": "W"} mode = self._load_mode.get() self._load_unit_label.config(text=f"Value ({units.get(mode, '?')}):") def _set_interval(self) -> None: try: val = float(self._poll_interval.get()) if val < 0.2: val = 0.2 self._send(Cmd.SET_INTERVAL, val) except ValueError: pass # ── Shade Profile ────────────────────────────────────────────────── def _browse_profile(self) -> None: path = filedialog.askopenfilename( filetypes=[("CSV files", "*.csv"), ("All files", "*.*")], ) if not path: return try: self._profile_steps = MPPTTestbench.load_shade_profile(path) n = len(self._profile_steps) dur = self._profile_steps[-1]["time"] if self._profile_steps else 0 self._profile_path_label.config(text=path.split("/")[-1].split("\\")[-1]) self._profile_status.config(text=f"Loaded: {n} steps, {dur:.0f}s") except Exception as e: messagebox.showerror("Profile Error", str(e)) def _run_profile(self) -> None: if not self._profile_steps: messagebox.showwarning("No Profile", "Load a profile CSV first.") return if not self.worker: return try: settle = float(self._profile_settle.get()) except ValueError: settle = 5.0 n = len(self._profile_steps) dur = self._profile_steps[-1]["time"] if self._profile_steps else 0 self._console(f"Shade profile started: {n} steps, {dur:.0f}s", "success") self._profile_running = True self._profile_index = 0 self._profile_t0 = time.time() self._btn_profile_run.config(state=tk.DISABLED) self._btn_profile_stop.config(state=tk.NORMAL) # Turn outputs on with first step first = self._profile_steps[0] self._send(Cmd.SET_CURRENT, first["current_limit"]) self._send(Cmd.SET_VOLTAGE, first["voltage"]) self._send(Cmd.OUTPUT_ON) if "load_mode" in first: self._send(Cmd.SET_MODE, first["load_mode"]) if "load_value" in first: self._send(Cmd.SET_MODE_VALUE, first.get("load_mode", "CC"), first["load_value"]) self._send(Cmd.LOAD_ON) self._profile_schedule_next() def _profile_schedule_next(self) -> None: if not self._profile_running: return if self._profile_index >= len(self._profile_steps): self._finish_profile() return step = self._profile_steps[self._profile_index] target = self._profile_t0 + step["time"] delay_ms = max(0, int((target - time.time()) * 1000)) self._profile_after_id = self.after(delay_ms, self._apply_profile_step) def _apply_profile_step(self) -> None: if not self._profile_running: return step = self._profile_steps[self._profile_index] self._send(Cmd.SET_CURRENT, step["current_limit"]) self._send(Cmd.SET_VOLTAGE, step["voltage"]) if "load_mode" in step: self._send(Cmd.SET_MODE, step["load_mode"]) if "load_value" in step: self._send(Cmd.SET_MODE_VALUE, step.get("load_mode", "CC"), step["load_value"]) elapsed = time.time() - self._profile_t0 n = len(self._profile_steps) self._profile_status.config( text=f"Step {self._profile_index + 1}/{n} " f"t={elapsed:.0f}s V={step['voltage']:.1f}V " f"I_lim={step['current_limit']:.1f}A" ) self._profile_index += 1 self._profile_schedule_next() def _stop_profile(self) -> None: self._profile_running = False if self._profile_after_id: self.after_cancel(self._profile_after_id) self._profile_after_id = None self._btn_profile_run.config(state=tk.NORMAL) self._btn_profile_stop.config(state=tk.DISABLED) self._profile_status.config(text="Stopped") self._console("Shade profile stopped", "warn") def _finish_profile(self) -> None: self._profile_running = False self._btn_profile_run.config(state=tk.NORMAL) self._btn_profile_stop.config(state=tk.DISABLED) elapsed = time.time() - self._profile_t0 self._profile_status.config(text=f"Done ({elapsed:.0f}s)") self._console(f"Shade profile complete ({elapsed:.0f}s)", "success") # ── 2D Sweep (V × I) ─────────────────────────────────────────────── def _run_sweep_vi(self) -> None: if not self.bench: return try: params = { "v_start": float(self._svi_v_start.get()), "v_stop": float(self._svi_v_stop.get()), "v_step": float(self._svi_v_step.get()), "l_start": float(self._svi_l_start.get()), "l_stop": float(self._svi_l_stop.get()), "l_step": float(self._svi_l_step.get()), "current_limit": float(self._svi_ilimit.get()), "settle_time": float(self._svi_settle.get()), "load_mode": self._svi_load_mode.get(), } except ValueError: messagebox.showerror("Invalid Input", "Check sweep parameters.") return if params["current_limit"] > PSU_MAX_CURRENT_A: self._console( f"I limit {params['current_limit']:g}A clamped to PSU max " f"{PSU_MAX_CURRENT_A:g}A", "warn") params["current_limit"] = PSU_MAX_CURRENT_A self._svi_ilimit.delete(0, tk.END) self._svi_ilimit.insert(0, f"{PSU_MAX_CURRENT_A:g}") # Ask for output file default_name = time.strftime("sweep_vi_%Y%m%d_%H%M%S.csv") path = filedialog.asksaveasfilename( defaultextension=".csv", filetypes=[("CSV files", "*.csv")], initialfile=default_name, ) if not path: return mode = params["load_mode"] unit = "A" if mode == "CC" else "W" self._console( f"2D sweep: V={params['v_start']}-{params['v_stop']}V " f"{mode}={params['l_start']}-{params['l_stop']}{unit}", "success", ) # Stop the normal worker and wait for it to finish so the # sweep thread has exclusive access to the instrument bus if self.worker: self.worker.stop() self.worker.join(timeout=10) self.worker = None # Restart _poll so it drains the sweep data queue self.after(POLL_MS, self._poll) self._btn_svi_run.config(state=tk.DISABLED) self._btn_svi_stop.config(state=tk.NORMAL) stop_event = threading.Event() self._svi_stop_event = stop_event def _sweep_thread(): try: self.after(0, lambda: self._svi_status.config(text="Running...")) results = self._sweep_vi_loop(params, stop_event) # Write CSV if results: self._write_sweep_vi_csv(results, path) fname = path.split('/')[-1].split(chr(92))[-1] msg = f"Done: {len(results)} pts saved to {fname}" self._console(msg, "success") else: msg = "Sweep stopped (no data)" self._console(msg, "warn") self.after(0, lambda: self._svi_status.config(text=msg)) except Exception as e: err_msg = f"Error: {e}" self._console(f"Sweep error: {e}", "error") self.after(0, lambda: self._svi_status.config(text=err_msg)) finally: self.after(0, self._sweep_vi_done) self._svi_thread = threading.Thread(target=_sweep_thread, daemon=True) self._svi_thread.start() def _sweep_temps(self) -> tuple[float, float] | None: """(etemp, btemp) in C from the STM32 link, or None if unlinked/stale.""" w = self.stm32 if not w: return None b, wall = w.get_latest() if b is None or time.time() - wall > TELEM_STALE_S: return None return b.etemp, b.btemp def _thermal_hold(self, bench, load_mode: str, vout_est: float, stop: threading.Event) -> float | None: """Pause the sweep while etemp/btemp are near the firmware trips. Drops the load to ~1 A output while cooling and blocks until both temps are back below pause - hysteresis (or the sweep is stopped). Returns the hold setpoint it applied, or None if no pause happened. Runs on the sweep thread. """ t = self._sweep_temps() if t is None or (t[0] < ETEMP_PAUSE_C and t[1] < BTEMP_PAUSE_C): return None hold = (THERMAL_HOLD_LOAD_A if load_mode == "CC" else THERMAL_HOLD_LOAD_A * vout_est) # CP: ~1 A worth of W unit = "A" if load_mode == "CC" else "W" bench._apply_load_value(load_mode, hold) self._console( f"Thermal pause: etemp {t[0]:.1f}C / btemp {t[1]:.1f}C near " f"firmware limits ({ETEMP_PAUSE_C:g}/{BTEMP_PAUSE_C:g}C) - " f"load held at {hold:g}{unit}", "warn") while not stop.is_set(): time.sleep(2.0) t = self._sweep_temps() if t is None: continue # link lost mid-pause: keep holding, stay safe self.after(0, lambda t=t: self._svi_status.config( text=f"THERMAL PAUSE hsink {t[0]:.1f}C board {t[1]:.1f}C " f"(resume < {ETEMP_PAUSE_C - TEMP_RESUME_HYST_C:g}/" f"{BTEMP_PAUSE_C - TEMP_RESUME_HYST_C:g}C)")) if (t[0] < ETEMP_PAUSE_C - TEMP_RESUME_HYST_C and t[1] < BTEMP_PAUSE_C - TEMP_RESUME_HYST_C): self._console( f"Cooled to etemp {t[0]:.1f}C / btemp {t[1]:.1f}C - " f"resuming sweep", "success") break return hold def _sweep_vi_loop(self, p: dict, stop: threading.Event) -> list: """Run the 2D sweep on a background thread. Returns list of SweepPoint.""" from testbench.bench import IDLE_VOLTAGE bench = self.bench v_start, v_stop, v_step = p["v_start"], p["v_stop"], p["v_step"] l_start, l_stop, l_step = p["l_start"], p["l_stop"], p["l_step"] current_limit = p["current_limit"] settle = p["settle_time"] load_mode = p.get("load_mode", "CC") unit = "A" if load_mode == "CC" else "W" # Auto-correct directions if v_start < v_stop and v_step < 0: v_step = -v_step elif v_start > v_stop and v_step > 0: v_step = -v_step if l_start < l_stop and l_step < 0: l_step = -l_step elif l_start > l_stop and l_step > 0: l_step = -l_step # Clear any stale state time.sleep(0.5) try: bench.supply.get_error() except Exception: pass # Sanity check max_v = max(abs(v_start), abs(v_stop)) min_v = min(abs(v_start), abs(v_stop)) max_l = max(abs(l_start), abs(l_stop)) bench.check_supply_capability( max_v, current_limit, min_voltage=min_v, load_mode=load_mode, max_load_setpoint=max_l, ) if self._sweep_temps() is None: self._console( "STM32 not linked - thermal pause guard inactive for this " "sweep", "warn") # PSU capability gate: never command an operating point whose # estimated input draw exceeds what the supply can source. psu_imax = min(current_limit, PSU_MAX_CURRENT_A) vout_est = VOUT_NOM_V # refined from measured load voltage as we go first_v = v_start if v_step > 0 else max(v_start, v_stop) if _est_input_current(load_mode, l_start, first_v, vout_est) > psu_imax: raise ValueError( f"first step {load_mode}={l_start:g}{unit} at {first_v:g}V " f"already needs > {psu_imax:g}A input") # Load range handling (load still OFF here). A mid-sweep auto-range # transition momentarily unloads the converter, so the range is # pinned ONCE for the whole run: CC is forced to Range II. Then the # selected range's reachable maximum is verified empirically by # programming the sweep max and reading it back -- a clamped readback # means the range tops out below the requested sweep maximum, and # steps above it are rejected instead of silently clamped. bench.load.set_mode(load_mode) if load_mode == "CC": try: bench.load.set_cc_range("R2") self._console("Load CC range pinned to R2 for the sweep " "(no auto-ranging mid-run)") except Exception as e: self._console(f"Could not pin load CC range ({e})", "warn") range_max = None try: bench._apply_load_value(load_mode, max_l) rb = (bench.load.get_cc_current() if load_mode == "CC" else bench.load.get_cp_power()) if rb < max_l - max(0.01 * max_l, 0.05): range_max = rb self._console( f"Load range tops out at {rb:g}{unit}: sweep steps " f"above will be rejected (asked up to {max_l:g}{unit})", "warn") if min(abs(l_start), abs(l_stop)) > range_max: raise ValueError( f"no sweep step fits the selected load range " f"(max {rb:g}{unit})") except ValueError: raise except Exception as e: self._console(f"Load range readback failed ({e}) - range check " f"skipped", "warn") bench.supply.set_current(current_limit) bench.supply.output_on() bench._apply_load_value(load_mode, l_start) bench.load.load_on() results = [] n = 0 v = v_start applied = l_start # last load setpoint actually commanded try: while not stop.is_set(): if v_step > 0 and v > v_stop + v_step / 2: break if v_step < 0 and v < v_stop + v_step / 2: break bench.supply.set_voltage(v) ll = l_start rejected = 0 while not stop.is_set(): if l_step > 0 and ll > l_stop + l_step / 2: break if l_step < 0 and ll < l_stop + l_step / 2: break # Reject steps beyond the pinned load range or that the # PSU cannot feed at this Vin/Vout ratio iin_est = _est_input_current(load_mode, ll, v, vout_est) if ((range_max is not None and ll > range_max * 1.001) or iin_est > psu_imax): rejected += 1 ll += l_step continue # Pause near the firmware thermal trips (holds at ~1 A) held = self._thermal_hold(bench, load_mode, vout_est, stop) if held is not None: applied = held if stop.is_set(): break bench._apply_load_value(load_mode, ll) applied = ll time.sleep(settle) if stop.is_set(): break point = bench._record_point(v, current_limit, load_setpoint=ll) results.append(point) n += 1 if point.load_voltage > 5.0: vout_est = point.load_voltage # Measured backstop: the estimate can be off (eff, Vout) if point.supply_current > psu_imax * 1.05: self._console( f"V={v:.1f}V {load_mode}={ll:.1f}{unit}: supply " f"current {point.supply_current:.1f}A over PSU " f"limit {psu_imax:g}A - backing off", "warn") rejected += 1 results.pop() # point measured in current limiting n -= 1 if l_step > 0: # up-sweep: heavier steps follow, stop here back = max(l_start, ll - l_step) bench._apply_load_value(load_mode, back) applied = back break # down-sweep: later steps are lighter, keep going ll += l_step continue # Push data for live graph/readout updates gui_data = { "supply_V": point.supply_voltage, "supply_I": point.supply_current, "supply_P": point.supply_power, "load_V": point.load_voltage, "load_I": point.load_current, "load_P": point.load_power, "meter_U5": point.supply_voltage, "meter_I5": point.supply_current, "meter_P5": point.input_power, "meter_U6": point.load_voltage, "meter_I6": point.load_current, "meter_P6": point.output_power, "meter_EFF1": point.efficiency, "supply_on": True, "load_on": True, "_error": None, "_timestamp": time.time(), } try: self._sweep_data_queue.put_nowait(gui_data) except queue.Full: try: self._sweep_data_queue.get_nowait() except queue.Empty: pass self._sweep_data_queue.put_nowait(gui_data) self.after( 0, lambda v=v, ll=ll, pt=point, n=n: self._svi_status.config( text=f"[{n}] V={v:.1f}V {load_mode}={ll:.1f}{unit} " f"EFF={pt.efficiency:.1f}%" ), ) ll += l_step if rejected: self._console( f"V={v:.1f}V: rejected {rejected} step(s) - PSU " f"input limit {psu_imax:g}A / load range", "warn") v += v_step finally: # Ramp load down gradually to avoid sudden transients cur_load = applied if n > 0 else l_start ramp_steps = max(int(abs(cur_load - l_start) / abs(l_step)), 1) if l_step != 0 else 1 ramp_steps = min(ramp_steps, 10) # cap at 10 steps if abs(cur_load) > abs(l_start) and ramp_steps > 1: decrement = (cur_load - l_start) / ramp_steps print(f"Ramping load down from {cur_load:.1f} to " f"{l_start:.1f}{unit} in {ramp_steps} steps...") for i in range(1, ramp_steps + 1): step_val = cur_load - decrement * i bench._apply_load_value(load_mode, step_val) time.sleep(0.5) else: bench._apply_load_value(load_mode, l_start) time.sleep(0.5) bench.load.load_off() if load_mode == "CC": try: bench.load.set_cc_range("AUTO") # load is OFF: no glitch except Exception: pass bench.supply.set_voltage(IDLE_VOLTAGE) return results @staticmethod def _write_sweep_vi_csv(results, path: str) -> None: import csv as _csv with open(path, "w", newline="") as f: w = _csv.writer(f) w.writerow([ "voltage_set", "current_limit", "load_setpoint", "supply_V", "supply_I", "supply_P", "load_V", "load_I", "load_P", "input_power", "output_power", "efficiency", ]) for pt in results: w.writerow([ f"{pt.voltage_set:.4f}", f"{pt.current_limit:.4f}", f"{pt.load_setpoint:.4f}", f"{pt.supply_voltage:.4f}", f"{pt.supply_current:.4f}", f"{pt.supply_power:.4f}", f"{pt.load_voltage:.4f}", f"{pt.load_current:.4f}", f"{pt.load_power:.4f}", f"{pt.input_power:.4f}", f"{pt.output_power:.4f}", f"{pt.efficiency:.4f}", ]) def _stop_sweep_vi(self) -> None: if self._svi_stop_event: self._svi_stop_event.set() def _sweep_vi_done(self) -> None: self._btn_svi_run.config(state=tk.NORMAL) self._btn_svi_stop.config(state=tk.DISABLED) # Wait for sweep thread to fully exit if self._svi_thread: self._svi_thread.join(timeout=5) self._svi_thread = None self._svi_stop_event = None # Restart normal worker polling if self.bench: interval = 1.0 try: interval = float(self._poll_interval.get()) except ValueError: pass self.worker = InstrumentWorker(self.bench, interval=interval) self.worker.start() self._poll() # ── Logging ─────────────────────────────────────────────────────── _INSTR_LOG_COLUMNS = [ "timestamp", "supply_V", "supply_I", "supply_P", "load_V", "load_I", "load_P", "meter_U5", "meter_I5", "meter_P5", "meter_U6", "meter_I6", "meter_P6", "meter_EFF1", ] # Latest STM32 telemetry snapshot merged into each log row (blank when the # link is down / stale). The full-rate stream goes to _telem.csv. _STM_LOG_COLUMNS = [ "stm_counter", "stm_vin_mV", "stm_vout_mV", "stm_iin_mA", "stm_iin_avg_mA", "stm_iout_mA", "stm_iout_slow_mA", "stm_sys_current_mA", "stm_vfly_mV", "stm_etemp_C", "stm_btemp_C", "stm_vfly_integral", "stm_vfly_avg_debug", "stm_cc_output_f", "stm_mppt_iref", "stm_mppt_last_vin", "stm_mppt_last_iin", "stm_p_in", "stm_p_out", "stm_last_tmp", "stm_VREF", "stm_vfly_correction", "stm_cmp_outer", "stm_cmp_inner", "stm_vfly_ofs_applied", "stm_ctrl_mode", "stm_vfly_active", "stm_status_flags", "stm_fmac_sr", "stm_fault_pc", "stm_cfsr", "stm_param_id", "stm_param_type", "stm_param_value", "stm_pong", "stm_p_in_W", "stm_p_out_W", "stm_p_sys_W", "stm_eff_net_pct", "stm_age_s", ] _LOG_COLUMNS = _INSTR_LOG_COLUMNS + _STM_LOG_COLUMNS def _start_log(self, path: str | None = None) -> None: if path is None: default_name = time.strftime("data_%Y%m%d_%H%M%S.csv") path = filedialog.asksaveasfilename( defaultextension=".csv", filetypes=[("CSV files", "*.csv")], initialfile=default_name, ) if not path: return self._log_file = open(path, "w", newline="") self._log_writer = csv.writer(self._log_file) self._log_writer.writerow(self._LOG_COLUMNS) self._log_count = 0 self._last_log_path = path # Full-rate (100 Hz) telemetry CSV alongside the main log stem = path[:-4] if path.lower().endswith(".csv") else path self._telem_log_path = stem + "_telem.csv" if self.stm32: self.stm32.start_csv(self._telem_log_path) self._console(f"Telemetry CSV: {self._telem_log_path}") self._btn_log_start.config(state=tk.DISABLED) self._btn_log_stop.config(state=tk.NORMAL) self._log_status.config(text=f"Logging to {path}") self._console(f"CSV logging started: {path}") def _stop_log(self) -> None: if self._log_file: self._console(f"CSV logging stopped ({self._log_count} samples)") self._log_file.close() self._log_file = None self._log_writer = None if self.stm32: self.stm32.stop_csv() self._telem_log_path = None self._btn_log_start.config(state=tk.NORMAL) self._btn_log_stop.config(state=tk.DISABLED) self._log_status.config(text=f"Stopped ({self._log_count} samples)") def _plot_eff(self) -> None: """Pick logged CSV(s) and open the efficiency map in a new process.""" kw = {} if self._last_log_path: kw["initialdir"] = os.path.dirname(self._last_log_path) kw["initialfile"] = os.path.basename(self._last_log_path) paths = filedialog.askopenfilenames( title="Select logged CSV(s) to plot", filetypes=[("CSV files", "*.csv"), ("All files", "*.*")], **kw) if not paths: return subprocess.Popen([sys.executable, "-m", "testbench.plot_eff", *paths]) self._console(f"Plotting {len(paths)} file(s) in a new window...") def _log_data(self, data: dict) -> None: if not self._log_writer: return row = [time.strftime("%Y-%m-%d %H:%M:%S")] for col in self._INSTR_LOG_COLUMNS[1:]: row.append(f"{data.get(col, 0.0):.6f}") row.extend(self._stm_log_values()) self._log_writer.writerow(row) self._log_file.flush() self._log_count += 1 self._log_status.config(text=f"Logging... {self._log_count} samples") def _stm_log_values(self) -> list: """Latest telemetry snapshot formatted for the merged log row.""" blanks = [""] * len(self._STM_LOG_COLUMNS) if not self.stm32: return blanks b, wall = self.stm32.get_latest() if b is None: return blanks age = time.time() - wall if age > 2.0: return blanks p_in = b.power_in_W p_out = b.power_out_W p_sys = b.power_sys_W eff_net = ((p_out - p_sys) / p_in * 100.0) if p_in > 0.1 else 0.0 return [ b.counter, f"{b.vin:.6g}", f"{b.vout:.6g}", f"{b.iin:.6g}", b.iin_avg_ma, f"{b.iout:.6g}", f"{b.iout_slow:.6g}", b.sys_current_ma, f"{b.vfly:.6g}", f"{b.etemp:.6g}", f"{b.btemp:.6g}", f"{b.vfly_integral:.6g}", f"{b.vfly_avg_debug:.6g}", f"{b.cc_output_f:.6g}", f"{b.mppt_iref:.6g}", f"{b.mppt_last_vin:.6g}", f"{b.mppt_last_iin:.6g}", f"{b.p_in:.6g}", f"{b.p_out:.6g}", b.last_tmp, b.VREF, b.vfly_correction, b.cmp_outer, b.cmp_inner, b.vfly_ofs_applied, b.ctrl_mode, b.vfly_active, f"0x{b.status_flags:08X}", f"0x{b.fmac_sr:08X}", f"0x{b.fault_pc:08X}", f"0x{b.cfsr:08X}", b.param_id, b.param_type, b.param_value, b.pong, f"{p_in:.4f}", f"{p_out:.4f}", f"{p_sys:.4f}", f"{eff_net:.3f}", f"{age:.2f}", ] def main() -> None: app = TestbenchGUI() app.mainloop() if __name__ == "__main__": main()