diff --git a/.gitignore b/.gitignore index 2f257d6..4c0a7a5 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ build/ .venv/ *.csv !samples/*.csv +logs/ diff --git a/README.md b/README.md index 4cb7c9b..d28cd9d 100644 --- a/README.md +++ b/README.md @@ -84,8 +84,42 @@ The GUI provides: - HIOKI channel range selectors + degauss buttons - Meter format selector (scientific/normal) - 2D sweep panel with time estimate -- Live-updating power, efficiency, voltage, and current plots -- Console log panel +- Live-updating power, efficiency, voltage, and current plots plus STM32 + Vfly and vfly_correction/vfly_ofs_applied plots (selectable) — all graphs + show the last 20 seconds +- STM32 telemetry panel ("Link STM32", COM4 @ 460800 8-O-1): every field of + the 100 Hz broadcast (V/I/P + net efficiency, temps, Vfly group, control + mode, HRTIM compare registers, status flags, fault registers, param echo), + with Ping and Clear Flags buttons; auto-reconnects on serial errors +- Data logging: instrument rows (with the latest STM32 snapshot merged in as + `stm_*` columns) to the chosen CSV, plus the full-rate 100 Hz telemetry + stream to `_telem.csv` alongside. Logging is ON by default: every + launch auto-starts a log at `logs/data_.csv` (relative to the + working directory); use Stop Log / Start Log to switch to a custom path +- Console log panel (STM32 fault flags are reported here as they latch) +- PSU capability guard: the HV supply can source at most 20 A + (`PSU_MAX_CURRENT_A` in `gui.py`). Sweep steps whose estimated input draw + `I_in = P_out / (0.90 * V_in)` exceeds that are rejected (skipped and + reported per voltage), regardless of the requested step range; a measured + backstop additionally drops any point where the supply actually exceeded + the limit and backs the load off. The programmed supply current limit + (sweep "I limit" field and manual supply controls) is clamped to 20 A, and + manual CC/CP load setpoints are checked against live Vin/Vout readings. +- Load range pinning: a mid-sweep auto-range transition on the Prodigit + momentarily unloads the converter, so at sweep start the CC range is + pinned to Range II for the whole run (auto-ranging restored after, with + the load off). The reachable maximum of the selected range is verified + empirically -- the sweep max is programmed with the load off and read + back; if the readback comes back clamped, steps above it are rejected + instead of silently clamped, and a sweep that fits nothing aborts up + front. +- Thermal sweep guard (needs the STM32 link): before every sweep step the + heatsink/board temperatures are checked against the firmware trip limits + (60 C / 80 C). At 57 C / 77 C the sweep pauses, holds the load at 1 A + (CC; ~1 A worth of W in CP), and waits until both temps drop 5 C below + the pause thresholds, then resumes at the same step -- no points are lost. + If the STM32 link is down the sweep still runs, with a console warning + that the guard is inactive. ### 5. Run efficiency sweeps @@ -141,6 +175,32 @@ Produces three PNG files: - `*_heatmap.png` -- 2D efficiency surface (voltage x load) - `*_loss.png` -- power loss vs load, all voltages overlaid +#### Efficiency vs Vin vs current from any logged CSV + +`bench-plot` auto-detects all three CSV formats the tooling produces -- sweep +CSVs, GUI data logs (`data_*.csv`), and full-rate telemetry logs +(`*_telem.csv`) -- and draws an operating-point map (x = Vin, y = current, +color = efficiency) plus efficiency-vs-current curves grouped by Vin bin: + +```bash +uv run bench-plot # no args -> file-picker dialog +uv run bench-plot data_20260703_140000.csv +uv run bench-plot run1_telem.csv run2_telem.csv --vin-bin 2 --save eff.png + +# options: --current iout|iin, --source auto|hioki|instr|stm (data logs), +# --min-pout W (default 5), --vin-bin V (default 1), --save PNG +``` + +Also reachable via the GUI's "Plot Eff..." button (Logging section, opens the +same dialog preselecting the last log) and `plot_eff.bat` one level up +(double-click for the dialog, or drag && drop CSV files onto it). + +For GUI data logs the efficiency source defaults to `auto`: HIOKI EFF1 if the +meter was connected, else supply/load power ratio, else the board's own +`stm_eff_net_pct`. Telemetry logs always use the board's net efficiency +(`(P_out - P_sys) / P_in`, iout_slow). Points below `--min-pout` (default +5 W, same as the GUI display gate) are dropped. + ### 7. Tune converter parameters The tuning commands combine the testbench instruments (ground truth efficiency from HIOKI) with direct STM32 parameter writes to find optimal settings. @@ -317,6 +377,8 @@ Names, IDs, types and ranges mirror the firmware (`code64/debug_console/protocol | `dither_band_lo` / `dither_band_hi` | uint16 | 716-6442 | Forbidden duty band edges (CMP ticks) | | `dither_anear` / `dither_afar` | uint16 | 716-6442 | Out-of-band dither anchors | | `dither_dzero` | uint16 | 716-6442 | \|e\| fold center (D=0.5) | +| `adc4_trig_phase` | uint16 | 3-14313 | HRTIM master CMP3: iout_slow sample instant | +| `iin_zero_sum` | uint16 | 0-32760 | IIN software zero offset (sum-of-8 counts) | ## CSV Output Format @@ -335,6 +397,16 @@ Sweep CSV files contain: Tuning CSV files additionally contain `param_name`, `param_value`, and STM32 telemetry columns (`stm_vin`, `stm_vout`, `stm_iin`, `stm_iout`, `stm_eff`, `stm_vfly`, `stm_etemp`). +GUI data-log CSVs contain the instrument columns plus the latest STM32 +broadcast snapshot per row (`stm_counter` … `stm_age_s`; flag/fault registers +as hex). While logging, the full-rate 100 Hz telemetry stream is additionally +written to `_telem.csv` with every broadcast field (one row per fresh +publish, `pc_time`/`t_mono` timestamps, computed `p_in_W`/`p_out_W`). + +Note: STM32-derived `power_out_W`/`efficiency` now use `iout_slow` (the +PWM-synchronous ADC4 output current) instead of the fast protection-path +`iout` — tuner numbers shift slightly vs. old logs. + ## Project Structure ``` @@ -348,7 +420,8 @@ mppt-testbench/ | +-- cli.py unified CLI entry point | +-- gui.py tkinter GUI with live plots | +-- gui_workers.py background instrument I/O thread -| +-- stm32_link.py synchronous STM32 debug protocol interface +| +-- stm32_link.py STM32 debug protocol: 114B broadcast RX + CRC-framed TX commands (8-O-1) +| +-- plot_eff.py efficiency vs Vin vs current plots from any logged CSV | +-- tuner.py automated tuning routines (param sweep, deadtime opt) +-- code64/ | +-- Core/ STM32G474 firmware (C) @@ -356,7 +429,7 @@ mppt-testbench/ | +-- debug_console/ Textual TUI for live debugging | +-- pyproject.toml uv-compatible package config +-- samples/ shade profile CSV examples -+-- pyproject.toml package config, entry points: bench, bench-gui ++-- pyproject.toml package config, entry points: bench, bench-gui, bench-plot ``` ## Dependencies diff --git a/pyproject.toml b/pyproject.toml index 4ba854e..27dac7f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,3 +26,4 @@ packages = ["testbench"] [project.scripts] bench = "testbench.cli:main" bench-gui = "testbench.gui:main" +bench-plot = "testbench.plot_eff:main" diff --git a/testbench/bench.py b/testbench/bench.py index b525b33..9005943 100644 --- a/testbench/bench.py +++ b/testbench/bench.py @@ -138,7 +138,7 @@ class MPPTTestbench: self.meter.set_current_auto(5, True) self.meter.set_voltage_auto(6, True) self.meter.set_current_auto(6, True) - self.meter.set_response_speed("SLOW") + self.meter.set_response_speed("FAST") self.meter.set_efficiency(1, "P6", "P5") # Display: 16-item SELECT view @@ -300,7 +300,7 @@ class MPPTTestbench: v_stop: float, v_step: float, current_limit: float, - settle_time: float = 1.0, + settle_time: float = 5.0, load_setpoint: float = 0.0, ) -> list[SweepPoint]: """Sweep supply voltage and record measurements at each point. @@ -366,7 +366,7 @@ class MPPTTestbench: i_start: float, i_stop: float, i_step: float, - settle_time: float = 1.0, + settle_time: float = 5.0, ) -> list[SweepPoint]: """Sweep load current (CC mode) at a fixed supply voltage. @@ -473,7 +473,7 @@ class MPPTTestbench: def run_shade_profile( self, steps: list[dict], - settle_time: float = 2.0, + settle_time: float = 5.0, ) -> list[SweepPoint]: """Run a shade / irradiance profile sequence. @@ -629,7 +629,7 @@ class MPPTTestbench: l_stop: float, l_step: float, current_limit: float, - settle_time: float = 2.0, + settle_time: float = 5.0, load_mode: str = "CC", ) -> list[SweepPoint]: """2D sweep: voltage (outer) × load setpoint (inner). @@ -770,7 +770,7 @@ class MPPTTestbench: self, voltage: float, current_limit: float, - settle_time: float = 2.0, + settle_time: float = 5.0, samples: int = 5, sample_interval: float = 1.0, ) -> dict[str, float]: diff --git a/testbench/cli.py b/testbench/cli.py index 1745ac1..939f83d 100644 --- a/testbench/cli.py +++ b/testbench/cli.py @@ -572,20 +572,29 @@ def cmd_stm32_read(bench: MPPTTestbench, args: argparse.Namespace) -> None: print(f" {name:<20s} = {val}") print() - # Read telemetry + # Read telemetry (100 Hz broadcast, 20 fresh publishes averaged) t = link.read_telemetry_avg(n=20) if t: + from testbench.stm32_link import flags_to_names + modes = ("OFF", "MPPT", "CV", "CC") + mode = modes[t.ctrl_mode] if 0 <= t.ctrl_mode < len(modes) else str(t.ctrl_mode) print("Telemetry (20-sample avg):") print(f" Vin = {t.vin_V:8.2f} V") print(f" Vout = {t.vout_V:8.2f} V") - print(f" Iin = {t.iin_A:8.2f} A") - print(f" Iout = {t.iout_A:8.2f} A") + print(f" Iin = {t.iin_A:8.2f} A (avg {t.iin_avg_ma:+d} mA)") + print(f" Iout = {t.iout_slow/1000:8.2f} A (fast {t.iout_A:.2f} A)") + print(f" Isys = {t.sys_current_ma:8d} mA") print(f" Pin = {t.power_in_W:8.2f} W") print(f" Pout = {t.power_out_W:8.2f} W") - print(f" EFF = {t.efficiency:8.1f} %") + print(f" EFF = {t.efficiency:8.1f} % (net {t.efficiency_net:.1f} %)") print(f" Vfly = {t.vfly/1000:8.2f} V") - print(f" Temp = {t.etemp:8.1f} °C (FET)") + print(f" Temp = {t.etemp:8.1f} °C (heatsink)") print(f" Tbrd = {t.btemp:8.1f} °C (board)") + print(f" Mode = {mode}") + if t.status_flags: + print(f" Flags= 0x{t.status_flags:08X}") + for name in flags_to_names(t.status_flags): + print(f" {name}") def cmd_stm32_write(bench: MPPTTestbench, args: argparse.Namespace) -> None: @@ -1192,7 +1201,7 @@ examples: p_sweep.add_argument("--v-stop", type=float, required=True, help="Stop voltage (V)") p_sweep.add_argument("--v-step", type=float, required=True, help="Voltage step (V)") p_sweep.add_argument("--current-limit", type=float, required=True, help="Current limit (A)") - p_sweep.add_argument("--settle", type=float, default=1.0, help="Settle time per step (s)") + p_sweep.add_argument("--settle", type=float, default=5.0, help="Settle time per step (s)") p_sweep.add_argument("--load-mode", choices=["CC", "CR", "CV", "CP"], help="Set load mode before sweep") p_sweep.add_argument("--load-value", type=float, help="Set load value before sweep") p_sweep.add_argument("-o", "--output", help="CSV output file") @@ -1204,7 +1213,7 @@ examples: p_swl.add_argument("--i-start", type=float, required=True, help="Start load current (A)") p_swl.add_argument("--i-stop", type=float, required=True, help="Stop load current (A)") p_swl.add_argument("--i-step", type=float, required=True, help="Current step (A)") - p_swl.add_argument("--settle", type=float, default=1.0, help="Settle time per step (s)") + p_swl.add_argument("--settle", type=float, default=5.0, help="Settle time per step (s)") p_swl.add_argument("-o", "--output", help="CSV output file") # efficiency @@ -1212,7 +1221,7 @@ examples: p_eff.add_argument("--voltage", type=float, required=True, help="Supply voltage (V)") p_eff.add_argument("--current-limit", type=float, required=True, help="Current limit (A)") p_eff.add_argument("--samples", type=int, default=5, help="Number of readings to average") - p_eff.add_argument("--settle", type=float, default=2.0, help="Initial settle time (s)") + p_eff.add_argument("--settle", type=float, default=5.0, help="Initial settle time (s)") p_eff.add_argument("-i", "--interval", type=float, default=1.0, help="Interval between samples") p_eff.add_argument("--load-mode", choices=["CC", "CR", "CV", "CP"]) p_eff.add_argument("--load-value", type=float) @@ -1227,13 +1236,13 @@ examples: p_svi.add_argument("--l-step", type=float, required=True, help="Load step size") p_svi.add_argument("--load-mode", choices=["CC", "CP"], default="CC", help="Load mode: CC (current) or CP (power)") p_svi.add_argument("--current-limit", type=float, required=True, help="Supply current limit (A)") - p_svi.add_argument("--settle", type=float, default=2.0, help="Settle time per step (s)") + p_svi.add_argument("--settle", type=float, default=5.0, help="Settle time per step (s)") p_svi.add_argument("-o", "--output", help="CSV output file") # shade-profile p_shade = sub.add_parser("shade-profile", help="Run a shade/irradiance profile from CSV") p_shade.add_argument("--profile", required=True, help="Profile CSV file (time,voltage,current_limit,...)") - p_shade.add_argument("--settle", type=float, default=2.0, help="Settle time per step (s)") + p_shade.add_argument("--settle", type=float, default=5.0, help="Settle time per step (s)") p_shade.add_argument("-o", "--output", help="CSV output file for results") # supply (direct control) @@ -1275,7 +1284,7 @@ examples: p_tp.add_argument("--current-limit", type=float, required=True, help="Supply current limit (A)") p_tp.add_argument("--load-mode", choices=["CC", "CP"], default="CP", help="Load mode") p_tp.add_argument("--load-value", type=float, default=200.0, help="Load setpoint (A or W)") - p_tp.add_argument("--settle", type=float, default=3.0, help="Settle time per step (s)") + p_tp.add_argument("--settle", type=float, default=5.0, help="Settle time per step (s)") p_tp.add_argument("--no-plot", action="store_true", help="Skip plot") p_tp.add_argument("-o", "--output", help="CSV output file") @@ -1291,7 +1300,7 @@ examples: p_tv.add_argument("--current-limit", type=float, required=True, help="Supply current limit (A)") p_tv.add_argument("--load-mode", choices=["CC", "CP"], default="CP", help="Load mode") p_tv.add_argument("--load-value", type=float, default=200.0, help="Load setpoint (A or W)") - p_tv.add_argument("--settle", type=float, default=3.0, help="Settle time per step (s)") + p_tv.add_argument("--settle", type=float, default=5.0, help="Settle time per step (s)") p_tv.add_argument("-o", "--output", help="Output prefix for CSVs (default: param name)") # tune-deadtime @@ -1303,7 +1312,7 @@ examples: p_td.add_argument("--current-limit", type=float, default=20.0, help="Supply current limit (A)") p_td.add_argument("--load-mode", choices=["CC", "CP"], default="CP", help="Load mode") p_td.add_argument("--load-values", help="Comma-separated load values to test (e.g. 100,300,500)") - p_td.add_argument("--settle", type=float, default=3.0, help="Settle time per step (s)") + p_td.add_argument("--settle", type=float, default=5.0, help="Settle time per step (s)") p_td.add_argument("--apply", action="store_true", help="Apply best deadtimes after sweep") p_td.add_argument("-o", "--output", help="CSV output file") diff --git a/testbench/gui.py b/testbench/gui.py index 81ac37f..6803090 100644 --- a/testbench/gui.py +++ b/testbench/gui.py @@ -6,7 +6,10 @@ 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 @@ -23,11 +26,78 @@ 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 +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: @@ -79,16 +149,31 @@ class TestbenchGUI(tk.Tk): def __init__(self) -> None: super().__init__() self.title("MPPT Testbench Control Panel") - self.geometry("1500x950") - self.minsize(1200, 800) + 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 @@ -109,6 +194,17 @@ class TestbenchGUI(tk.Tk): 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: @@ -143,6 +239,10 @@ class TestbenchGUI(tk.Tk): 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() @@ -170,6 +270,16 @@ class TestbenchGUI(tk.Tk): 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) @@ -345,7 +455,7 @@ class TestbenchGUI(tk.Tk): 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("SLOW") + 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) @@ -468,6 +578,7 @@ class TestbenchGUI(tk.Tk): 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) @@ -486,7 +597,7 @@ class TestbenchGUI(tk.Tk): 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, "2.0") + self._profile_settle.insert(0, "5.0") self._profile_settle.pack(side=tk.LEFT, padx=2) row = ttk.Frame(frame) @@ -556,11 +667,11 @@ class TestbenchGUI(tk.Tk): 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, "35") + 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, "2.0") + 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) @@ -588,13 +699,26 @@ class TestbenchGUI(tk.Tk): self._svi_stop_event = None def _build_graphs(self, parent) -> None: - self._fig = Figure(figsize=(10, 8), dpi=100) + # 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(4, 1, 1) - self._ax_eff = self._fig.add_subplot(4, 1, 2) - self._ax_volt = self._fig.add_subplot(4, 1, 3) - self._ax_curr = self._fig.add_subplot(4, 1, 4) + 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)") @@ -623,13 +747,32 @@ class TestbenchGUI(tk.Tk): # Current self._ax_curr.set_ylabel("Current (A)") - self._ax_curr.set_xlabel("Time (s)") 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) @@ -689,6 +832,310 @@ class TestbenchGUI(tk.Tk): 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: @@ -715,6 +1162,10 @@ class TestbenchGUI(tk.Tk): 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) @@ -729,9 +1180,14 @@ class TestbenchGUI(tk.Tk): self._console(f"Connected: supply={supply_addr}, load={load_port}, meter={meter_addr}", "success") self._poll() - except Exception as e: - self._console(f"Connection failed: {e}", "error") - messagebox.showerror("Connection Error", str(e)) + # 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() @@ -773,6 +1229,7 @@ class TestbenchGUI(tk.Tk): def _on_close(self) -> None: self._stop_log() + self._disconnect_stm32() if self.bench: try: self.bench.safe_off() @@ -806,6 +1263,7 @@ class TestbenchGUI(tk.Tk): 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) @@ -879,13 +1337,20 @@ class TestbenchGUI(tk.Tk): ) def _update_graphs(self, data: dict) -> None: - """Append data to series and redraw graphs.""" - now = time.time() - self._t0 + """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"])) @@ -897,11 +1362,7 @@ class TestbenchGUI(tk.Tk): self._ln_i5.set_data(t, list(self._series["meter_I5"])) self._ln_i6.set_data(t, list(self._series["meter_I6"])) - for ax in [self._ax_power, self._ax_eff, self._ax_volt, self._ax_curr]: - ax.relim() - ax.autoscale_view() - - self._canvas.draw_idle() + self._redraw_axes() # ── Format Helpers ───────────────────────────────────────────────── @@ -990,9 +1451,24 @@ class TestbenchGUI(tk.Tk): 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()) @@ -1010,10 +1486,27 @@ class TestbenchGUI(tk.Tk): try: val = float(self._load_value.get()) mode = self._load_mode.get() - self._send(Cmd.SET_MODE_VALUE, mode, val) 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.""" @@ -1057,7 +1550,7 @@ class TestbenchGUI(tk.Tk): try: settle = float(self._profile_settle.get()) except ValueError: - settle = 2.0 + settle = 5.0 n = len(self._profile_steps) dur = self._profile_steps[-1]["time"] if self._profile_steps else 0 @@ -1157,6 +1650,14 @@ class TestbenchGUI(tk.Tk): 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( @@ -1215,6 +1716,53 @@ class TestbenchGUI(tk.Tk): 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 @@ -1255,15 +1803,66 @@ class TestbenchGUI(tk.Tk): 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.load.set_mode(load_mode) 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(): @@ -1274,6 +1873,7 @@ class TestbenchGUI(tk.Tk): 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: @@ -1281,7 +1881,24 @@ class TestbenchGUI(tk.Tk): 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 @@ -1289,6 +1906,26 @@ class TestbenchGUI(tk.Tk): 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 = { @@ -1328,10 +1965,14 @@ class TestbenchGUI(tk.Tk): ) 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 = ll - l_step if n > 0 else l_start + 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: @@ -1347,6 +1988,11 @@ class TestbenchGUI(tk.Tk): 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 @@ -1400,7 +2046,7 @@ class TestbenchGUI(tk.Tk): # ── Logging ─────────────────────────────────────────────────────── - _LOG_COLUMNS = [ + _INSTR_LOG_COLUMNS = [ "timestamp", "supply_V", "supply_I", "supply_P", "load_V", "load_I", "load_P", @@ -1409,19 +2055,46 @@ class TestbenchGUI(tk.Tk): "meter_EFF1", ] - def _start_log(self) -> 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 + # 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}") @@ -1433,21 +2106,72 @@ class TestbenchGUI(tk.Tk): 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._LOG_COLUMNS[1:]: + 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() diff --git a/testbench/gui_workers.py b/testbench/gui_workers.py index 3f4bb5a..25171b9 100644 --- a/testbench/gui_workers.py +++ b/testbench/gui_workers.py @@ -6,11 +6,19 @@ never freezes during instrument queries. from __future__ import annotations +import csv import queue import threading import time +from collections import deque from enum import Enum, auto +import serial + +from testbench.stm32_link import ( + BroadcastData, BroadcastParser, PARAM_BY_ID, decode_param_bits, +) + class Cmd(Enum): """Commands sent from GUI to worker thread.""" @@ -226,3 +234,214 @@ class InstrumentWorker(threading.Thread): }) except queue.Full: pass + + +# ── STM32 broadcast reader ─────────────────────────────────────────── + +# All BroadcastData wire fields, in dataclass order (full-rate CSV columns). +TELEM_CSV_FIELDS = ( + "counter", "vin", "vout", "iin", "iout", "vfly", "etemp", "btemp", + "vfly_integral", "vfly_avg_debug", "cc_output_f", "mppt_iref", + "mppt_last_vin", "mppt_last_iin", "p_in", "p_out", "iout_slow", + "last_tmp", "VREF", "vfly_correction", "cmp_outer", "cmp_inner", + "vfly_ofs_applied", "ctrl_mode", "vfly_active", "status_flags", + "fmac_sr", "fault_pc", "cfsr", "param_id", "param_type", + "param_value", "pong", "sys_current_ma", "iin_avg_ma", +) +_TELEM_HEX_FIELDS = frozenset({"status_flags", "fmac_sr", "fault_pc", "cfsr"}) + + +class STM32Worker(threading.Thread): + """Daemon thread reading the STM32 broadcast stream. + + The board streams 114-byte frames continuously (100 Hz publishes, each + repeated ~3-4x at line rate); this thread validates them, keeps the + latest sample plus 20 s graph history, and owns the full-rate telemetry + CSV (single writer). Auto-reconnects on serial errors. + """ + + HISTORY = 2200 # 100 Hz x 20 s + margin + CSV_FLUSH_S = 1.0 # never flush per-row at 100 rows/s + + def __init__(self, port: str, baudrate: int = 460800) -> None: + super().__init__(daemon=True) + self.port = port + self.baudrate = baudrate + self.connected = False + self._stop_event = threading.Event() + self._tx_queue: queue.Queue = queue.Queue() + + self._lock = threading.Lock() + self.latest: BroadcastData | None = None + self.latest_wall: float = 0.0 + self._last_counter = -1 + self._t: deque = deque(maxlen=self.HISTORY) # time.monotonic per fresh publish + self._vfly: deque = deque(maxlen=self.HISTORY) # mV + self._corr: deque = deque(maxlen=self.HISTORY) # vfly_correction, ticks + self._ofs: deque = deque(maxlen=self.HISTORY) # vfly_ofs_applied, ticks + self._fresh_count = 0 # validated fresh publishes since last get_rates() + self._sent_count = 0 # publishes the board sent (counter deltas) + self.params: dict[int, float] = {} + + self._csv_lock = threading.Lock() + self._csv_file = None + self._csv_writer = None + self._csv_last_flush = 0.0 + + # ── Thread-safe API for the GUI ────────────────────────────────── + + def get_latest(self) -> tuple[BroadcastData | None, float]: + """Latest validated sample and its wall-clock arrival time.""" + with self._lock: + return self.latest, self.latest_wall + + def get_graph_snapshot(self, decimate: int = 4): + """(t_mono, vfly_mV, corr, ofs) lists, decimated for display. + + Decimation is anchored at the end so the newest sample always shows. + Full rate is still recorded to the CSV. + """ + with self._lock: + t, v = list(self._t), list(self._vfly) + c, o = list(self._corr), list(self._ofs) + if decimate > 1 and t: + k = (len(t) - 1) % decimate + t, v, c, o = t[k::decimate], v[k::decimate], c[k::decimate], o[k::decimate] + return t, v, c, o + + def get_rates(self) -> tuple[int, int]: + """(fresh publishes received, publishes sent) since the last call.""" + with self._lock: + f, s = self._fresh_count, self._sent_count + self._fresh_count = 0 + self._sent_count = 0 + return f, s + + def send_frame(self, data: bytes) -> None: + """Queue a pre-built command frame for TX on the worker thread.""" + self._tx_queue.put(data) + + def start_csv(self, path: str) -> None: + """Open the full-rate telemetry CSV (one row per fresh publish).""" + with self._csv_lock: + self._close_csv_locked() + f = open(path, "w", newline="", encoding="utf-8") + w = csv.writer(f) + w.writerow(("pc_time", "t_mono") + TELEM_CSV_FIELDS + ("p_in_W", "p_out_W")) + self._csv_file, self._csv_writer = f, w + self._csv_last_flush = time.monotonic() + + def stop_csv(self) -> None: + with self._csv_lock: + self._close_csv_locked() + + def _close_csv_locked(self) -> None: + if self._csv_file is not None: + try: + self._csv_file.close() + except OSError: + pass + self._csv_file = None + self._csv_writer = None + + def stop(self) -> None: + """Stop the thread and close the CSV (blocks up to 2 s).""" + self._stop_event.set() + if self.is_alive(): + self.join(timeout=2.0) + self.stop_csv() + + # ── Worker loop ────────────────────────────────────────────────── + + def run(self) -> None: + ser = None + parser = None + while not self._stop_event.is_set(): + if ser is None: + try: + ser = serial.Serial( + self.port, self.baudrate, timeout=0.05, + bytesize=serial.EIGHTBITS, + parity=serial.PARITY_ODD, + stopbits=serial.STOPBITS_ONE, + ) + parser = BroadcastParser() + self.connected = True + except (serial.SerialException, OSError): + self.connected = False + self._stop_event.wait(1.0) + continue + try: + while True: + try: + ser.write(self._tx_queue.get_nowait()) + except queue.Empty: + break + data = ser.read(4096) + if data: + for b in parser.feed(data): + self._on_frame(b) + except (serial.SerialException, OSError): + self.connected = False + try: + ser.close() + except Exception: + pass + ser = None + self._stop_event.wait(1.0) + if ser is not None: + try: + ser.close() + except Exception: + pass + self.connected = False + self.stop_csv() + + # NB: name must not collide with threading.Thread instance attributes + # (Thread.__init__ sets self._handle on Python 3.13+). + def _on_frame(self, b: BroadcastData) -> None: + now = time.monotonic() + with self._lock: + # Flags/pong are valid even at counter == 0 (post-reset window); + # telemetry/param fields are only valid once counter > 0. + self.latest = b + self.latest_wall = time.time() + fresh = b.counter > 0 and b.counter != self._last_counter + if fresh: + if 0 < self._last_counter < b.counter: + self._sent_count += b.counter - self._last_counter + else: + self._sent_count += 1 # first valid frame, or a reboot + self._fresh_count += 1 + self._last_counter = b.counter + self._t.append(now) + self._vfly.append(b.vfly) + self._corr.append(b.vfly_correction) + self._ofs.append(b.vfly_ofs_applied) + if b.param_id in PARAM_BY_ID: + self.params[b.param_id] = decode_param_bits(b.param_type, b.param_value) + if fresh: + self._csv_row(b, now) + + def _csv_row(self, b: BroadcastData, t_mono: float) -> None: + with self._csv_lock: + if self._csv_writer is None: + return + row = [f"{b.timestamp:.3f}", f"{t_mono:.3f}"] + for name in TELEM_CSV_FIELDS: + v = getattr(b, name) + if name in _TELEM_HEX_FIELDS: + row.append(f"0x{v:08X}") + elif isinstance(v, float): + row.append(f"{v:.6g}") + else: + row.append(v) + row.append(f"{b.power_in_W:.4f}") + row.append(f"{b.power_out_W:.4f}") + try: + self._csv_writer.writerow(row) + if t_mono - self._csv_last_flush >= self.CSV_FLUSH_S: + self._csv_file.flush() + self._csv_last_flush = t_mono + except OSError: + self._close_csv_locked() diff --git a/testbench/plot_eff.py b/testbench/plot_eff.py new file mode 100644 index 0000000..54a11a1 --- /dev/null +++ b/testbench/plot_eff.py @@ -0,0 +1,234 @@ +"""Plot efficiency vs input voltage vs current from bench CSV logs. + +Auto-detects the three CSV formats produced by the tooling: + - GUI data log (data_*.csv: instrument columns + merged stm_* snapshot) + - GUI telemetry log (*_telem.csv: full-rate 100 Hz board broadcast) + - CLI sweep (sweep_vi_*.csv: voltage_set/load_setpoint grid) + +Left panel: operating-point scatter (x = Vin, y = current, color = efficiency). +Right panel: efficiency vs current, one curve per Vin bin. + +Usage: + bench-plot data_20260703_120000.csv + bench-plot run_telem.csv another_telem.csv --current iin --save eff.png +""" + +from __future__ import annotations + +import argparse +import csv +import math +import sys + +import matplotlib.pyplot as plt +import numpy as np + +MIN_P_IN_W = 0.1 # same gate as the firmware/GUI efficiency calc + + +def _f(row: dict, key: str) -> float: + """Float cell value; blank/missing/garbage -> NaN.""" + v = row.get(key, "") + if v is None or v == "": + return math.nan + try: + return float(v) + except ValueError: + return math.nan + + +def _detect_format(header: list[str]) -> str: + cols = set(header) + if "stm_eff_net_pct" in cols: + return "datalog" + if "iout_slow" in cols and "p_in_W" in cols: + return "telem" + if "voltage_set" in cols and "efficiency" in cols: + return "sweep" + raise ValueError(f"unrecognized CSV header: {header[:6]}...") + + +def _extract(path: str, source: str) -> tuple[dict, str]: + """Read one CSV -> dict of float lists (vin_V, iin_A, iout_A, eff_pct, + p_out_W) plus the efficiency-source label actually used.""" + with open(path, newline="") as fh: + reader = csv.DictReader(fh) + header = reader.fieldnames or [] + rows = list(reader) + fmt = _detect_format(header) + + out: dict[str, list] = {k: [] for k in ("vin_V", "iin_A", "iout_A", "eff_pct", "p_out_W")} + + if fmt == "sweep": + for r in rows: + out["vin_V"].append(_f(r, "supply_V")) + out["iin_A"].append(_f(r, "supply_I")) + out["iout_A"].append(_f(r, "load_I")) + out["eff_pct"].append(_f(r, "efficiency")) + out["p_out_W"].append(_f(r, "output_power")) + return out, "sweep efficiency (HIOKI)" + + if fmt == "telem": + # Wire units: vin/vout in mV, currents in mA (iin negative into the + # converter); eff net = (P_out - P_sys) / P_in, same as the GUI panel. + for r in rows: + p_in = _f(r, "p_in_W") + p_out = _f(r, "p_out_W") + p_sys = _f(r, "vout") * _f(r, "sys_current_ma") / 1e6 + eff = (p_out - p_sys) / p_in * 100.0 if p_in > MIN_P_IN_W else math.nan + out["vin_V"].append(_f(r, "vin") / 1000.0) + out["iin_A"].append(-_f(r, "iin") / 1000.0) + out["iout_A"].append(_f(r, "iout_slow") / 1000.0) + out["eff_pct"].append(eff) + out["p_out_W"].append(p_out) + return out, "board eff net (iout_slow, -P_sys)" + + # datalog: three consistent (vin, current, eff) triples to choose from + if source == "auto": + med_eff1 = np.nanmedian([_f(r, "meter_EFF1") for r in rows]) if rows else math.nan + med_psup = np.nanmedian([_f(r, "supply_P") for r in rows]) if rows else math.nan + if med_eff1 > 1.0: + source = "hioki" + elif med_psup > MIN_P_IN_W: + source = "instr" + else: + source = "stm" + + for r in rows: + if source == "hioki": + out["vin_V"].append(_f(r, "meter_U5")) + out["iin_A"].append(_f(r, "meter_I5")) + out["iout_A"].append(_f(r, "meter_I6")) + out["eff_pct"].append(_f(r, "meter_EFF1")) + out["p_out_W"].append(_f(r, "meter_P6")) + elif source == "instr": + p_sup = _f(r, "supply_P") + eff = _f(r, "load_P") / p_sup * 100.0 if p_sup > MIN_P_IN_W else math.nan + out["vin_V"].append(_f(r, "supply_V")) + out["iin_A"].append(_f(r, "supply_I")) + out["iout_A"].append(_f(r, "load_I")) + out["eff_pct"].append(eff) + out["p_out_W"].append(_f(r, "load_P")) + else: # stm + out["vin_V"].append(_f(r, "stm_vin_mV") / 1000.0) + out["iin_A"].append(-_f(r, "stm_iin_mA") / 1000.0) + out["iout_A"].append(_f(r, "stm_iout_slow_mA") / 1000.0) + out["eff_pct"].append(_f(r, "stm_eff_net_pct")) + out["p_out_W"].append(_f(r, "stm_p_out_W")) + labels = {"hioki": "HIOKI EFF1", "instr": "supply/load power", + "stm": "board eff net"} + return out, labels[source] + + +def main() -> None: + ap = argparse.ArgumentParser( + description="Plot efficiency vs input voltage vs current from bench CSVs.") + ap.add_argument("csv", nargs="*", + help="logged CSV file(s); a file dialog opens if omitted") + ap.add_argument("--current", choices=("iout", "iin"), default="iout", + help="current axis: output (default) or input current") + ap.add_argument("--source", choices=("auto", "hioki", "instr", "stm"), + default="auto", + help="efficiency source for GUI data logs (default auto: " + "HIOKI if present, else supply/load, else board)") + ap.add_argument("--vin-bin", type=float, default=1.0, metavar="V", + help="Vin bin width for the per-voltage curves (default 1.0)") + ap.add_argument("--min-pout", type=float, default=5.0, metavar="W", + help="drop points below this output power (default 5.0)") + ap.add_argument("--save", metavar="PNG", help="write the figure instead of showing it") + ap.add_argument("--title", default=None, help="figure title override") + args = ap.parse_args() + + if not args.csv: + import tkinter as tk + from tkinter import filedialog + root = tk.Tk() + root.withdraw() + args.csv = list(filedialog.askopenfilenames( + title="Select logged CSV(s) to plot", + filetypes=[("CSV files", "*.csv"), ("All files", "*.*")])) + root.destroy() + if not args.csv: + sys.exit("no file selected") + + data: dict[str, list] = {k: [] for k in ("vin_V", "iin_A", "iout_A", "eff_pct", "p_out_W")} + labels = set() + for path in args.csv: + part, label = _extract(path, args.source) + for k in data: + data[k].extend(part[k]) + labels.add(label) + + vin = np.asarray(data["vin_V"]) + cur = np.asarray(data["iin_A" if args.current == "iin" else "iout_A"]) + eff = np.asarray(data["eff_pct"]) + pout = np.asarray(data["p_out_W"]) + + keep = (np.isfinite(vin) & np.isfinite(cur) & np.isfinite(eff) + & (eff > 0.0) & (eff <= 105.0) & (pout >= args.min_pout)) + n_total = len(vin) + vin, cur, eff = vin[keep], cur[keep], eff[keep] + if len(vin) == 0: + sys.exit(f"no usable points ({n_total} rows read; all filtered — " + f"check --min-pout / --source)") + + ipk = int(np.argmax(eff)) + print(f"{len(vin)} points ({n_total - len(vin)} filtered) | " + f"Vin {vin.min():.1f}..{vin.max():.1f} V | " + f"I {cur.min():.2f}..{cur.max():.2f} A | " + f"peak eff {eff[ipk]:.2f} % @ {vin[ipk]:.1f} V, {cur[ipk]:.2f} A") + + cur_name = "Input current (A)" if args.current == "iin" else "Output current (A)" + fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(13.5, 5.8)) + fig.suptitle(args.title or f"Efficiency map — {', '.join(sorted(labels))}") + + # Left: operating-point scatter, color = efficiency + vmin = np.percentile(eff, 5) + sc = ax1.scatter(vin, cur, c=eff, s=14, cmap="viridis", + vmin=vmin, vmax=eff.max(), rasterized=True) + fig.colorbar(sc, ax=ax1, label="Efficiency (%)") + ax1.plot(vin[ipk], cur[ipk], "r*", ms=14, mec="k", + label=f"peak {eff[ipk]:.2f} %") + ax1.set_xlabel("Input voltage (V)") + ax1.set_ylabel(cur_name) + ax1.legend(loc="best", fontsize=8) + ax1.grid(alpha=0.3) + + # Right: efficiency vs current, one mean curve per Vin bin + w = args.vin_bin + centers = np.unique(np.round(vin / w) * w) + cmap = plt.cm.plasma(np.linspace(0.0, 0.9, len(centers))) + for color, c0 in zip(cmap, centers): + m = np.abs(vin - c0) <= w / 2 + if m.sum() < 2: + ax2.plot(cur[m], eff[m], "o", color=color, ms=4, + label=f"{c0:g} V") + continue + edges = np.linspace(cur[m].min(), cur[m].max() + 1e-9, 41) + idx = np.digitize(cur[m], edges) + xs, ys = [], [] + for b in np.unique(idx): + bm = idx == b + xs.append(cur[m][bm].mean()) + ys.append(eff[m][bm].mean()) + ax2.plot(xs, ys, "-o", color=color, ms=3, lw=1.2, label=f"{c0:g} V") + ax2.set_xlabel(cur_name) + ax2.set_ylabel("Efficiency (%)") + ax2.grid(alpha=0.3) + if len(centers) <= 14: + ax2.legend(title="Vin bin", fontsize=8, ncols=1 + len(centers) // 8) + else: + norm = plt.Normalize(centers.min(), centers.max()) + fig.colorbar(plt.cm.ScalarMappable(norm=norm, cmap="plasma"), + ax=ax2, label="Vin bin (V)") + + fig.tight_layout() + if args.save: + fig.savefig(args.save, dpi=140) + print(f"saved: {args.save}") + else: + plt.show() + + +if __name__ == "__main__": + main() diff --git a/testbench/stm32_link.py b/testbench/stm32_link.py index c76a4df..3b68ea1 100644 --- a/testbench/stm32_link.py +++ b/testbench/stm32_link.py @@ -1,17 +1,20 @@ """Synchronous serial link to the STM32 debug protocol. -Provides blocking read/write of telemetry and parameters, suitable -for automated tuning scripts (not a TUI). Mirrors the binary protocol -from code64/debug_console/protocol.py (kept in sync with the firmware's -debug_protocol.h — CRC-16, 78-byte telemetry, current parameter map). +STM32 -> PC is a continuous 114-byte binary broadcast (magic-delimited, +repetition-validated, UART odd parity, ~100 Hz publish rate, each publish +repeated ~3-4x at line rate). PC -> STM32 commands remain CRC-16 framed. +Mirrors code64/debug_console/protocol.py (kept in sync with the firmware's +debug_protocol.h). """ from __future__ import annotations +import dataclasses import struct import time +from collections import deque from dataclasses import dataclass, field -from typing import Optional +from typing import Iterator, Optional import serial @@ -19,13 +22,13 @@ import serial SYNC_BYTE = 0xAA -CMD_TELEMETRY = 0x01 +CMD_TELEMETRY = 0x01 # legacy; STM32->PC framed telemetry no longer sent CMD_PARAM_WRITE = 0x02 -CMD_PARAM_WRITE_ACK = 0x03 -CMD_PARAM_READ_ALL = 0x04 -CMD_PARAM_VALUE = 0x05 +CMD_PARAM_WRITE_ACK = 0x03 # legacy; acks now come via the broadcast round-robin +CMD_PARAM_READ_ALL = 0x04 # restarts the broadcast param round-robin cursor +CMD_PARAM_VALUE = 0x05 # legacy; STM32->PC framed replies no longer sent CMD_PING = 0x10 -CMD_PONG = 0x11 +CMD_PONG = 0x11 # legacy; pong now increments a broadcast field CMD_SHUTDOWN = 0x12 # turn off converter CMD_RESET = 0x13 # system reset CMD_TEST_50 = 0x14 # 50% duty test mode @@ -33,7 +36,8 @@ CMD_RELAY_ON = 0x15 # latch input relay closed (bench test) CMD_RELAY_OFF = 0x16 # latch input relay open (bench test) CMD_HOLD_CONVERTER = 0x17 # toggle "hold converter off" (boot guard + disarm trips) CMD_TOGGLE_PRECHARGE = 0x18 # toggle the precharge FET (bench test) -CMD_ERROR_MSG = 0xE0 +CMD_CLEAR_FLAGS = 0x19 # clear latched status flags +CMD_ERROR_MSG = 0xE0 # legacy; STM32->PC no longer sends framed text PTYPE_FLOAT = 0 PTYPE_UINT16 = 1 @@ -57,36 +61,116 @@ def crc16(data: bytes) -> int: return crc -# ── Telemetry ──────────────────────────────────────────────────────── +# ── STM32 -> PC broadcast ──────────────────────────────────────────── +# A single fixed 114-byte struct streamed continuously (circular DMA). No framing +# length, no CRC. Frames are delimited by the 4-byte magic. Integrity = REPETITION +# (each `counter` value is re-sent back-to-back, so a corrupt/torn copy differs +# from its neighbours) + UART odd parity. See debug_protocol.h BroadcastFrame. + +BCAST_MAGIC = 0xAA55AA55 +BROADCAST_FMT = " human label (matches the firmware FLAG_* defines) +FLAG_NAMES = { + 0: "STARTUP: waiting (Vfly guard)", + 1: "PRECHARGE TIMEOUT", + 2: "GUARD: VIN_MAX", + 3: "GUARD: VIN list: + return [name for bit, name in FLAG_NAMES.items() if flags & (1 << bit)] + + +def decode_param_bits(param_type: int, raw: int) -> float: + """Interpret a raw 32-bit param_value from the broadcast per its type.""" + if param_type == PTYPE_FLOAT: + return struct.unpack(" int: # back-compat for code that used t.seq + return self.counter & 0xFFFF + @property def vin_V(self) -> float: return self.vin / 1000.0 @@ -103,39 +187,91 @@ class Telemetry: def iout_A(self) -> float: return self.iout / 1000.0 + @property + def vfly_V(self) -> float: + return self.vfly / 1000.0 + @property def power_in_W(self) -> float: return self.vin * (-self.iin) / 1e6 @property def power_out_W(self) -> float: - return self.vout * self.iout / 1e6 + # iout_slow is the accurate output current (fast iout is protection-only) + return self.vout * self.iout_slow / 1e6 + + @property + def power_sys_W(self) -> float: + return self.vout * self.sys_current_ma / 1e6 @property def efficiency(self) -> float: + """Gross efficiency P_out/P_in (%).""" p_in = self.power_in_W return (self.power_out_W / p_in * 100.0) if p_in > 0.1 else 0.0 - -_TELEM_FMT = "<7f hHhH 6f 3f BxH h" # 78 bytes -_TELEM_SIZE = struct.calcsize(_TELEM_FMT) + @property + def efficiency_net(self) -> float: + """Net efficiency (P_out - P_sys)/P_in (%) — self-supply subtracted.""" + p_in = self.power_in_W + if p_in <= 0.1: + return 0.0 + return (self.power_out_W - self.power_sys_W) / p_in * 100.0 -def _decode_telemetry(payload: bytes) -> Optional[Telemetry]: - if len(payload) < _TELEM_SIZE: - return None - v = struct.unpack(_TELEM_FMT, payload[:_TELEM_SIZE]) - return Telemetry( - vin=v[0], vout=v[1], iin=v[2], iout=v[3], vfly=v[4], etemp=v[5], btemp=v[6], - last_tmp=v[7], VREF=v[8], vfly_correction=v[9], cmp_outer=v[10], - vfly_integral=v[11], vfly_avg_debug=v[12], - cc_output_f=v[13], mppt_iref=v[14], - mppt_last_vin=v[15], mppt_last_iin=v[16], - p_in=v[17], p_out=v[18], iout_slow=v[19], - seq=v[20], cmp_inner=v[21], vfly_ofs_applied=v[22], +# Back-compat alias: tuner/cli were written against the old Telemetry class. +Telemetry = BroadcastData + + +def decode_broadcast(frame: bytes) -> BroadcastData: + v = struct.unpack(BROADCAST_FMT, frame) + return BroadcastData( + counter=v[1], + vin=v[2], vout=v[3], iin=v[4], iout=v[5], vfly=v[6], etemp=v[7], btemp=v[8], + vfly_integral=v[9], vfly_avg_debug=v[10], cc_output_f=v[11], + mppt_iref=v[12], mppt_last_vin=v[13], mppt_last_iin=v[14], + p_in=v[15], p_out=v[16], iout_slow=v[17], + last_tmp=v[18], VREF=v[19], vfly_correction=v[20], + cmp_outer=v[21], cmp_inner=v[22], vfly_ofs_applied=v[23], + ctrl_mode=v[24], vfly_active=v[25], + status_flags=v[26], fmac_sr=v[27], fault_pc=v[28], cfsr=v[29], + param_id=v[30], param_type=v[31], param_value=v[33], pong=v[34], + sys_current_ma=v[35], iin_avg_ma=v[32], ) +class BroadcastParser: + """Parse the continuous 114-byte broadcast stream. Resync on the 4-byte magic; + validate a frame by REPETITION -- accept only when two consecutive byte-identical + copies arrive (a corrupt/torn copy differs from its neighbours -> discarded). No + CRC. Re-sends of the same `counter` are normal; dedup on counter downstream.""" + + def __init__(self): + self.buf = bytearray() + self._prev = None # previous raw frame awaiting a matching repeat + + def feed(self, data: bytes): + self.buf += data + while True: + i = self.buf.find(MAGIC_BYTES) + if i < 0: + # no magic yet; keep only a trailing partial-magic (3 bytes) + if len(self.buf) > 3: + del self.buf[:-3] + return + if i > 0: + del self.buf[:i] # drop junk / dropped-byte shift before magic + if len(self.buf) < BROADCAST_SIZE: + return # wait for a full frame + frame = bytes(self.buf[:BROADCAST_SIZE]) + del self.buf[:BROADCAST_SIZE] + if frame == self._prev: + self._prev = None # two identical copies -> accept, need a fresh pair next + yield decode_broadcast(frame) + else: + self._prev = frame # first sighting / differs -> hold, wait for the repeat + + # ── Parameter definitions ──────────────────────────────────────────── @dataclass @@ -196,22 +332,25 @@ PARAMS = [ ParamDef(0x73, "dither_anear", PTYPE_UINT16, "Dither", 716, 6442, ".0f"), ParamDef(0x74, "dither_afar", PTYPE_UINT16, "Dither", 716, 6442, ".0f"), ParamDef(0x75, "dither_dzero", PTYPE_UINT16, "Dither", 716, 6442, ".0f"), + # ADC calibration + ParamDef(0x79, "adc4_trig_phase", PTYPE_UINT16, "ADC", 3, 14313, ".0f"), # HRTIM master CMP3: iout_slow sample instant + ParamDef(0x7A, "iin_zero_sum", PTYPE_UINT16, "ADC", 0, 32760, ".0f"), # IIN software zero offset, sum-of-8 counts ] PARAM_BY_ID: dict[int, ParamDef] = {p.id: p for p in PARAMS} PARAM_BY_NAME: dict[str, ParamDef] = {p.name: p for p in PARAMS} -# ── Frame building ─────────────────────────────────────────────────── +# ── Frame building (PC -> STM32 commands, CRC-16 framed) ───────────── -def _build_frame(cmd: int, payload: bytes = b"") -> bytes: +def build_frame(cmd: int, payload: bytes = b"") -> bytes: header = bytes([SYNC_BYTE, cmd, len(payload)]) frame = header + payload crc = crc16(frame) return frame + bytes([(crc >> 8) & 0xFF, crc & 0xFF]) # big-endian: hi, lo -def _build_param_write(param_id: int, ptype: int, value) -> bytes: +def build_param_write(param_id: int, ptype: int, value) -> bytes: if ptype == PTYPE_FLOAT: val_bytes = struct.pack(" bytes: else: val_bytes = struct.pack(" Optional[tuple[int, float]]: - if len(payload) < 8: - return None - param_id, ptype = payload[0], payload[1] - vb = payload[4:8] - if ptype == PTYPE_FLOAT: - value = struct.unpack(" bytes: + return build_frame(CMD_PING) -# ── Frame parser state machine ─────────────────────────────────────── +def build_shutdown() -> bytes: + return build_frame(CMD_SHUTDOWN) -class _FrameParser: - WAIT_SYNC = 0 - WAIT_CMD = 1 - WAIT_LEN = 2 - WAIT_PAYLOAD = 3 - WAIT_CRC_HI = 4 - WAIT_CRC_LO = 5 - def __init__(self): - self.state = self.WAIT_SYNC - self.cmd = 0 - self.length = 0 - self.buf = bytearray() - self.payload = bytearray() - self.idx = 0 - self.crc_hi = 0 +def build_reset() -> bytes: + return build_frame(CMD_RESET) - def feed(self, data: bytes): - for b in data: - if self.state == self.WAIT_SYNC: - if b == SYNC_BYTE: - self.buf = bytearray([b]) - self.state = self.WAIT_CMD - elif self.state == self.WAIT_CMD: - self.cmd = b - self.buf.append(b) - self.state = self.WAIT_LEN - elif self.state == self.WAIT_LEN: - self.length = b - self.buf.append(b) - self.payload = bytearray() - self.idx = 0 - if b == 0: - self.state = self.WAIT_CRC_HI - elif b > 128: - self.state = self.WAIT_SYNC - else: - self.state = self.WAIT_PAYLOAD - elif self.state == self.WAIT_PAYLOAD: - self.payload.append(b) - self.buf.append(b) - self.idx += 1 - if self.idx >= self.length: - self.state = self.WAIT_CRC_HI - elif self.state == self.WAIT_CRC_HI: - self.crc_hi = b - self.state = self.WAIT_CRC_LO - elif self.state == self.WAIT_CRC_LO: - received = (self.crc_hi << 8) | b - expected = crc16(bytes(self.buf)) - self.state = self.WAIT_SYNC - if received == expected: - yield (self.cmd, bytes(self.payload)) + +def build_test_50() -> bytes: + return build_frame(CMD_TEST_50) + + +def build_relay_on() -> bytes: + return build_frame(CMD_RELAY_ON) + + +def build_relay_off() -> bytes: + return build_frame(CMD_RELAY_OFF) + + +def build_hold_converter() -> bytes: + return build_frame(CMD_HOLD_CONVERTER) + + +def build_toggle_precharge() -> bytes: + return build_frame(CMD_TOGGLE_PRECHARGE) + + +def build_param_read_all() -> bytes: + return build_frame(CMD_PARAM_READ_ALL) + + +def build_clear_flags() -> bytes: + return build_frame(CMD_CLEAR_FLAGS) # ── STM32Link — synchronous serial interface ───────────────────────── class STM32Link: - """Blocking serial link to STM32 debug protocol. + """Blocking serial link to the STM32 debug protocol. + + Telemetry and parameter echoes arrive via the continuous broadcast; + commands go out CRC-16 framed. The port MUST be opened with odd + parity (8-O-1) or the broadcast never validates. Usage:: @@ -321,9 +427,17 @@ class STM32Link: """ def __init__(self, port: str, baudrate: int = 460800, timeout: float = 2.0): - self.ser = serial.Serial(port, baudrate, timeout=timeout) - self._parser = _FrameParser() - self._param_cache: dict[int, float] = {} + self.timeout = timeout + self.ser = serial.Serial( + port, baudrate, timeout=0.1, + bytesize=serial.EIGHTBITS, + parity=serial.PARITY_ODD, + stopbits=serial.STOPBITS_ONE, + ) + self._parser = BroadcastParser() + self._params: dict[int, float] = {} + self._pending: deque = deque() # decoded frames not yet consumed + self._last_counter = -1 def close(self): if self.ser and self.ser.is_open: @@ -340,92 +454,95 @@ class STM32Link: def _send(self, frame: bytes): self.ser.write(frame) - def _recv_frames(self, timeout: float = 1.0) -> list[tuple[int, bytes]]: - """Read available data and return decoded frames.""" - frames = [] - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - data = self.ser.read(self.ser.in_waiting or 1) - if data: - for cmd, payload in self._parser.feed(data): - frames.append((cmd, payload)) - if frames: - # Drain any remaining data - time.sleep(0.02) - data = self.ser.read(self.ser.in_waiting) - if data: - for cmd, payload in self._parser.feed(data): - frames.append((cmd, payload)) - break - return frames + def _drain_serial(self) -> None: + """Read one serial chunk and queue ALL decoded frames. - def _wait_for(self, target_cmd: int, timeout: float = 2.0) -> Optional[bytes]: - """Wait for a specific command response, processing others.""" - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - remaining = deadline - time.monotonic() - if remaining <= 0: - break - data = self.ser.read(self.ser.in_waiting or 1) - if data: - for cmd, payload in self._parser.feed(data): - if cmd == target_cmd: - return payload - # Cache param values seen in passing - if cmd in (CMD_PARAM_VALUE, CMD_PARAM_WRITE_ACK): - result = _decode_param_value(payload) - if result: - self._param_cache[result[0]] = result[1] - # Cache telemetry too - if cmd == CMD_TELEMETRY: - self._last_telemetry = _decode_telemetry(payload) - return None + Param round-robin echoes are stashed into self._params in passing + (only once counter > 0 — fields are stale during the post-reset window). + """ + data = self.ser.read(4096) + if not data: + return + for b in self._parser.feed(data): + if b.counter > 0 and b.param_id in PARAM_BY_ID: + self._params[b.param_id] = decode_param_bits(b.param_type, b.param_value) + self._pending.append(b) + + def _pump(self, deadline: float) -> Iterator[BroadcastData]: + """Yield validated frames until deadline. + + Frames are staged through self._pending so nothing is lost when a + caller stops iterating early (generator abandoned mid-chunk). + """ + while True: + while self._pending: + yield self._pending.popleft() + if time.monotonic() >= deadline: + return + self._drain_serial() # ── Commands ───────────────────────────────────────────────────── def ping(self, timeout: float = 2.0) -> bool: - """Send PING, return True if PONG received.""" - self._send(_build_frame(CMD_PING)) - return self._wait_for(CMD_PONG, timeout) is not None + """Send PING, return True when the broadcast pong counter increments.""" + baseline: Optional[int] = None + for b in self._pump(time.monotonic() + min(0.7, timeout)): + baseline = b.pong # pong is valid even at counter == 0 + break + if baseline is None: + return False # no broadcast at all -> not connected + self._send(build_ping()) + for b in self._pump(time.monotonic() + timeout): + if b.pong != baseline: + return True + return False def shutdown(self): """Command the converter off.""" - self._send(_build_frame(CMD_SHUTDOWN)) + self._send(build_shutdown()) def reset(self): """Command a system reset.""" - self._send(_build_frame(CMD_RESET)) + self._send(build_reset()) def test_50(self): """Enter 50% duty test mode.""" - self._send(_build_frame(CMD_TEST_50)) + self._send(build_test_50()) def relay_on(self): """Latch the input relay closed (bench test).""" - self._send(_build_frame(CMD_RELAY_ON)) + self._send(build_relay_on()) def relay_off(self): """Latch the input relay open (bench test).""" - self._send(_build_frame(CMD_RELAY_OFF)) + self._send(build_relay_off()) def hold_converter(self): """Toggle 'hold converter off' (boot guard + disarm trips).""" - self._send(_build_frame(CMD_HOLD_CONVERTER)) + self._send(build_hold_converter()) def toggle_precharge(self): """Toggle the precharge FET (bench test).""" - self._send(_build_frame(CMD_TOGGLE_PRECHARGE)) + self._send(build_toggle_precharge()) - def read_telemetry(self, timeout: float = 2.0) -> Optional[Telemetry]: - """Wait for next telemetry packet.""" - payload = self._wait_for(CMD_TELEMETRY, timeout) - if payload: - return _decode_telemetry(payload) + def clear_flags(self): + """Clear latched status flags.""" + self._send(build_clear_flags()) + + # ── Telemetry ──────────────────────────────────────────────────── + + def read_telemetry(self, timeout: float = 2.0) -> Optional[BroadcastData]: + """Return the next FRESH publish (counter-deduped — repeats skipped).""" + for b in self._pump(time.monotonic() + timeout): + if b.counter > 0 and b.counter != self._last_counter: + self._last_counter = b.counter + return b return None - def read_telemetry_avg(self, n: int = 10, timeout: float = 5.0) -> Optional[Telemetry]: - """Read n telemetry packets and return the average.""" - samples: list[Telemetry] = [] + def read_telemetry_avg(self, n: int = 10, timeout: float = 5.0) -> Optional[BroadcastData]: + """Average n fresh publishes (~n/100 s). Float fields are averaged; + int/flag fields come from the last sample.""" + samples: list[BroadcastData] = [] deadline = time.monotonic() + timeout while len(samples) < n and time.monotonic() < deadline: t = self.read_telemetry(timeout=deadline - time.monotonic()) @@ -433,43 +550,38 @@ class STM32Link: samples.append(t) if not samples: return None - # Average all analog float fields - avg = Telemetry() + avg = dataclasses.replace(samples[-1]) for attr in ("vin", "vout", "iin", "iout", "vfly", "etemp", "btemp", - "vfly_integral", "vfly_avg_debug", "cc_output_f", - "mppt_iref", "mppt_last_vin", "mppt_last_iin", - "p_in", "p_out", "iout_slow"): + "vfly_integral", "vfly_avg_debug", "cc_output_f", + "mppt_iref", "mppt_last_vin", "mppt_last_iin", + "p_in", "p_out", "iout_slow"): setattr(avg, attr, sum(getattr(s, attr) for s in samples) / len(samples)) - avg.seq = samples[-1].seq return avg - def request_all_params(self): - """Request all parameter values from the STM32.""" - self._send(_build_frame(CMD_PARAM_READ_ALL)) + # ── Parameters ─────────────────────────────────────────────────── - def read_all_params(self, timeout: float = 3.0) -> dict[str, float]: - """Request and collect all parameter values.""" - self._param_cache.clear() + def request_all_params(self): + """Restart the broadcast param round-robin from the first param.""" + self._send(build_param_read_all()) + + def read_all_params(self, timeout: float = 5.0) -> dict[str, float]: + """Collect all parameter values from the broadcast round-robin + (one param per publish -> full rotation ~0.4 s at 100 Hz).""" + self._params.clear() self.request_all_params() - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - data = self.ser.read(self.ser.in_waiting or 1) - if data: - for cmd, payload in self._parser.feed(data): - if cmd == CMD_PARAM_VALUE: - result = _decode_param_value(payload) - if result: - self._param_cache[result[0]] = result[1] - time.sleep(0.05) - # Convert to name->value + for _ in self._pump(time.monotonic() + timeout): + if len(self._params) >= len(PARAMS): + break return { PARAM_BY_ID[pid].name: val - for pid, val in self._param_cache.items() + for pid, val in self._params.items() if pid in PARAM_BY_ID } def write_param(self, name: str, value: float, wait_ack: bool = True) -> bool: - """Write a parameter by name. Returns True if ACK received.""" + """Write a parameter by name. The write is acknowledged when the + broadcast round-robin echoes the new value back (clamped + formatted + the way the firmware reports it); retries until it matches.""" pdef = PARAM_BY_NAME.get(name) if not pdef: raise ValueError(f"Unknown parameter: {name!r}") @@ -477,17 +589,22 @@ class STM32Link: raise ValueError( f"{name}: {value} out of range [{pdef.min_val}, {pdef.max_val}]" ) - frame = _build_param_write(pdef.id, pdef.ptype, value) - self._send(frame) - if wait_ack: - payload = self._wait_for(CMD_PARAM_WRITE_ACK, timeout=2.0) - if payload: - result = _decode_param_value(payload) - if result: - self._param_cache[result[0]] = result[1] - return True - return False - return True + frame = build_param_write(pdef.id, pdef.ptype, value) + if not wait_ack: + self._send(frame) + return True + clamped = max(pdef.min_val, min(pdef.max_val, float(value))) + expect = f"{clamped:{pdef.fmt}}" + for _ in range(6): + self._send(frame) + # one full round-robin rotation is ~0.4 s; 0.5 s sees the echo once + for b in self._pump(time.monotonic() + 0.5): + if b.counter > 0 and b.param_id == pdef.id: + echoed = decode_param_bits(b.param_type, b.param_value) + if f"{echoed:{pdef.fmt}}" == expect: + self._params[pdef.id] = echoed + return True + return False def write_param_by_id(self, param_id: int, value: float) -> bool: """Write a parameter by ID.""" diff --git a/testbench/tuner.py b/testbench/tuner.py index 883ee7a..8411af4 100644 --- a/testbench/tuner.py +++ b/testbench/tuner.py @@ -61,7 +61,7 @@ class Tuner: self, bench: MPPTTestbench, link: STM32Link, - settle_time: float = 3.0, + settle_time: float = 5.0, stm_avg_samples: int = 10, ): self.bench = bench