321 lines
13 KiB
Python
321 lines
13 KiB
Python
#!/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()
|