#!/usr/bin/env python3 # -*- coding: utf-8 -*- # ESP32 Logger GUI con ttkbootstrap (tema flatly) # Gráficos de X/Y y C/P usando hora del log en eje X # Fine-Tune Parameters en una sola fila # Fix: cierre limpio sin reiniciar kernel en Spyder import threading, queue, re, os, requests from datetime import datetime from tkinter import Text, Scrollbar, END from tkinter import filedialog from websocket import WebSocketApp import matplotlib matplotlib.use("TkAgg") from matplotlib.figure import Figure from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg import matplotlib.dates as mdates import ttkbootstrap as tb from ttkbootstrap.constants import * def parse_xy(line: str): pats = [ r'X\s*[:=]\s*(-?\d+).*?Y\s*[:=]\s*(-?\d+)', r'POS:\s*(-?\d+):\s*(-?\d+)' ] for p in pats: m = re.search(p, line) if m: try: return int(m.group(1)), int(m.group(2)) except: pass return None, None def parse_cp(line: str): m = re.search(r'C\s*[:=]\s*(\d+).*?P\s*[:=]\s*(\d+)', line) if m: try: return int(m.group(1)), int(m.group(2)) except: pass return None, None def parse_time(line: str): m = re.search(r'(\d{2}:\d{2}:\d{2})', line) if m: try: return datetime.strptime(m.group(1), "%H:%M:%S") except: return None return None class App: def __init__(self, root: tb.Window): self.root = root root.title("ESP32 Logger GUI") root.geometry("1400x850") self.ip = tb.StringVar(value="192.168.137.55") self.base = None self.ws = None self.ws_thread = None self.log_q = queue.Queue() self.log_to_file = tb.IntVar(value=1) self.log_path = os.path.abspath(f"esp32_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt") self.fh = None self.t, self.xs, self.ys = [], [], [] self.tc, self.cs, self.ps = [], [], [] self.last_xy = tb.StringVar(value="Último X,Y: (—, —)") self.last_c = tb.StringVar(value="Último C: — mA") self.last_p = tb.StringVar(value="Último P: — W/m²") self.param_entries = {} self.running = True # flag para control de cierre self._ui() self._tick_log() self._tick_plot() def _ui(self): top = tb.Frame(self.root); top.pack(fill="x", padx=8, pady=6) tb.Label(top, text="ESP32 IP:").pack(side="left") tb.Entry(top, textvariable=self.ip, width=18).pack(side="left", padx=5) tb.Button(top, text="Connect", command=self.connect, bootstyle="success").pack(side="left", padx=3) tb.Button(top, text="Disconnect", command=self.disconnect, bootstyle="danger").pack(side="left", padx=3) tb.Checkbutton(top, text="Save log to file", variable=self.log_to_file, bootstyle="check").pack(side="left", padx=10) tb.Button(top, text="Pick log file...", command=self._pick_file, bootstyle="info").pack(side="left", padx=3) tb.Button(top, text="Download Log", command=self.download_log, bootstyle="secondary").pack(side="right", padx=3) # Command buttons cmds = tb.Frame(self.root); cmds.pack(fill="x", padx=8, pady=6) tb.Button(cmds, text="Track Sun", command=lambda: self.send_cmd(2), bootstyle="primary").pack(side="left", padx=5) tb.Button(cmds, text="Fine-Tune", command=lambda: self.send_cmd(3), bootstyle="warning").pack(side="left", padx=5) tb.Button(cmds, text="Set XY (Homing)", command=lambda: self.send_cmd(4), bootstyle="danger").pack(side="left", padx=5) tb.Label(cmds, textvariable=self.last_xy, bootstyle="dark").pack(side="left", padx=12) tb.Label(cmds, textvariable=self.last_c, bootstyle="dark").pack(side="left", padx=12) tb.Label(cmds, textvariable=self.last_p, bootstyle="dark").pack(side="left", padx=12) # Fine-Tune parameters en una fila tun = tb.Frame(self.root); tun.pack(fill="x", padx=8, pady=6) tb.Label(tun, text="Fine-Tuning Parameters:").pack(side="left") params = [ ("LPF Beta", "setLPFBeta", "beta"), ("Error Max", "setErrMax", "err"), ("Delay (s)", "setDelay", "delay"), ("Min Err Max", "setMinErrMax", "minErr"), ("Abrupt Thres", "setAbruptThreshold", "abrupt"), ("Small Thres", "setSmallThreshold", "small"), ("Reset Thres", "setResetThreshold", "reset"), ("Peak Prox", "setPeakProximity", "peak"), ("Cycle Max", "setCicleMax", "cicle"), ] for (label, endpoint, arg) in params: tb.Label(tun, text=label+":").pack(side="left", padx=3) var = tb.StringVar(value="") tb.Entry(tun, textvariable=var, width=6).pack(side="left", padx=2) tb.Button(tun, text="Set", command=lambda ep=endpoint, a=arg, v=var: self.set_param(ep, a, v.get()), bootstyle="secondary-outline").pack(side="left", padx=2) self.param_entries[endpoint] = var # Notebook (tabs) notebook = tb.Notebook(self.root, bootstyle="primary") notebook.pack(fill="both", expand=True, padx=8, pady=6) # --- Tab X/Y --- tab_xy = tb.Frame(notebook) notebook.add(tab_xy, text="X / Y") fig1 = Figure(figsize=(6.6, 3.6), dpi=125) self.ax1 = fig1.add_subplot(111) self.ax1.set_title("posX/posY vs hora"); self.ax1.set_xlabel("Hora"); self.ax1.set_ylabel("pasos") self.ax1.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S")) self.lx, = self.ax1.plot([], [], label="X"); self.ly, = self.ax1.plot([], [], label="Y") self.ax1.legend() self.canvas1 = FigureCanvasTkAgg(fig1, master=tab_xy); self.canvas1.draw() self.canvas1.get_tk_widget().pack(side="left", fill="both", expand=True) # --- Tab C/P --- tab_cp = tb.Frame(notebook) notebook.add(tab_cp, text="C / P") fig2 = Figure(figsize=(6.6, 3.6), dpi=125) self.ax2 = fig2.add_subplot(111) self.ax2.set_title("Corriente (mA) e Irradiancia (W/m²) vs hora") self.ax2.set_xlabel("Hora"); self.ax2.set_ylabel("Corriente [mA]") self.ax2.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S")) self.ax2r = self.ax2.twinx() self.ax2r.set_ylabel("Irradiancia [W/m²]") self.lc, = self.ax2.plot([], [], label="C (mA)") self.lp, = self.ax2r.plot([], [], 'r', label="P (W/m²)") self.ax2.legend(loc="upper left"); self.ax2r.legend(loc="upper right") self.canvas2 = FigureCanvasTkAgg(fig2, master=tab_cp); self.canvas2.draw() self.canvas2.get_tk_widget().pack(side="left", fill="both", expand=True) # Log panel right = tb.Frame(self.root); right.pack(fill="both", expand=True, padx=(8,0)) tb.Label(right, text="Log (local)").pack(anchor="w") self.text = Text(right, wrap="none"); yscroll = Scrollbar(right, command=self.text.yview) self.text.configure(yscrollcommand=yscroll.set); self.text.pack(side="left", fill="both", expand=True) yscroll.pack(side="left", fill="y") # ============ WebSocket ============ def connect(self): ip = self.ip.get().strip() if not ip: return self.base = f"http://{ip}" ws_url = f"ws://{ip}:81/ws" self.ws = WebSocketApp(ws_url, on_open=lambda ws: self._log("WS abierto"), on_message=self._ws_msg, on_error=lambda ws,err: self._log(f"WS error: {err}"), on_close=lambda ws,*a: self._log("WS cerrado")) self.ws_thread = threading.Thread(target=self.ws.run_forever, daemon=True) self.ws_thread.start() self._log(f"Conectando a {ws_url} ...") def disconnect(self): if self.ws: try: self.ws.close() except: pass self._log("Desconectado") def _ws_msg(self, ws, message): for line in str(message).splitlines(): self._log(line) t = parse_time(line) if not t: continue # --- Parse X/Y --- x, y = parse_xy(line) if x is not None and y is not None: self.t.append(t); self.xs.append(x); self.ys.append(y) self.last_xy.set(f"Último X,Y: ({x}, {y})") # --- Parse C/P --- c, p = parse_cp(line) if c is not None and p is not None: self.tc.append(t); self.cs.append(c/6144); self.ps.append(p/8192) self.last_c.set(f"Último C: {self.cs[-1]:.1f} mA") self.last_p.set(f"Último P: {self.ps[-1]:.1f} W/m²") # Guardar todo excepto POS: if not line.startswith("POS:") and self.log_to_file.get(): self._write_file(line + "\n") # ============ HTTP ============ def send_cmd(self, cmd): try: r = requests.get(self.base+"/command?status=" + str(cmd), timeout=10) self._log(f"[HTTP {r.status_code}] /command?status={cmd}") except Exception as e: self._log(f"HTTP error: {e}") def set_param(self, endpoint, argname, value): try: r = requests.get(self.base+"/"+endpoint, params={argname:value}, timeout=5) self._log(f"[HTTP {r.status_code}] /{endpoint}?{argname}={value}") except Exception as e: self._log(f"HTTP error: {e}") def download_log(self): path = filedialog.asksaveasfilename(defaultextension=".txt", filetypes=[("Text","*.txt")]) if not path: return try: r = requests.get(self.base+"/download", stream=True, timeout=20) if r.ok: with open(path, "wb") as f: for chunk in r.iter_content(4096): f.write(chunk) self._log(f"Log descargado en {path}") else: self._log(f"Download error: {r.status_code}") except Exception as e: self._log(f"HTTP error: {e}") # ============ Utils ============ def _log(self, s): self.log_q.put(s) def _tick_log(self): try: while True: s = self.log_q.get_nowait() self.text.insert(END, s+"\n"); self.text.see(END) except queue.Empty: pass if self.running: self.root.after(100, self._tick_log) def _tick_plot(self): N=20000 # XY if len(self.t)>N: self.t,self.xs,self.ys=self.t[-N:],self.xs[-N:],self.ys[-N:] self.lx.set_data(self.t,self.xs); self.ly.set_data(self.t,self.ys) if self.t: self.ax1.set_xlim(self.t[0], self.t[-1]) if self.xs or self.ys: vals=self.xs+self.ys; vmin,vmax=min(vals),max(vals) if vmin==vmax: vmin-=1; vmax+=1 self.ax1.set_ylim(vmin,vmax) self.canvas1.draw_idle() # CP if len(self.tc)>N: self.tc,self.cs,self.ps=self.tc[-N:],self.cs[-N:],self.ps[-N:] self.lc.set_data(self.tc,self.cs); self.lp.set_data(self.tc,self.ps) if self.tc: self.ax2.set_xlim(self.tc[0], self.tc[-1]) if self.cs: vmin,vmax=min(self.cs),max(self.cs) if vmin==vmax: vmin-=10; vmax+=10 self.ax2.set_ylim(vmin,vmax) if self.ps: vmin,vmax=min(self.ps),max(self.ps) if vmin==vmax: vmin-=1; vmax+=1 self.ax2r.set_ylim(vmin,vmax) self.canvas2.draw_idle() if self.running: self.root.after(500, self._tick_plot) def _toggle_file(self): if self.log_to_file.get() and not self.fh: try: self.fh=open(self.log_path,"a",encoding="utf-8") self._log(f"Guardando en {self.log_path}") except: self.log_to_file.set(0) elif not self.log_to_file.get(): self._close_file() def _pick_file(self): path=filedialog.asksaveasfilename(defaultextension=".txt") if path: self.log_path=path if self.log_to_file.get(): self._close_file(); self._toggle_file() def _write_file(self,txt): try: if self.fh: self.fh.write(txt) except: pass def _close_file(self): if self.fh: try: self.fh.flush(); self.fh.close() except: pass self.fh=None def on_close(self): self.running = False self.disconnect() self._close_file() self.root.destroy() self.root.quit() def main(): root = tb.Window(themename="sandstone") # tema minimalista claro app = App(root) root.protocol("WM_DELETE_WINDOW", app.on_close) root.mainloop() if __name__=="__main__": main()