#!/usr/bin/env python3 """ AS7341 STEM Визуализатор — все 5 режимов в реальном времени Запуск: python3 monitor.py [PORT] (по умолчанию /dev/ttyUSB0) Режимы переключаются автоматически по приходящим данным, либо нажми кнопку на окне. """ import sys import threading import serial import serial.tools.list_ports from datetime import datetime import matplotlib matplotlib.use("TkAgg") import matplotlib.pyplot as plt import matplotlib.patches as mpatches from matplotlib.animation import FuncAnimation from collections import deque # ─── Настройки ──────────────────────────────────────────────────────────────── PORT = sys.argv[1] if len(sys.argv) > 1 else None BAUD = 115200 HISTORY = 60 # точек в истории для линейных режимов # ─── Длины волн и цвета AS7341 ──────────────────────────────────────────────── WAVELENGTHS = [415, 445, 480, 515, 555, 590, 630, 680] BAR_COLORS = ["#9B30FF", "#4A4AFF", "#0080FF", "#00D0A0", "#80FF00", "#FFDD00", "#FF6600", "#FF1111"] LABELS_S = ["415nm","445nm","480nm","515nm","555nm","590nm","630nm","680nm","NIR","Clear"] LABELS_A = ["415nm","445nm","480nm","515nm","555nm","590nm","630nm","680nm"] LABELS_P = ["PAR","Blue445","Red680"] LABELS_C = ["CCT_K","Clear"] LABELS_I = ["NIR","VIS","NIR/VIS"] # Таблица усиления — зеркало As7341Lab::GAIN_STR (include/As7341Lab.h) и порядка # GAINS[] (src/As7341Lab.cpp). Прошивка не транслирует текущий Gain в потоке # телеметрии (только разово по команде +/-, а такие строки специально не # парсятся, см. фильтр "!!"/">>" ниже), поэтому индекс отслеживается локально: # кнопки ниже одновременно шлют команду на плату и двигают этот же индекс. GAIN_STR = ["0.5x","1x","2x","4x","8x","16x","32x","64x","128x","256x","512x"] GAIN_DEFAULT_IDX = 7 # As7341Lab::kDefaultGainIdx — соответствует 64x # ─── Авто-определение порта ─────────────────────────────────────────────────── def find_port(): ports = list(serial.tools.list_ports.comports()) for p in ports: if "USB" in p.device or "ACM" in p.device: return p.device return "/dev/ttyUSB0" # ─── Общее состояние ────────────────────────────────────────────────────────── class State: def __init__(self): self.lock = threading.Lock() self.mode = "S" self.values = {} # {label: float} self.history = {L: deque([0]*HISTORY, maxlen=HISTORY) for L in LABELS_S + LABELS_A + LABELS_P + LABELS_C + LABELS_I} self.status = "Подключение..." self.connected = False self.gain_idx = GAIN_DEFAULT_IDX self.link_lost = False # эхо As7341Lab::isAlive() из прошивки (main.cpp) state = State() # ─── Serial write (отправка команд без PlatformIO Monitor) ──────────────────── _ser_ref = [None] # [0] = serial.Serial instance or None _ser_wlock = threading.Lock() def serial_write(cmd: str): with _ser_wlock: ser = _ser_ref[0] if ser and ser.is_open: try: ser.write((cmd.strip() + '\n').encode()) except Exception: pass # ─── Serial reader ──────────────────────────────────────────────────────────── def detect_mode(vals: dict) -> str: keys = set(vals.keys()) if "PAR" in keys: return "P" if "CCT_K" in keys: return "C" if "NIR/VIS" in keys: return "I" # Absorption vs Spectrum — у обоих есть 415nm, но у A значения <5.0 if "415nm" in keys: if vals.get("415nm", 999) < 5.0: return "A" return "S" return "S" def serial_reader(port): while True: try: ser = serial.Serial(port, BAUD, timeout=2) with _ser_wlock: _ser_ref[0] = ser with state.lock: state.status = f"Подключено: {port}" state.connected = True with ser: while True: raw = ser.readline() try: line = raw.decode("utf-8", errors="replace").strip() except Exception: continue # Вотчдог связи: main.cpp печатает "!! ПОТЕРЯНА СВЯЗЬ ... !!", # когда As7341Lab::isAlive() возвращает false. Не даём этой # строке потеряться в фильтре декоративных строк ниже. if line.startswith("!!"): with state.lock: state.link_lost = True continue if not line or line.startswith(">>") or line.startswith("╔") \ or line.startswith("║") or line.startswith("╚") \ or line.startswith("╠") or line.startswith("---"): continue # Парсим пары label:value разделённые \t или пробелами pairs = {} for token in line.replace("\t", " ").split(): if ":" in token: k, _, v = token.partition(":") try: pairs[k.strip()] = float(v.strip()) except ValueError: pass if not pairs: continue with state.lock: state.values = pairs state.link_lost = False # свежая телеметрия = read() только что был успешен for k, v in pairs.items(): if k in state.history: state.history[k].append(v) except serial.SerialException as e: with _ser_wlock: _ser_ref[0] = None with state.lock: state.status = f"⚠ Порт недоступен ({port}). Закрой PlatformIO Monitor!" state.connected = False import time; time.sleep(3) # ─── Построение фигуры ──────────────────────────────────────────────────────── fig = plt.figure(figsize=(12, 6)) fig.patch.set_facecolor("#1A1A2E") ax = fig.add_subplot(111) # Панель статуса status_ax = fig.add_axes([0, 0.92, 1, 0.08]) status_ax.set_facecolor("#16213E") status_ax.axis("off") status_text = status_ax.text(0.5, 0.5, "", ha="center", va="center", color="white", fontsize=11, fontweight="bold") # ─── Панель «Светофор здоровья» ────────────────────────────────────────────── health_ax = fig.add_axes([0.745, 0.09, 0.245, 0.82]) health_ax.set_facecolor("#0D0D22") health_ax.set_xlim(0, 1) health_ax.set_ylim(0, 1) health_ax.axis("off") health_ax.text(0.5, 0.97, "Svetofor zdorovya", ha="center", va="top", color="#AAAAAA", fontsize=9, fontweight="bold") health_ax.text(0.5, 0.90, "(cirkadinnyj ritm)", ha="center", va="top", color="#555555", fontsize=7.5) # Ободок-свечение и основной круг health_glow = plt.Circle((0.5, 0.63), 0.30, facecolor="#444444", edgecolor="none", alpha=0.30, zorder=1) health_circle = plt.Circle((0.5, 0.63), 0.23, facecolor="#555555", edgecolor="none", zorder=2) health_ax.add_patch(health_glow) health_ax.add_patch(health_circle) health_icon_text = health_ax.text(0.5, 0.63, "?", ha="center", va="center", color="white", fontsize=22, fontweight="bold", zorder=3) health_label_text = health_ax.text(0.5, 0.34, "...", ha="center", va="top", color="#888888", fontsize=10, fontweight="bold", multialignment="center") health_detail_text = health_ax.text(0.5, 0.22, "", ha="center", va="top", color="#777777", fontsize=7.5, multialignment="center") health_time_text = health_ax.text(0.5, 0.03, "", ha="center", va="bottom", color="#555555", fontsize=9) # Оси, которые нельзя трогать (кнопки добавятся позже) _protected_axes = {health_ax} def clear_axes(): """Удаляет twin-оси (от режима I), которые ax.cla() не трогает.""" for a in fig.axes[:]: if a is not ax and a is not status_ax and a not in _protected_axes: a.remove() def style_ax(title, xlabel, ylabel): ax.set_facecolor("#0F3460") ax.tick_params(colors="white") ax.xaxis.label.set_color("white") ax.yaxis.label.set_color("white") ax.title.set_color("white") for spine in ax.spines.values(): spine.set_edgecolor("#444") ax.set_title(title, fontsize=14, pad=10) ax.set_xlabel(xlabel) ax.set_ylabel(ylabel) # ─── Рендер каждого режима ──────────────────────────────────────────────────── def draw_S(vals): clear_axes() ax.cla() values = [vals.get(L, 0) for L in LABELS_S] colors = BAR_COLORS + ["#AAAAAA", "#FFFFFF"] x = list(range(len(LABELS_S))) bars = ax.bar(x, values, color=colors, edgecolor="#222", linewidth=0.5, width=0.7) ax.set_xticks(x) ax.set_xticklabels(LABELS_S, rotation=30, ha="right") # Подписи значений for bar, val in zip(bars, values): if val > 0: ax.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 50, f"{int(val)}", ha="center", va="bottom", color="white", fontsize=8) style_ax("[S] Spektr (AS7341)", "Dlina volny (nm)", "Intensivnost (counts)") ax.set_ylim(0, max(max(values)*1.15, 1000)) # Легенда с физическими цветами patches = [mpatches.Patch(color=BAR_COLORS[i], label=f"{WAVELENGTHS[i]} нм") for i in range(len(WAVELENGTHS))] ax.legend(handles=patches, loc="upper right", ncol=4, facecolor="#1A1A2E", labelcolor="white", fontsize=7) def draw_A(vals): clear_axes() ax.cla() values = [vals.get(L, 0.0) for L in LABELS_A] # Проверяем что это действительно данные поглощения (все значения < 10) # а не остаток от спектра (тысячи counts) is_absorption_data = all(v < 10.0 for v in values) bars = ax.bar(range(8), values if is_absorption_data else [0]*8, color=BAR_COLORS, edgecolor="#222", width=0.7) ax.set_xticks(range(8)) ax.set_xticklabels([f"{w} nm" for w in WAVELENGTHS], rotation=30, ha="right") if is_absorption_data and any(v > 0.001 for v in values): for bar, val in zip(bars, values): ax.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.005, f"{val:.3f}", ha="center", va="bottom", color="white", fontsize=8) else: ax.text(0.5, 0.5, "Nажми [Ref/B] pri chistom osveshenii\n" "zatem dobavlyaj krasyashchee veshchestvo", ha="center", va="center", transform=ax.transAxes, color="#FFDD00", fontsize=13, fontweight="bold", bbox=dict(facecolor="#1A1A2E", alpha=0.85, boxstyle="round,pad=0.6")) style_ax("[A] Pogloshhenie — zakon Bugera-Lamberta-Bera", "Dlina volny (nm)", "Opticheskaya plotnost A = log10(I0/I)") ax.set_ylim(0, max(max(values)*1.2 if is_absorption_data else 0, 0.5)) def draw_P(vals): clear_axes() ax.cla() par = vals.get("PAR", 0.0) blue = vals.get("Blue445", 0.0) red = vals.get("Red680", 0.0) h_par = list(state.history["PAR"]) # ── Левая часть: крупный PAR + цветовые зоны ──────────────────────────── # Зоны фона ax.axhspan(0, 0.10, color="#220000", alpha=0.5) ax.axhspan(0.10, 0.30, color="#333300", alpha=0.5) ax.axhspan(0.30, 0.70, color="#003300", alpha=0.5) ax.axhspan(0.70, 2.00, color="#004400", alpha=0.5) # История PAR ax.fill_between(range(HISTORY), h_par, color="#00CC55", alpha=0.45, label="PAR история") ax.plot(range(HISTORY), h_par, color="#00FF77", linewidth=2) # Пороговые линии ax.axhline(0.10, color="#FF4444", linestyle="--", alpha=0.8, linewidth=1.2, label="тень <0.10") ax.axhline(0.30, color="#FFDD00", linestyle="--", alpha=0.8, linewidth=1.2, label="мин. рост 0.30") ax.axhline(0.70, color="#FF8800", linestyle="--", alpha=0.8, linewidth=1.2, label="полн. свет 0.70") # Крупное текущее значение в центре if par > 0: par_color = ("#FF4444" if par < 0.10 else "#FFDD00" if par < 0.30 else "#AAFFAA" if par < 0.70 else "#00FF88") ax.text(HISTORY * 0.5, max(max(h_par), 0.8) * 0.55, f"PAR = {par:.3f}", ha="center", va="center", color=par_color, fontsize=28, fontweight="bold", bbox=dict(facecolor="#0A0A1A", alpha=0.85, boxstyle="round,pad=0.4")) ax.text(HISTORY * 0.5, max(max(h_par), 0.8) * 0.28, f"Blue 445нм: {int(blue)} Red 680нм: {int(red)}", ha="center", va="center", color="#CCCCCC", fontsize=11, bbox=dict(facecolor="#0A0A1A", alpha=0.75, boxstyle="round,pad=0.3")) style_ax("[P] PAR-indeks fotosinteza (hlorofill A+B)", f"Poslednie {HISTORY} izmereny", "PAR-indeks") ax.legend(facecolor="#1A1A2E", labelcolor="white", fontsize=8, loc="upper left") ax.set_ylim(0, max(max(h_par + [0.1]) * 1.25, 1.1)) def draw_C(vals): clear_axes() ax.cla() cct = vals.get("CCT_K", 0) clear = vals.get("Clear", 0) h_cct = list(state.history["CCT_K"]) # Цвет линии — аппроксимация ЦТ def cct_color(k): if k <= 0: return "#888888" if k < 3000: return "#FF6600" if k < 4500: return "#FFD700" if k < 6500: return "#FFFFFF" return "#AADDFF" line_color = cct_color(cct) ax.fill_between(range(HISTORY), h_cct, color=line_color, alpha=0.25) ax.plot(range(HISTORY), h_cct, color=line_color, linewidth=2.5) for temp, label, color in [(2700,"Лампа накаливания","#FF6600"), (4000,"Тёплый белый","#FFD700"), (6500,"Дневной свет","#AADDFF")]: ax.axhline(temp, linestyle=":", color=color, alpha=0.6, linewidth=1, label=label) if cct > 0: ax.text(2, cct + 100, f"{int(cct)} K Clear={int(clear)}", color="white", fontsize=11, fontweight="bold", bbox=dict(facecolor="#1A1A2E", alpha=0.7)) style_ax("[C] Cvetovaya temperatura (McCamy / CIE XYZ)", f"Poslednie {HISTORY} izmereny", "CCT (Kelvin)") ax.legend(facecolor="#1A1A2E", labelcolor="white", fontsize=8) ax.set_ylim(0, max(max(h_cct+[1000])*1.1, 9000)) def draw_I(vals): clear_axes() ax.cla() nir = vals.get("NIR", 0.0) vis = vals.get("VIS", 0.0) ratio = vals.get("NIR/VIS", 0.0) h_nir = list(state.history["NIR"]) h_vis = list(state.history["VIS"]) h_rat = list(state.history["NIR/VIS"]) ax2 = ax.twinx() ax.fill_between(range(HISTORY), h_nir, color="#FF4400", alpha=0.35, label="NIR") ax.plot(range(HISTORY), h_nir, color="#FF6633", linewidth=2, label="NIR") ax.fill_between(range(HISTORY), h_vis, color="#4488FF", alpha=0.2, label="VIS") ax.plot(range(HISTORY), h_vis, color="#88AAFF", linewidth=2, label="VIS (сумма каналов)") ax2.plot(range(HISTORY), h_rat, color="#FFFF00", linewidth=1.5, linestyle="--", label="NIR/VIS коэф.") ax2.set_ylabel("Коэффициент NIR/VIS", color="yellow") ax2.tick_params(colors="yellow") ax.text(2, max(max(h_nir+[1]), max(h_vis+[1]))*0.85, f"NIR={int(nir)} VIS={int(vis)} NIR/VIS={ratio:.2f}", color="white", fontsize=10, bbox=dict(facecolor="#1A1A2E", alpha=0.7)) style_ax("[I] IK-detektor (NIR vs Vidimyj svet)", f"Poslednie {HISTORY} izmereny", "Intensivnost (counts)") lines1, labs1 = ax.get_legend_handles_labels() lines2, labs2 = ax2.get_legend_handles_labels() ax.legend(lines1+lines2, labs1+labs2, facecolor="#1A1A2E", labelcolor="white", fontsize=8, loc="upper right") ax2.set_facecolor("#0F3460") DRAW = {"S": draw_S, "A": draw_A, "P": draw_P, "C": draw_C, "I": draw_I} # ─── Светофор здоровья (циркадный ритм) ────────────────────────────────────── def circadian_health(cct: float, nm445: float, nm480: float, clear: float): """ Возвращает (color, icon, label, detail) на основе CCT, синего канала и текущего времени суток. Логика: синий свет (445-480 нм) вечером/ночью подавляет мелатонин. """ hour = datetime.now().hour if clear < 150: return "#666666", "?", "TEMNO", "Nedostatochno sveta\ndlya ocenki" blue_ratio = (nm445 + nm480) / max(clear, 1.0) has_cct = cct > 100 cct_str = f"CCT = {int(cct)} K" if has_cct else "CCT neizvestna" # ── Ночь 22:00–6:00 ────────────────────────────────────────────────────── if hour >= 22 or hour < 6: if (has_cct and cct > 3200) or blue_ratio > 0.25: return "#FF3333", "!", "OPASNO",\ f"Noch + sinij svet\nMelatonin podavlen!\n{cct_str}" return "#33CC55", "OK", "HOROSHO",\ f"Teplyj nochnoj svet\nMelatonin v norme\n{cct_str}" # ── Вечер 20:00–22:00 ──────────────────────────────────────────────────── if hour >= 20: if (has_cct and cct > 4500) or blue_ratio > 0.30: return "#FF3333", "!", "OPASNO",\ f"Vecher: slishkom holodnyj\nZasypanie zatrudneno\n{cct_str}" if has_cct and cct > 3200: return "#FFAA00", "~", "VNIMANIE",\ f"Vecher: luchshe teplee\nperekluchi na 2700 K\n{cct_str}" return "#33CC55", "OK", "HOROSHO",\ f"Teplyj vechernyj svet\nMelatonin v norme\n{cct_str}" # ── День 6:00–20:00 ────────────────────────────────────────────────────── if has_cct and cct > 4500: return "#33CC55", "OK", "OTLICHNO",\ f"Dnevnoj / holodnyj svet\nAktivnost, kontsentraciya\n{cct_str}" if has_cct and cct > 3000: return "#AAFFAA", "OK", "HOROSHO",\ f"Nejtralnyj belyj svet\n{cct_str}" if has_cct: return "#FFAA00", "~", "VNIMANIE",\ f"Den + teplyj svet\nMozhet vyzyvat sonlivost\n{cct_str}" return "#AAFFAA", "OK", "HOROSHO", "Den: svet v norme" def update(_frame): with state.lock: mode = state.mode vals = dict(state.values) status = state.status connected = state.connected gain_str = GAIN_STR[state.gain_idx] link_lost = state.link_lost mode_names = {"S":"SPEKTR","A":"POGLOSCH.","P":"PAR","C":"CCT","I":"IK"} hint = "knopki pereključajut rezhim na datchike" if connected else "ZAKROJ PlatformIO Monitor!" if connected and link_lost: # Эхо As7341Lab::isAlive()==false — физическая связь с датчиком по I2C # оборвана (провод/питание), хотя serial-порт ESP32 по-прежнему открыт. status_text.set_text(f"!! СВЯЗЬ С AS7341 ПОТЕРЯНА (проверь I2C/питание) !!") status_text.set_color("#FF3333") else: status_text.set_text(f"{status} | Rezhim: {mode_names.get(mode, mode)} | Gain: {gain_str} | {hint}") status_text.set_color("#FF6666" if not connected else "white") if vals: DRAW.get(mode, draw_S)(vals) # ── Светофор здоровья: обновляем каждый кадр ───────────────────────────── with state.lock: _cct = state.history["CCT_K"][-1] _nm445 = state.history["445nm"][-1] _nm480 = state.history["480nm"][-1] _clear = state.history["Clear"][-1] hcolor, hicon, hlabel, hdetail = circadian_health(_cct, _nm445, _nm480, _clear) health_circle.set_facecolor(hcolor) health_glow.set_facecolor(hcolor) health_icon_text.set_text(hicon) health_label_text.set_text(hlabel) health_label_text.set_color(hcolor) health_detail_text.set_text(hdetail) health_time_text.set_text(datetime.now().strftime("%H:%M")) return [] ani = FuncAnimation(fig, update, interval=900, blit=False, cache_frame_data=False) # ─── Кнопки ─────────────────────────────────────────────────────────────────── from matplotlib.widgets import Button # Общий шаг сетки кнопок: 8 кнопок (6 режимов + Gain -/+) должны поместиться # слева от панели "Светофор здоровья" (health_ax начинается на x=0.745). BTN_PITCH = 0.089 BTN_WIDTH = 0.078 BTN_MODES = [("S","Spektr"), ("A","Poglosch."), ("P","PAR"), ("C","CCT"), ("I","IK"), ("B","Ref/B")] btns = [] for i, (m, label) in enumerate(BTN_MODES): bax = fig.add_axes([0.01 + i*BTN_PITCH, 0.01, BTN_WIDTH, 0.06]) _protected_axes.add(bax) # защищаем от clear_axes() btn_color = "#1A3A1A" if m == "B" else "#16213E" b = Button(bax, label, color=btn_color, hovercolor="#0F3460") b.label.set_color("#AAFFAA" if m == "B" else "white") def on_click(event, _m=m): if _m == "B": # Ref — не меняем режим отображения, просто отправляем команду serial_write("B") else: with state.lock: state.mode = _m serial_write(_m) b.on_clicked(on_click) btns.append(b) # Gain +/- — датчик даёт 11 ступеней усиления (As7341Lab::GAIN_STR), и без этих # кнопок регулировать его можно только из PlatformIO Serial Monitor, который # при работающем monitor.py держать открытым нельзя (см. README, п.2) — тогда # управление Gain было бы недоступно прямо во время наблюдения за графиком. BTN_GAIN = [("-", -1), ("+", +1)] for i, (label, delta) in enumerate(BTN_GAIN): bax = fig.add_axes([0.01 + (len(BTN_MODES) + i) * BTN_PITCH, 0.01, BTN_WIDTH, 0.06]) _protected_axes.add(bax) b = Button(bax, f"Gain {label}", color="#16213E", hovercolor="#0F3460") b.label.set_color("white") def on_gain_click(event, _delta=delta): with state.lock: state.gain_idx = max(0, min(len(GAIN_STR) - 1, state.gain_idx + _delta)) serial_write("+" if _delta > 0 else "-") b.on_clicked(on_gain_click) btns.append(b) plt.subplots_adjust(left=0.08, right=0.72, top=0.91, bottom=0.18) # ─── Старт ──────────────────────────────────────────────────────────────────── if __name__ == "__main__": port = PORT or find_port() print(f"Подключение к {port} @ {BAUD} baud...") t = threading.Thread(target=serial_reader, args=(port,), daemon=True) t.start() plt.show()