"""RTC Time Drift Tester - 단일 화면 GUI""" import os import sys import tkinter as tk from tkinter import ttk, messagebox, filedialog import threading from datetime import datetime from config import APP_TITLE, APP_WIDTH, APP_HEIGHT, DEFAULT_MONITOR_INTERVAL from models.controller import Controller from models.reading import TimeReading from network.udp_discovery import get_network_interfaces, search_controllers from services.time_sync_service import sync_all from services.drift_monitor import DriftMonitor from services.export_service import ( export_readings_csv, export_per_controller_csv, export_summary_csv, ) from gui.settings_dialog import SettingsDialog class App: def __init__(self): self.root = tk.Tk() self.root.title(APP_TITLE) self.root.geometry(f"{APP_WIDTH}x{APP_HEIGHT}") self.root.minsize(900, 600) # 아이콘 설정 (윈도우 좌상단 + 작업표시줄) if getattr(sys, 'frozen', False): base_path = sys._MEIPASS else: base_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) icon_path = os.path.join(base_path, "icon.ico") if os.path.exists(icon_path): self.root.iconbitmap(icon_path) self._controllers: list[Controller] = [] self._monitor = DriftMonitor() self._searching = False self._syncing = False self._log_queue: list[str] = [] self._search_finished = False self._search_results: list[Controller] = [] self._build_ui() self._update_clock() self.root.protocol("WM_DELETE_WINDOW", self._on_close) # ── UI 구성 ───────────────────────────────────────────── def _build_ui(self): # 상단: NIC + 검색 + PC 시간 self._build_top_bar() # 컨트롤러 테이블 self._build_ctrl_table() # 로그 (2줄) self._build_log() # 액션 바 (동기화 + 모니터링) self._build_action_bar() # 하단: 기록 + 요약 (좌우 분할) self._build_bottom_pane() def _build_top_bar(self): top = ttk.Frame(self.root) top.pack(fill=tk.X, padx=8, pady=(8, 2)) ttk.Label(top, text="네트워크:").pack(side=tk.LEFT) self._nic_var = tk.StringVar() self._nic_combo = ttk.Combobox( top, textvariable=self._nic_var, state="readonly", width=22) self._nic_combo.pack(side=tk.LEFT, padx=(4, 8)) self._btn_search = ttk.Button(top, text="검색", command=self._on_search) self._btn_search.pack(side=tk.LEFT) self._search_status = ttk.Label(top, text="", foreground="gray") self._search_status.pack(side=tk.LEFT, padx=8) # PC 시간 (오른쪽) self._clock_label = ttk.Label(top, text="", font=("Consolas", 13, "bold")) self._clock_label.pack(side=tk.RIGHT) ttk.Label(top, text="PC 시간 ").pack(side=tk.RIGHT) self._refresh_nics() def _build_ctrl_table(self): frame = ttk.LabelFrame(self.root, text="컨트롤러", padding=4) frame.pack(fill=tk.BOTH, padx=8, pady=2, expand=False) # Treeview cols = ("sel", "mac", "ip", "port", "name", "last_drift") self._ctrl_tree = ttk.Treeview(frame, columns=cols, show="headings", height=5) self._ctrl_tree.heading("sel", text="✓") self._ctrl_tree.heading("mac", text="MAC") self._ctrl_tree.heading("ip", text="IP") self._ctrl_tree.heading("port", text="포트") self._ctrl_tree.heading("name", text="이름") self._ctrl_tree.heading("last_drift", text="마지막 오차") self._ctrl_tree.column("sel", width=30, anchor=tk.CENTER, stretch=False) self._ctrl_tree.column("mac", width=140) self._ctrl_tree.column("ip", width=120) self._ctrl_tree.column("port", width=50) self._ctrl_tree.column("name", width=160) self._ctrl_tree.column("last_drift", width=90, anchor=tk.CENTER) scr = ttk.Scrollbar(frame, orient=tk.VERTICAL, command=self._ctrl_tree.yview) self._ctrl_tree.configure(yscrollcommand=scr.set) self._ctrl_tree.pack(side=tk.TOP, fill=tk.BOTH, expand=True) scr.place(relx=1.0, rely=0, relheight=1.0, anchor=tk.NE) self._ctrl_tree.bind("", self._on_ctrl_click) # 버튼 행 btn_row = ttk.Frame(frame) btn_row.pack(fill=tk.X, pady=(4, 0)) ttk.Button(btn_row, text="전체 선택", command=self._select_all).pack(side=tk.LEFT, padx=2) ttk.Button(btn_row, text="전체 해제", command=self._deselect_all).pack(side=tk.LEFT, padx=2) ttk.Separator(btn_row, orient=tk.VERTICAL).pack(side=tk.LEFT, fill=tk.Y, padx=8) ttk.Button(btn_row, text="시간 동기화", command=self._on_sync).pack(side=tk.LEFT, padx=2) ttk.Button(btn_row, text="설정 변경...", command=self._on_settings).pack(side=tk.LEFT, padx=2) def _build_log(self): frame = ttk.LabelFrame(self.root, text="로그", padding=2) frame.pack(fill=tk.X, padx=8, pady=2) self._log = tk.Text(frame, height=8, state=tk.DISABLED, wrap=tk.WORD, font=("Consolas", 9)) scr_log = ttk.Scrollbar(frame, orient=tk.VERTICAL, command=self._log.yview) self._log.configure(yscrollcommand=scr_log.set) scr_log.pack(side=tk.RIGHT, fill=tk.Y) self._log.pack(fill=tk.X) def _build_action_bar(self): bar = ttk.Frame(self.root) bar.pack(fill=tk.X, padx=8, pady=2) ttk.Label(bar, text="모니터링", font=("", 9, "bold")).pack(side=tk.LEFT, padx=(0, 8)) ttk.Label(bar, text="주기:").pack(side=tk.LEFT) self._interval_var = tk.StringVar(value=str(DEFAULT_MONITOR_INTERVAL)) ttk.Entry(bar, textvariable=self._interval_var, width=6).pack(side=tk.LEFT, padx=2) ttk.Label(bar, text="초").pack(side=tk.LEFT, padx=(0, 6)) self._btn_start = ttk.Button(bar, text="시작", command=self._on_monitor_start) self._btn_start.pack(side=tk.LEFT, padx=2) self._btn_stop = ttk.Button(bar, text="중지", command=self._on_monitor_stop, state=tk.DISABLED) self._btn_stop.pack(side=tk.LEFT, padx=2) self._btn_read = ttk.Button(bar, text="즉시 읽기", command=self._on_read_now) self._btn_read.pack(side=tk.LEFT, padx=2) ttk.Separator(bar, orient=tk.VERTICAL).pack(side=tk.LEFT, fill=tk.Y, padx=6) self._btn_hourly = ttk.Button(bar, text="매시 정시 읽기", command=self._on_hourly_toggle) self._btn_hourly.pack(side=tk.LEFT, padx=2) ttk.Separator(bar, orient=tk.VERTICAL).pack(side=tk.LEFT, fill=tk.Y, padx=6) ttk.Button(bar, text="기록 초기화", command=self._on_clear).pack(side=tk.LEFT, padx=2) self._mon_status = ttk.Label(bar, text="대기중", foreground="gray") self._mon_status.pack(side=tk.RIGHT) def _build_bottom_pane(self): pane = ttk.PanedWindow(self.root, orient=tk.HORIZONTAL) pane.pack(fill=tk.BOTH, expand=True, padx=8, pady=(2, 8)) # ── 왼쪽: 오차 기록 ── left = ttk.LabelFrame(pane, text="오차 기록", padding=4) filt = ttk.Frame(left) filt.pack(fill=tk.X, pady=(0, 4)) ttk.Label(filt, text="컨트롤러:").pack(side=tk.LEFT) self._filter_var = tk.StringVar(value="전체") self._filter_combo = ttk.Combobox( filt, textvariable=self._filter_var, state="readonly", width=28) self._filter_combo["values"] = ["전체"] self._filter_combo.pack(side=tk.LEFT, padx=4) self._filter_combo.bind("<>", self._on_filter_changed) cols_r = ("ctrl", "time", "ctrl_time", "drift") self._read_tree = ttk.Treeview(left, columns=cols_r, show="headings", height=10) self._read_tree.heading("ctrl", text="컨트롤러") self._read_tree.heading("time", text="PC 시간") self._read_tree.heading("ctrl_time", text="컨트롤러 시간") self._read_tree.heading("drift", text="오차") self._read_tree.column("ctrl", width=130) self._read_tree.column("time", width=135) self._read_tree.column("ctrl_time", width=135) self._read_tree.column("drift", width=70, anchor=tk.CENTER) scr_r = ttk.Scrollbar(left, orient=tk.VERTICAL, command=self._read_tree.yview) self._read_tree.configure(yscrollcommand=scr_r.set) self._read_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scr_r.pack(side=tk.RIGHT, fill=tk.Y) # ── 오른쪽: 요약 + 내보내기 ── right = ttk.LabelFrame(pane, text="요약", padding=4) cols_s = ("ctrl", "count", "avg", "max_d") self._sum_tree = ttk.Treeview(right, columns=cols_s, show="headings", height=10) self._sum_tree.heading("ctrl", text="컨트롤러") self._sum_tree.heading("count", text="읽기 횟수") self._sum_tree.heading("avg", text="평균 오차(초)") self._sum_tree.heading("max_d", text="최대 오차(초)") self._sum_tree.column("ctrl", width=150) self._sum_tree.column("count", width=70, anchor=tk.CENTER) self._sum_tree.column("avg", width=90, anchor=tk.CENTER) self._sum_tree.column("max_d", width=90, anchor=tk.CENTER) scr_s = ttk.Scrollbar(right, orient=tk.VERTICAL, command=self._sum_tree.yview) self._sum_tree.configure(yscrollcommand=scr_s.set) self._sum_tree.pack(side=tk.TOP, fill=tk.BOTH, expand=True) scr_s.place(relx=1.0, rely=0, relheight=0.85, anchor=tk.NE) # 내보내기 버튼 exp = ttk.Frame(right) exp.pack(fill=tk.X, pady=(6, 0)) mb = ttk.Menubutton(exp, text="CSV 내보내기 ▾") menu = tk.Menu(mb, tearoff=0) menu.add_command(label="전체 내보내기 (단일 파일)", command=self._export_all) menu.add_command(label="컨트롤러별 내보내기 (폴더)", command=self._export_per_ctrl) menu.add_separator() menu.add_command(label="요약 내보내기", command=self._export_summary) mb["menu"] = menu mb.pack(side=tk.LEFT) pane.add(left, weight=3) pane.add(right, weight=2) # ── 유틸 ──────────────────────────────────────────────── def _log_msg(self, msg: str): self._log.config(state=tk.NORMAL) self._log.insert(tk.END, msg + "\n") self._log.see(tk.END) self._log.config(state=tk.DISABLED) def _update_clock(self): self._clock_label.config(text=datetime.now().strftime("%Y-%m-%d %H:%M:%S")) self.root.after(1000, self._update_clock) def _refresh_nics(self): nics = get_network_interfaces() self._nic_list = nics self._nic_combo["values"] = [f"{ip}" for ip, _ in nics] if nics: self._nic_combo.current(0) def _get_bind_ip(self) -> str: idx = self._nic_combo.current() if 0 <= idx < len(self._nic_list): return self._nic_list[idx][0] return "0.0.0.0" def _refresh_ctrl_tree(self): self._ctrl_tree.delete(*self._ctrl_tree.get_children()) for ctrl in self._controllers: sel = "☑" if ctrl.selected else "☐" self._ctrl_tree.insert("", tk.END, values=( sel, ctrl.mac, ctrl.ip, ctrl.port, ctrl.name, "-" )) def _update_ctrl_drift(self, mac: str, drift_text: str): """컨트롤러 테이블의 '마지막 오차' 컬럼 갱신.""" for item in self._ctrl_tree.get_children(): vals = self._ctrl_tree.item(item, "values") if vals[1] == mac: new_vals = list(vals) new_vals[5] = drift_text self._ctrl_tree.item(item, values=new_vals) break def _update_filter_options(self): labels = ["전체"] + [c.display_label for c in self._controllers] self._filter_combo["values"] = labels def _get_selected_ctrl(self) -> Controller | None: sel = self._ctrl_tree.selection() if not sel: return None vals = self._ctrl_tree.item(sel[0], "values") mac = vals[1] for c in self._controllers: if c.mac == mac: return c return None # ── 컨트롤러 체크박스 ─────────────────────────────────── def _on_ctrl_click(self, event): col = self._ctrl_tree.identify_column(event.x) item = self._ctrl_tree.identify_row(event.y) if col == "#1" and item: # sel 컬럼 idx = self._ctrl_tree.index(item) if idx < len(self._controllers): ctrl = self._controllers[idx] ctrl.selected = not ctrl.selected vals = list(self._ctrl_tree.item(item, "values")) vals[0] = "☑" if ctrl.selected else "☐" self._ctrl_tree.item(item, values=vals) def _select_all(self): for ctrl in self._controllers: ctrl.selected = True self._refresh_ctrl_tree_sel() def _deselect_all(self): for ctrl in self._controllers: ctrl.selected = False self._refresh_ctrl_tree_sel() def _refresh_ctrl_tree_sel(self): for item in self._ctrl_tree.get_children(): idx = self._ctrl_tree.index(item) if idx < len(self._controllers): vals = list(self._ctrl_tree.item(item, "values")) vals[0] = "☑" if self._controllers[idx].selected else "☐" self._ctrl_tree.item(item, values=vals) # ── 검색 ──────────────────────────────────────────────── def _on_search(self): if self._searching: return self._searching = True self._btn_search.config(state=tk.DISABLED) self._search_status.config(text="검색중...") self._controllers.clear() self._ctrl_tree.delete(*self._ctrl_tree.get_children()) self._log_queue.clear() self._search_finished = False bind_ip = self._get_bind_ip() self._log_msg(f"검색 시작 (NIC: {bind_ip})") threading.Thread(target=self._search_thread, args=(bind_ip,), daemon=True).start() self._poll_search() def _search_thread(self, bind_ip: str): results = search_controllers( bind_ip=bind_ip, timeout=3.0, on_log=lambda m: self._log_queue.append(m), ) self._search_results = results self._search_finished = True def _poll_search(self): while self._log_queue: self._log_msg(self._log_queue.pop(0)) if self._search_finished: self._search_finished = False self._searching = False self._btn_search.config(state=tk.NORMAL) self._controllers.clear() self._controllers.extend(self._search_results) self._refresh_ctrl_tree() self._update_filter_options() self._search_status.config(text=f"{len(self._controllers)}개 발견") return self.root.after(100, self._poll_search) # ── 설정 변경 ─────────────────────────────────────────── def _on_settings(self): ctrl = self._get_selected_ctrl() if not ctrl: messagebox.showinfo("알림", "컨트롤러를 선택하세요.") return SettingsDialog(self.root, ctrl, self._get_bind_ip()) # 다이얼로그 닫힌 후 테이블 갱신 self._refresh_ctrl_tree() # ── 동기화 ────────────────────────────────────────────── def _on_sync(self): if self._syncing: return selected = [c for c in self._controllers if c.selected] if not selected: self._log_msg("동기화할 컨트롤러가 없습니다.") return self._syncing = True self._log_msg(f"동기화 시작 ({len(selected)}대)...") threading.Thread(target=self._sync_thread, daemon=True).start() def _sync_thread(self): results = sync_all(self._controllers) self.root.after(0, self._sync_done, results) def _sync_done(self, results): self._syncing = False ok_count = 0 for ctrl, ok, msg in results: tag = "OK" if ok else "FAIL" self._log_msg(f" [{tag}] {ctrl.display_label}: {msg}") drift_text = "동기화OK" if ok else "동기화실패" self._update_ctrl_drift(ctrl.mac, drift_text) if ok: ok_count += 1 self._log_msg(f"동기화 완료: {ok_count}/{len(results)} 성공") # ── 모니터링 ──────────────────────────────────────────── def _on_monitor_start(self): selected = [c for c in self._controllers if c.selected] if not selected: messagebox.showwarning("알림", "모니터링할 컨트롤러가 없습니다.") return try: interval = int(self._interval_var.get().strip()) if interval < 1: raise ValueError except ValueError: messagebox.showerror("오류", "주기는 1 이상의 정수(초)여야 합니다.") return self._monitor.configure( controllers=self._controllers, interval=interval, on_readings=lambda r: self.root.after(0, self._on_new_readings, r), on_error=lambda m: self.root.after(0, self._log_msg, m), ) self._monitor.start() self._btn_start.config(state=tk.DISABLED) self._btn_stop.config(state=tk.NORMAL) self._mon_status.config(text=f"모니터링 중 (주기 {interval}초)", foreground="green") def _on_monitor_stop(self): self._monitor.stop() self._btn_start.config(state=tk.NORMAL) self._btn_stop.config(state=tk.DISABLED) self._mon_status.config(text="정지됨", foreground="gray") def _on_read_now(self): if not self._controllers: messagebox.showwarning("알림", "컨트롤러가 없습니다.") return if not self._monitor.is_running: try: interval = int(self._interval_var.get().strip()) except ValueError: interval = DEFAULT_MONITOR_INTERVAL self._monitor.configure( controllers=self._controllers, interval=interval, on_readings=lambda r: self.root.after(0, self._on_new_readings, r), on_error=lambda m: self.root.after(0, self._log_msg, m), ) self._monitor.read_once() self._mon_status.config(text="읽기 중...", foreground="orange") def _on_clear(self): self._monitor.clear_readings() self._read_tree.delete(*self._read_tree.get_children()) self._sum_tree.delete(*self._sum_tree.get_children()) for ctrl in self._controllers: self._update_ctrl_drift(ctrl.mac, "-") def _on_new_readings(self, readings: list[TimeReading]): for r in readings: self._read_tree.insert("", tk.END, iid=None, values=( r.controller_label, r.pc_time.strftime("%Y-%m-%d %H:%M:%S"), r.controller_time.strftime("%Y-%m-%d %H:%M:%S"), r.drift_display, ), tags=(r.controller_mac,)) self._update_ctrl_drift(r.controller_mac, r.drift_display) # 필터 적용 self._apply_filter() # 스크롤 끝 children = self._read_tree.get_children() if children: self._read_tree.see(children[-1]) # 요약 갱신 self._refresh_summary() now = datetime.now().strftime("%H:%M:%S") if self._monitor.is_running: self._mon_status.config( text=f"모니터링 중 (마지막: {now})", foreground="green") else: self._mon_status.config(text=f"읽기 완료 ({now})", foreground="gray") # ── 필터 ──────────────────────────────────────────────── def _on_filter_changed(self, event=None): self._apply_filter() def _apply_filter(self): selected = self._filter_var.get() # 모든 항목을 보이거나 숨기기 (detach/reattach) # 간단하게: 트리 다시 빌드 self._read_tree.delete(*self._read_tree.get_children()) for r in self._monitor.readings: if selected == "전체" or r.controller_label == selected: self._read_tree.insert("", tk.END, values=( r.controller_label, r.pc_time.strftime("%Y-%m-%d %H:%M:%S"), r.controller_time.strftime("%Y-%m-%d %H:%M:%S"), r.drift_display, ), tags=(r.controller_mac,)) def _refresh_summary(self): self._sum_tree.delete(*self._sum_tree.get_children()) summary = self._monitor.get_summary() selected = self._filter_var.get() for mac, s in summary.items(): if selected != "전체" and s["label"] != selected: continue self._sum_tree.insert("", tk.END, values=( s["label"], s["count"], f"{s['avg_drift']:.2f}", f"{s['max_drift']:.2f}", )) # ── CSV 내보내기 ──────────────────────────────────────── def _export_all(self): readings = self._monitor.readings if not readings: messagebox.showinfo("알림", "내보낼 기록이 없습니다.") return fp = filedialog.asksaveasfilename( title="전체 기록 CSV 저장", defaultextension=".csv", filetypes=[("CSV", "*.csv")], initialfile=f"drift_all_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv", ) if fp: try: export_readings_csv(readings, fp) messagebox.showinfo("성공", f"저장 완료:\n{fp}") except Exception as e: messagebox.showerror("오류", str(e)) def _export_per_ctrl(self): readings = self._monitor.readings if not readings: messagebox.showinfo("알림", "내보낼 기록이 없습니다.") return folder = filedialog.askdirectory(title="컨트롤러별 CSV 저장 폴더 선택") if folder: try: files = export_per_controller_csv(readings, folder) messagebox.showinfo("성공", f"{len(files)}개 파일 생성:\n" + "\n".join(files)) except Exception as e: messagebox.showerror("오류", str(e)) def _export_summary(self): summary = self._monitor.get_summary() if not summary: messagebox.showinfo("알림", "내보낼 요약이 없습니다.") return fp = filedialog.asksaveasfilename( title="요약 CSV 저장", defaultextension=".csv", filetypes=[("CSV", "*.csv")], initialfile=f"drift_summary_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv", ) if fp: try: export_summary_csv(summary, fp) messagebox.showinfo("성공", f"저장 완료:\n{fp}") except Exception as e: messagebox.showerror("오류", str(e)) # ── 매시 정시 읽기 ──────────────────────────────────────── def _on_hourly_toggle(self): if self._monitor.is_hourly_running: # 중지 self._monitor.stop_hourly() self._btn_hourly.config(text="매시 정시 읽기") self._log_msg("매시 정시 읽기 중지") else: # 시작 selected = [c for c in self._controllers if c.selected] if not selected: messagebox.showwarning("알림", "컨트롤러를 선택하세요.") return self._monitor.configure( controllers=self._controllers, interval=self._monitor.interval, on_readings=lambda r: self.root.after(0, self._on_new_readings, r), on_error=lambda m: self.root.after(0, self._log_msg, m), ) self._monitor.start_hourly() self._btn_hourly.config(text="정시 읽기 중지") self._log_msg("매시 정시 읽기 시작 (매시 00분 00초에 읽기)") # ── 앱 생명주기 ───────────────────────────────────────── def _on_close(self): self._monitor.stop() self._monitor.stop_hourly() self.root.destroy() def run(self): self.root.mainloop()