"""
telemetry_gui_equinox_full_with_samples.py
EQUINOX GUI including sample product image & data icons.
Toggle theme fixes. You can replace sample icons later.
Dependencies:
pip install customtkinter pyserial matplotlib pillow
"""
import json
import threading
import queue
import time
from collections import deque
from datetime import datetime
import os
import customtkinter as ctk
import matplotlib
[Link]("TkAgg")
from [Link] import Figure
from [Link].backend_tkagg import FigureCanvasTkAgg
from PIL import Image # for icon/image handling
import serial
# ----------------- USER CONFIG / PATHS FOR SAMPLE ICONS -----------------
SERIAL_PORT = "COM7"
SERIAL_BAUD = 115200
SERIAL_TIMEOUT = 0.5
MAX_SAMPLES = 150
# Logo/product image sample (use these sample icons)
PRODUCT_IMAGE_SAMPLE_PATH = "icons/product_sample1.png"
WINDOW_ICON_PATH = "icons/badge_sample.png"
# Data icons mapping: key → sample icon path
ICON_PATHS = {
"dist_cm": "icons/data_icon_sample.png",
"mq": "icons/data_icon_sample.png",
"temp_c": "icons/data_icon_sample.png",
"hum": "icons/data_icon_sample.png",
"fix": "icons/data_icon_sample.png",
"sats": "icons/data_icon_sample.png",
# You can assign different sample icons for each key
CUSTOM_THEMES = [
# put paths to custom theme files here if any
# ----------------- Appearance / theme -----------------
ctk.set_appearance_mode("dark")
ctk.set_default_color_theme("dark-blue")
# ----------------- Telemetry & History -----------------
telemetry_queue = [Link](maxsize=500)
dist_history = deque(maxlen=MAX_SAMPLES)
mq_history = deque(maxlen=MAX_SAMPLES)
last_snapshot = {}
last_recv_time = None
# ----------------- Serial Reader Thread -----------------
class SerialReader([Link]):
def __init__(self, port, baud, timeout, q):
super().__init__(daemon=True)
[Link] = port
[Link] = baud
[Link] = timeout
[Link] = q
self._stop_event = [Link]()
[Link] = None
def stop(self):
"""Stop the thread and close serial port."""
self._stop_event.set()
try:
if [Link] and [Link].is_open:
[Link]()
except:
pass
def run(self):
"""Continuously read serial, parse JSON, enqueue."""
while not self._stop_event.is_set():
if [Link] is None:
try:
[Link] = [Link]([Link], [Link], timeout=[Link])
print(f"[serial] opened {[Link]} @ {[Link]}")
except Exception as e:
print(f"[serial] open fail: {e}, retry in 2s")
[Link](2)
continue
try:
raw = [Link]()
if not raw:
continue
text = [Link](errors='ignore').strip()
if not text:
continue
if not [Link]("{"):
idx = [Link]("{")
if idx != -1:
text = text[idx:]
else:
continue
try:
data = [Link](text)
except [Link]:
continue
try:
[Link].put_nowait(data)
except [Link]:
_ = [Link].get_nowait()
[Link].put_nowait(data)
except Exception as e:
print(f"[serial] read error: {e}")
try:
if [Link]:
[Link]()
except:
pass
[Link] = None
[Link](1)
# ----------------- Main GUI -----------------
class EquinoxGUI([Link]):
def __init__(self):
super().__init__() # root window
# Window icon using sample icon
if [Link](WINDOW_ICON_PATH):
try:
[Link](WINDOW_ICON_PATH)
except:
try:
icon_img = [Link](WINDOW_ICON_PATH)
icon_ctk = [Link](light_image=icon_img, dark_image=icon_img, size=(32,32))
[Link](False, icon_ctk._photo_image)
except:
pass
# Fonts
self.font_title = [Link](family="Helvetica", size=24, weight="bold")
self.font_label = [Link](family="Arial", size=12)
self.font_value = [Link](family="Consolas", size=14, weight="bold")
self.font_log = [Link](family="Courier", size=10)
# Colors
self.COLOR_BG_FRAME = "#2C2F33"
self.COLOR_BORDER = "#7289DA"
self.COLOR_ACCENT = "#00BFFF"
self.COLOR_TEXT = "#E0E0E0"
self.COLOR_PLOT_BG = "#1E2124"
# Window setup
[Link]("EQUINOX Telemetry Monitor")
[Link]("950x650")
[Link](width=True, height=True)
# Header frame: product image + title + toggle
header = [Link](self,
fg_color=self.COLOR_BG_FRAME,
corner_radius=8,
border_width=2,
border_color=self.COLOR_BORDER)
[Link](side="top", fill="x", padx=10, pady=(10,5))
# If sample product image exists, load and display
if [Link](PRODUCT_IMAGE_SAMPLE_PATH):
pil_prod = [Link](PRODUCT_IMAGE_SAMPLE_PATH)
prod_ctk = [Link](light_image=pil_prod,
dark_image=pil_prod,
size=(80,80))
prod_lab = [Link](header, image=prod_ctk, text="")
prod_lab.pack(side="left", padx=10, pady=5)
self.prod_img_ref = prod_ctk # keep reference
# Title
lbl_title = [Link](header,
text="EQUINOX",
font=self.font_title,
text_color=self.COLOR_ACCENT)
lbl_title.pack(side="left", padx=(10,0), pady=10)
# Toggle theme switch (fixed repeatedly)
self.theme_var = [Link](value=ctk.get_appearance_mode())
def on_toggle_theme():
current = self.theme_var.get()
new_mode = "light" if current == "dark" else "dark"
ctk.set_appearance_mode(new_mode)
self.theme_var.set(new_mode)
theme_switch = [Link](header,
text="Theme",
command=on_toggle_theme,
variable=self.theme_var,
onvalue="dark",
offvalue="light",
font=self.font_label,
fg_color=self.COLOR_BORDER,
progress_color=self.COLOR_ACCENT,
button_color=self.COLOR_ACCENT,
text_color=self.COLOR_TEXT,
border_width=1,
border_color=self.COLOR_ACCENT)
theme_switch.pack(side="right", padx=10, pady=10)
# Data panel: two columns, with data icons (using sample icon for all keys here)
data_frame = [Link](self,
fg_color=self.COLOR_BG_FRAME,
corner_radius=8,
border_width=1,
border_color=self.COLOR_BORDER)
data_frame.pack(side="top", fill="both", expand=True, padx=10, pady=5)
[Link] = {}
label_defs = [
("Distance (cm)", "dist_cm"),
("Air Quality (MQ-135)", "mq"),
("Temperature (°C)", "temp_c"),
("Humidity (%)", "hum"),
("GPS Fix", "fix"),
("Satellites", "sats"),
("Latitude", "lat"),
("Longitude", "lon"),
("Speed (cm/s)", "spd_cms"),
("Sequence", "seq"),
("Buzzer (D2)", "buzzer")
left = label_defs[:6]
right = label_defs[6:]
# helper to load sample icon for key
def load_icon_for(key, size=(24,24)):
path = ICON_PATHS.get(key)
if path and [Link](path):
pil = [Link](path)
return [Link](light_image=pil, dark_image=pil, size=size)
else:
return None
# Left column
for i, (text, key) in enumerate(left):
icon = load_icon_for(key)
if icon:
lbl_icon = [Link](data_frame, image=icon, text="")
lbl_icon.image = icon
lbl_icon.grid(row=i, column=0, padx=(10,2), pady=3, sticky="w")
col_text = 1
else:
col_text = 0
lbl = [Link](data_frame,
text=text + ":",
font=self.font_label,
text_color=self.COLOR_TEXT,
anchor="w")
[Link](row=i, column=col_text, sticky="w", padx=5, pady=3)
val = [Link](data_frame,
text="--",
font=self.font_value,
text_color=self.COLOR_ACCENT,
anchor="e")
[Link](row=i, column=col_text+1, sticky="e", padx=5, pady=3)
[Link][key] = val
# Right column
for j, (text, key) in enumerate(right):
icon = load_icon_for(key)
if icon:
lbl_icon = [Link](data_frame, image=icon, text="")
lbl_icon.image = icon
lbl_icon.grid(row=j, column=2, padx=(30,2), pady=3, sticky="w")
col_text = 3
else:
col_text = 2
lbl = [Link](data_frame,
text=text + ":",
font=self.font_label,
text_color=self.COLOR_TEXT,
anchor="w")
[Link](row=j, column=col_text, sticky="w", padx=5, pady=3)
val = [Link](data_frame,
text="--",
font=self.font_value,
text_color=self.COLOR_ACCENT,
anchor="e")
[Link](row=j, column=col_text+1, sticky="e", padx=5, pady=3)
[Link][key] = val
# Graphs area
graph_frame = [Link](self,
fg_color=self.COLOR_BG_FRAME,
corner_radius=8,
border_width=1,
border_color=self.COLOR_BORDER)
graph_frame.pack(side="bottom", fill="x", padx=10, pady=(5,15))
graphs_inner = [Link](graph_frame,
fg_color=self.COLOR_BG_FRAME,
corner_radius=8)
graphs_inner.pack(fill="both", expand=False, padx=10, pady=10)
# Ultrasonic graph
fig1 = Figure(figsize=(4.5,2.8), dpi=100, tight_layout=True)
ax1 = fig1.add_subplot(111)
ax1.set_title("Ultrasonic (cm)", color=self.COLOR_ACCENT, fontsize=13, fontfamily="Arial")
ax1.set_xlabel("Samples", color=self.COLOR_TEXT, fontsize=11)
ax1.set_ylabel("cm", color=self.COLOR_TEXT, fontsize=11)
ax1.tick_params(colors=self.COLOR_TEXT)
[Link]['bottom'].set_color(self.COLOR_TEXT)
[Link]['left'].set_color(self.COLOR_TEXT)
ax1.set_facecolor(self.COLOR_PLOT_BG)
canvas1 = FigureCanvasTkAgg(fig1, master=graphs_inner)
canvas1.get_tk_widget().pack(side="left", padx=15, pady=5)
# MQ-135 graph
fig2 = Figure(figsize=(4.5,2.8), dpi=100, tight_layout=True)
ax2 = fig2.add_subplot(111)
ax2.set_title("MQ-135", color=self.COLOR_ACCENT, fontsize=13, fontfamily="Arial")
ax2.set_xlabel("Samples", color=self.COLOR_TEXT, fontsize=11)
ax2.set_ylabel("ADC", color=self.COLOR_TEXT, fontsize=11)
ax2.tick_params(colors=self.COLOR_TEXT)
[Link]['bottom'].set_color(self.COLOR_TEXT)
[Link]['left'].set_color(self.COLOR_TEXT)
ax2.set_facecolor(self.COLOR_PLOT_BG)
canvas2 = FigureCanvasTkAgg(fig2, master=graphs_inner)
canvas2.get_tk_widget().pack(side="left", padx=15, pady=5)
# Log box
bottom = [Link](self,
fg_color=self.COLOR_BG_FRAME,
corner_radius=8,
border_width=1,
border_color=self.COLOR_BORDER)
[Link](side="bottom", fill="x", padx=10, pady=(5,10))
[Link] = [Link](bottom,
height=5,
font=self.font_log,
text_color=self.COLOR_TEXT)
[Link](fill="both", padx=5, pady=5)
# Save references
self.ax1 = ax1; self.fig1 = fig1; self.canvas1 = canvas1
self.ax2 = ax2; self.fig2 = fig2; self.canvas2 = canvas2
# Start serial reader thread
self.serial_thread = SerialReader(SERIAL_PORT, SERIAL_BAUD, SERIAL_TIMEOUT,
telemetry_queue)
self.serial_thread.start()
# Schedule periodic update
[Link](200, self.periodic_update)
def log_msg(self, message: str):
"""Append message with timestamp to log box."""
ts = [Link]().strftime("%H:%M:%S")
[Link]("end", f"[{ts}] {message}\n")
[Link]("end")
def periodic_update(self):
"""Read telemetry queue, update indicators & graphs."""
global last_snapshot, last_recv_time
updated = False
while True:
try:
data = telemetry_queue.get_nowait()
except [Link]:
break
last_snapshot = data
last_recv_time = [Link]()
updated = True
# Extract data
dist_mm = [Link]("dist_mm") or 0
dist_cm = round(dist_mm / 10.0, 1)
mq = [Link]("mq") or [Link]("mq135") or 0
temp = [Link]("temp_c")
hum = [Link]("hum")
fix = [Link]("fix") or 0
sats = [Link]("sats") or 0
lat = [Link]("lat") or ""
lon = [Link]("lon") or ""
seq = [Link]("seq") or 0
spd = [Link]("spd_cms") or 0
buz = [Link]("buzzer") if "buzzer" in data else (([Link]("flags",0) >> 0) & 1)
# Update indicators
[Link]["dist_cm"].configure(text=f"{dist_cm}")
[Link]["mq"].configure(text=str(mq))
if temp is not None:
[Link]["temp_c"].configure(text=f"{temp:.1f}")
if hum is not None:
[Link]["hum"].configure(text=f"{hum:.1f}")
[Link]["fix"].configure(text=str(fix))
[Link]["sats"].configure(text=str(sats))
[Link]["lat"].configure(text=str(lat))
[Link]["lon"].configure(text=str(lon))
[Link]["spd_cms"].configure(text=str(spd))
[Link]["seq"].configure(text=str(seq))
[Link]["buzzer"].configure(text=str(int(bool(buz))))
if updated:
self.redraw_graphs()
[Link](200, self.periodic_update)
def redraw_graphs(self):
"""Redraw graph plots with history data."""
# Ultrasonic
[Link]()
self.ax1.set_title("Ultrasonic (cm)", color=self.COLOR_ACCENT, fontsize=13, fontfamily="Arial")
self.ax1.set_xlabel("Samples", color=self.COLOR_TEXT, fontsize=11)
self.ax1.set_ylabel("cm", color=self.COLOR_TEXT, fontsize=11)
self.ax1.tick_params(colors=self.COLOR_TEXT)
[Link]['bottom'].set_color(self.COLOR_TEXT)
[Link]['left'].set_color(self.COLOR_TEXT)
self.ax1.set_facecolor(self.COLOR_PLOT_BG)
if len(dist_history) > 0:
xs = list(range(len(dist_history)))
ys = list(dist_history)
[Link](xs, ys, color="#00FF7F", linestyle='-', linewidth=1.5)
self.ax1.set_ylim(0, max(max(ys)*1.2, 20))
self.canvas1.draw_idle()
# MQ-135
[Link]()
self.ax2.set_title("MQ-135", color=self.COLOR_ACCENT, fontsize=13, fontfamily="Arial")
self.ax2.set_xlabel("Samples", color=self.COLOR_TEXT, fontsize=11)
self.ax2.set_ylabel("ADC", color=self.COLOR_TEXT, fontsize=11)
self.ax2.tick_params(colors=self.COLOR_TEXT)
[Link]['bottom'].set_color(self.COLOR_TEXT)
[Link]['left'].set_color(self.COLOR_TEXT)
self.ax2.set_facecolor(self.COLOR_PLOT_BG)
if len(mq_history) > 0:
xs2 = list(range(len(mq_history)))
ys2 = list(mq_history)
[Link](xs2, ys2, color="#FFA500", linestyle='-', linewidth=1.5)
ymin = min(ys2)
ymax = max(ys2)
if ymin == ymax:
self.ax2.set_ylim(ymin - 10, ymax + 10)
else:
pad = (ymax - ymin) * 0.2
self.ax2.set_ylim(ymin - pad, ymax + pad)
self.canvas2.draw_idle()
# ----------------- Main -----------------
def main():
app = EquinoxGUI()
try:
[Link]()
except KeyboardInterrupt:
pass
finally:
if hasattr(app, "serial_thread"):
app.serial_thread.stop()
if __name__ == "__main__":
main()