234 lines
9.5 KiB
Python
234 lines
9.5 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
ESP32 Logger GUI ajustada:
|
|
- Botones: Track Sun, Fine-Tune, Set XY (Homing)
|
|
- Parámetros de Fine-Tuning visibles (como en la web), pero solo activos los soportados por firmware actual:
|
|
- setErrMax
|
|
- setMinErrMax
|
|
- Los demás aparecen deshabilitados (grisados)
|
|
- Log + gráfica de X/Y (POS filtrado para archivo)
|
|
"""
|
|
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 ajustada")
|
|
root.geometry("1100x750")
|
|
|
|
self.ip = StringVar(value="192.168.4.1")
|
|
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 = {}
|
|
params = [
|
|
("LPF Beta", "setLPFBeta", False),
|
|
("Error Max", "setErrMax", True),
|
|
("Delay (s)", "setDelay", False),
|
|
("Min Error Max", "setMinErrMax", True),
|
|
("Abrupt Threshold", "setAbruptThreshold", False),
|
|
("Small Threshold", "setSmallThreshold", False),
|
|
("Reset Threshold", "setResetThreshold", False),
|
|
("Peak Proximity", "setPeakProximity", False),
|
|
("Cycle Max", "setCicleMax", False),
|
|
]
|
|
|
|
for r, (label, endpoint, active) in enumerate(params, start=1):
|
|
Label(tun, text=label+":").grid(row=r, column=0, sticky="e")
|
|
var = StringVar(value="")
|
|
ent = Entry(tun, textvariable=var, width=10, state="normal" if active else "disabled")
|
|
ent.grid(row=r, column=1, padx=3)
|
|
btn = Button(tun, text="Set", command=lambda ep=endpoint, v=var: self.set_param(ep, v.get()),
|
|
state="normal" if active else "disabled")
|
|
btn.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=100)
|
|
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})")
|
|
continue
|
|
if self.log_to_file.get():
|
|
self._write_file(line + "\n")
|
|
|
|
# HTTP
|
|
def send_cmd(self, cmd):
|
|
try:
|
|
r = requests.get(self.base+"/sendCommand", params=dict(cmd=cmd), timeout=5)
|
|
self._log(f"[HTTP {r.status_code}] /sendCommand?cmd={cmd}")
|
|
except Exception as e:
|
|
self._log(f"HTTP error: {e}")
|
|
|
|
def set_param(self, endpoint, value):
|
|
try:
|
|
r = requests.get(self.base+"/"+endpoint, params=dict(val=value), timeout=5)
|
|
self._log(f"[HTTP {r.status_code}] /{endpoint}?val={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=2000
|
|
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()
|