#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ESP32 Logger GUI final - Botones: Track Sun, Fine-Tune, Set XY (Homing) - Parámetros de Fine-Tuning con los nombres exactos del firmware - Guarda todas las líneas LOG: (con X,Y,...), filtra solo POS: - Gráfica en vivo de X/Y """ import threading, time, queue, re, os, requests from datetime import datetime from tkinter import * from tkinter import filedialog, messagebox from websocket import WebSocketApp import matplotlib matplotlib.use("TkAgg") from matplotlib.figure import Figure from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg def ts(): return datetime.now().strftime("%Y-%m-%d %H:%M:%S") 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 class App: def __init__(self, root: Tk): self.root = root root.title("ESP32 Logger GUI") root.geometry("1100x750") self.ip = 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 = 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.t0 = time.time() self.t, self.xs, self.ys = [], [], [] self.last_xy = StringVar(value="Último X,Y: (—, —)") self._ui() self._tick_log() self._tick_plot() def _ui(self): top = Frame(self.root); top.pack(fill="x", padx=8, pady=6) Label(top, text="ESP32 IP:").pack(side="left") Entry(top, textvariable=self.ip, width=18).pack(side="left", padx=5) Button(top, text="Connect", command=self.connect).pack(side="left", padx=3) Button(top, text="Disconnect", command=self.disconnect).pack(side="left", padx=3) Checkbutton(top, text="Save log to file", variable=self.log_to_file, command=self._toggle_file).pack(side="left", padx=10) Button(top, text="Pick log file...", command=self._pick_file).pack(side="left", padx=3) Button(top, text="Download Log", command=self.download_log).pack(side="right", padx=3) # Command buttons cmds = Frame(self.root); cmds.pack(fill="x", padx=8, pady=6) Button(cmds, text="Track Sun", command=lambda: self.send_cmd(2)).pack(side="left", padx=5) Button(cmds, text="Fine-Tune", command=lambda: self.send_cmd(3)).pack(side="left", padx=5) Button(cmds, text="Set XY (Homing)", command=lambda: self.send_cmd(4)).pack(side="left", padx=5) Label(cmds, textvariable=self.last_xy).pack(side="left", padx=12) # Fine-Tuning parameters tun = Frame(self.root); tun.pack(fill="x", padx=8, pady=6) Label(tun, text="Fine-Tuning Parameters").grid(row=0, column=0, sticky="w") self.param_entries = {} # cada tupla: (label, endpoint, argname) params = [ ("LPF Beta", "setLPFBeta", "beta"), ("Error Max", "setErrMax", "err"), ("Delay (s)", "setDelay", "delay"), ("Min Err Max", "setMinErrMax", "minErr"), ("Abrupt Threshold", "setAbruptThreshold", "abrupt"), ("Small Threshold", "setSmallThreshold", "small"), ("Reset Threshold", "setResetThreshold", "reset"), ("Peak Proximity", "setPeakProximity", "peak"), ("Cycle Max", "setCicleMax", "cicle"), ] for r, (label, endpoint, arg) in enumerate(params, start=1): Label(tun, text=label+":").grid(row=r, column=0, sticky="e") var = StringVar(value="") Entry(tun, textvariable=var, width=10).grid(row=r, column=1, padx=3) Button(tun, text="Set", command=lambda ep=endpoint, a=arg, v=var: self.set_param(ep, a, v.get())).grid(row=r, column=2, padx=3) self.param_entries[endpoint] = var # Plot + log mid = Frame(self.root); mid.pack(fill="both", expand=True, padx=8, pady=6) fig = Figure(figsize=(6.6, 3.6), dpi=125) self.ax = fig.add_subplot(111) self.ax.set_title("posX/posY vs tiempo"); self.ax.set_xlabel("s"); self.ax.set_ylabel("pasos") self.lx, = self.ax.plot([], [], label="X"); self.ly, = self.ax.plot([], [], label="Y") self.ax.legend() self.canvas = FigureCanvasTkAgg(fig, master=mid); self.canvas.draw() self.canvas.get_tk_widget().pack(side="left", fill="both", expand=True) right = Frame(mid); right.pack(side="left", fill="both", expand=True, padx=(8,0)) 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) x, y = parse_xy(line) if x is not None and y is not None: t = time.time() - self.t0 self.t.append(t); self.xs.append(x); self.ys.append(y) self.last_xy.set(f"Último X,Y: ({x}, {y})") # 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), params=dict(cmd=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 self.root.after(100, self._tick_log) def _tick_plot(self): N=20000 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.ax.set_xlim(max(0,self.t[0]), max(10,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.ax.set_ylim(vmin,vmax) self.canvas.draw_idle() 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 main(): root=Tk(); app=App(root) root.protocol("WM_DELETE_WINDOW", lambda: (app.disconnect(), app._close_file(), root.destroy())) root.mainloop() if __name__=="__main__": main()