#!/usr/bin/env python3
"""
File: app/gui.py
Project: 22HLT01 QUMPHY
Contact: oskar.pfeffer@ptb.de
Gitlab: https://gitlab.com/qumphy
Description: Graphical user interface for training the models.
Cross-platform Tk GUI that:
* lets the user pick (or create) a Python environment to run in:
- current Python interpreter,
- an existing conda environment,
- an existing virtualenv / venv directory,
- a new venv created from requirements.txt,
- a new conda env created from requirements.txt;
* lets the user pick a YAML config from app/configs (or browse the
filesystem) and runs:
<env-python> train.py --config <CONFIG_FILE>
Standard-library only — runs on Windows, macOS and Linux.
"""
from __future__ import annotations
import json
import os
import queue
import shlex
import shutil
import subprocess
import sys
import tempfile
import threading
from pathlib import Path
try:
import yaml # type: ignore
except ImportError: # PyYAML is optional — used only for validation
yaml = None
from tkinter import (
BOTH,
END,
LEFT,
RIGHT,
BOTTOM,
TOP,
X,
Y,
DISABLED,
NORMAL,
StringVar,
Tk,
Toplevel,
filedialog,
messagebox,
ttk,
Text,
Scrollbar,
)
APP_DIR = Path(__file__).resolve().parent
REPO_ROOT = APP_DIR.parent
CONFIGS_DIR = APP_DIR / "configs"
TRAIN_SCRIPT = APP_DIR / "train.py"
REQUIREMENTS_FILE = REPO_ROOT / "requirements.txt"
CONFIG_EXTENSIONS = (".yaml", ".yml")
ENV_MODES = [
("current", "Current Python interpreter"),
("conda", "Existing conda environment"),
("venv", "Existing venv / virtualenv directory"),
]
[docs]
def venv_python(venv_path: Path) -> Path:
"""Return the python executable inside a venv directory."""
if os.name == "nt":
return venv_path / "Scripts" / "python.exe"
return venv_path / "bin" / "python"
[docs]
def list_conda_envs() -> list[str]:
"""Return conda environment names. Empty list if conda is unavailable."""
conda = shutil.which("conda") or shutil.which("mamba")
if not conda:
return []
try:
out = subprocess.check_output(
[conda, "env", "list", "--json"],
stderr=subprocess.STDOUT,
text=True,
timeout=10,
)
except (subprocess.SubprocessError, OSError):
return []
try:
data = json.loads(out)
except ValueError:
return []
names: list[str] = []
for env_path in data.get("envs", []):
name = Path(env_path).name
if name and name not in names:
names.append(name)
return names
[docs]
class TrainGUI:
"""Tk train-launcher window.
Wraps the full launcher workflow: pick (or create) a Python environment,
pick or edit a YAML config from ``app/configs``, and run :mod:`app.train`
against the selected config in a background subprocess whose output is
streamed back into the GUI.
Parameters
----------
root : tkinter.Tk
Top-level Tk window the GUI is attached to.
"""
def __init__(self, root: Tk) -> None:
"""Build the GUI, populate the config tree and start the output pump."""
self.root = root
self.root.title("QUMPHY — Train Launcher")
self.root.geometry("960x780")
# State
self.selected_config: StringVar = StringVar(value="")
self.env_mode: StringVar = StringVar(value="current")
self.env_target: StringVar = StringVar(value="")
self.status: StringVar = StringVar(
value="Pick an environment and a config, then press Run."
)
self.editor_status: StringVar = StringVar(value="No config loaded.")
self.process: subprocess.Popen | None = None
self.output_queue: queue.Queue[str] = queue.Queue()
# Editor state
self._editor_disk_content: str = "" # what's on disk for the current path
self._editor_loaded_path: str | None = None
self._tempfiles: list[Path] = [] # cleaned up on close
self._build_ui()
self._populate_tree()
self._on_env_mode_change()
self.selected_config.trace_add("write", self._on_config_path_changed)
self.root.after(100, self._drain_output)
self.root.protocol("WM_DELETE_WINDOW", self._on_close)
# ----------------------------- UI ---------------------------------- #
def _build_ui(self) -> None:
"""Lay out all widgets: environment selector, config tree, editor and output."""
# -------- Environment section --------
env_frame = ttk.LabelFrame(self.root, text="Python environment", padding=8)
env_frame.pack(side=TOP, fill=X, padx=8, pady=(8, 4))
ttk.Label(env_frame, text="Mode:").grid(row=0, column=0, sticky="w")
self.mode_combo = ttk.Combobox(
env_frame,
state="readonly",
values=[label for _, label in ENV_MODES],
width=42,
)
self.mode_combo.current(0)
self.mode_combo.grid(row=0, column=1, sticky="ew", padx=6)
self.mode_combo.bind(
"<<ComboboxSelected>>", lambda _e: self._on_env_mode_change()
)
self.env_target_label = ttk.Label(env_frame, text="Target:")
self.env_target_label.grid(row=1, column=0, sticky="w", pady=(6, 0))
self.env_target_entry = ttk.Entry(env_frame, textvariable=self.env_target)
self.env_target_entry.grid(row=1, column=1, sticky="ew", padx=6, pady=(6, 0))
self.env_browse_button = ttk.Button(
env_frame, text="Browse…", command=self._browse_env
)
self.env_browse_button.grid(row=1, column=2, padx=2, pady=(6, 0))
self.env_refresh_button = ttk.Button(
env_frame, text="Refresh", command=self._refresh_conda_envs
)
self.env_refresh_button.grid(row=1, column=3, padx=2, pady=(6, 0))
ttk.Button(
env_frame,
text="Create new env from requirements.txt…",
command=self._open_create_env_dialog,
).grid(row=2, column=1, sticky="w", padx=6, pady=(8, 0))
env_frame.columnconfigure(1, weight=1)
# -------- Config section --------
cfg_frame = ttk.LabelFrame(self.root, text="Training config", padding=8)
cfg_frame.pack(side=TOP, fill=BOTH, expand=True, padx=8, pady=4)
header = ttk.Frame(cfg_frame)
header.pack(side=TOP, fill=X)
ttk.Label(header, text=f"Configs in: {CONFIGS_DIR}").pack(side=LEFT)
ttk.Button(header, text="Refresh", command=self._populate_tree).pack(side=RIGHT)
tree_frame = ttk.Frame(cfg_frame)
tree_frame.pack(side=TOP, fill=BOTH, expand=True, pady=(6, 0))
self.tree = ttk.Treeview(tree_frame, show="tree", selectmode="browse")
tree_scroll = Scrollbar(tree_frame, orient="vertical", command=self.tree.yview)
self.tree.configure(yscrollcommand=tree_scroll.set)
self.tree.pack(side=LEFT, fill=BOTH, expand=True)
tree_scroll.pack(side=RIGHT, fill=Y)
self.tree.bind("<<TreeviewSelect>>", self._on_tree_select)
self.tree.bind("<Double-1>", lambda _e: self._run())
selrow = ttk.Frame(cfg_frame)
selrow.pack(side=TOP, fill=X, pady=(6, 0))
ttk.Label(selrow, text="Selected config:").grid(row=0, column=0, sticky="w")
ttk.Entry(selrow, textvariable=self.selected_config).grid(
row=0, column=1, sticky="ew", padx=6
)
ttk.Button(selrow, text="Browse…", command=self._browse_config).grid(
row=0, column=2, padx=2
)
self.run_button = ttk.Button(selrow, text="Run", command=self._run)
self.run_button.grid(row=0, column=3, padx=2)
self.stop_button = ttk.Button(
selrow, text="Stop", command=self._stop, state=DISABLED
)
self.stop_button.grid(row=0, column=4, padx=2)
selrow.columnconfigure(1, weight=1)
# -------- Notebook: Config editor + Output --------
nb = ttk.Notebook(self.root)
nb.pack(side=TOP, fill=BOTH, expand=True, padx=8, pady=(4, 0))
self.notebook = nb
# Editor tab
editor_tab = ttk.Frame(nb, padding=4)
nb.add(editor_tab, text="Config (editable)")
ed_actions = ttk.Frame(editor_tab)
ed_actions.pack(side=TOP, fill=X, pady=(0, 4))
ttk.Label(ed_actions, textvariable=self.editor_status).pack(side=LEFT)
self.editor_reload_button = ttk.Button(
ed_actions,
text="Reload from disk",
command=self._editor_reload,
state=DISABLED,
)
self.editor_reload_button.pack(side=RIGHT, padx=2)
self.editor_save_button = ttk.Button(
ed_actions, text="Save", command=self._editor_save, state=DISABLED
)
self.editor_save_button.pack(side=RIGHT, padx=2)
ed_text_frame = ttk.Frame(editor_tab)
ed_text_frame.pack(side=TOP, fill=BOTH, expand=True)
self.editor = Text(ed_text_frame, wrap="none", undo=True)
ed_scroll_y = Scrollbar(
ed_text_frame, orient="vertical", command=self.editor.yview
)
ed_scroll_x = Scrollbar(
ed_text_frame, orient="horizontal", command=self.editor.xview
)
self.editor.configure(
yscrollcommand=ed_scroll_y.set,
xscrollcommand=ed_scroll_x.set,
font=("TkFixedFont", 10),
)
ed_scroll_y.pack(side=RIGHT, fill=Y)
ed_scroll_x.pack(side=BOTTOM, fill=X)
self.editor.pack(side=LEFT, fill=BOTH, expand=True)
self.editor.bind("<<Modified>>", self._on_editor_modified)
self.editor.configure(state=DISABLED) # disabled until a config is loaded
# Output tab
out_tab = ttk.Frame(nb, padding=4)
nb.add(out_tab, text="Output")
self.output = Text(out_tab, height=15, wrap="none", state=DISABLED)
out_scroll_y = Scrollbar(out_tab, orient="vertical", command=self.output.yview)
self.output.configure(yscrollcommand=out_scroll_y.set)
self.output.pack(side=LEFT, fill=BOTH, expand=True)
out_scroll_y.pack(side=RIGHT, fill=Y)
status_bar = ttk.Frame(self.root, padding=(8, 2))
status_bar.pack(side=BOTTOM, fill=X)
ttk.Label(status_bar, textvariable=self.status).pack(side=LEFT)
# ----------------------------- Env -------------------------------- #
def _current_env_mode(self) -> str:
"""Return the active environment mode key (``current``/``conda``/``venv``)."""
idx = self.mode_combo.current()
return ENV_MODES[idx][0] if 0 <= idx < len(ENV_MODES) else "current"
def _on_env_mode_change(self) -> None:
"""Update widget states and seed the target field for the selected mode."""
mode = self._current_env_mode()
self.env_mode.set(mode)
self.env_target.set("")
if mode == "current":
self.env_target_entry.configure(state=DISABLED)
self.env_browse_button.configure(state=DISABLED)
self.env_refresh_button.configure(state=DISABLED)
self.env_target.set(sys.executable)
elif mode == "conda":
self.env_target_entry.configure(state=NORMAL)
self.env_browse_button.configure(state=DISABLED)
self.env_refresh_button.configure(state=NORMAL)
self._refresh_conda_envs()
elif mode == "venv":
self.env_target_entry.configure(state=NORMAL)
self.env_browse_button.configure(state=NORMAL)
self.env_refresh_button.configure(state=DISABLED)
def _refresh_conda_envs(self) -> None:
"""Re-query conda for environments and prompt the user to pick one."""
envs = list_conda_envs()
if not envs:
self.env_target.set("")
messagebox.showwarning(
"No conda envs",
"Could not find any conda environments (or `conda` is not on PATH).",
)
return
# Replace the entry with a combobox-like behavior: just put the first one
# in the entry, and offer a popup picker.
choice = self._pick_from_list("Select conda environment", envs)
if choice:
self.env_target.set(choice)
def _pick_from_list(self, title: str, items: list[str]) -> str | None:
"""Open a modal combobox dialog and return the selected item (or ``None``)."""
if not items:
return None
dlg = Toplevel(self.root)
dlg.title(title)
dlg.transient(self.root)
dlg.grab_set()
ttk.Label(dlg, text=title).pack(padx=12, pady=(12, 4))
var = StringVar(value=items[0])
combo = ttk.Combobox(
dlg, values=items, textvariable=var, state="readonly", width=40
)
combo.pack(padx=12, pady=4)
combo.current(0)
result: dict[str, str | None] = {"value": None}
def ok() -> None:
result["value"] = var.get()
dlg.destroy()
def cancel() -> None:
dlg.destroy()
btns = ttk.Frame(dlg)
btns.pack(pady=(8, 12))
ttk.Button(btns, text="OK", command=ok).pack(side=LEFT, padx=4)
ttk.Button(btns, text="Cancel", command=cancel).pack(side=LEFT, padx=4)
dlg.bind("<Return>", lambda _e: ok())
dlg.bind("<Escape>", lambda _e: cancel())
self.root.wait_window(dlg)
return result["value"]
def _browse_env(self) -> None:
"""Show a directory chooser for the venv directory (venv mode only)."""
if self._current_env_mode() != "venv":
return
path = filedialog.askdirectory(
title="Select venv directory",
initialdir=str(REPO_ROOT),
)
if path:
self.env_target.set(path)
def _resolve_python_cmd(self) -> list[str] | None:
"""Return the command prefix to invoke Python in the chosen environment.
Returns None on failure (and shows an error message).
"""
mode = self._current_env_mode()
target = self.env_target.get().strip()
if mode == "current":
return [sys.executable]
if mode == "venv":
if not target:
messagebox.showwarning("No env", "Please choose a venv directory.")
return None
py = venv_python(Path(target))
if not py.is_file():
messagebox.showerror(
"Invalid venv",
f"No python executable found at:\n{py}",
)
return None
return [str(py)]
if mode == "conda":
if not target:
messagebox.showwarning("No env", "Please choose a conda environment.")
return None
conda = shutil.which("conda") or shutil.which("mamba")
if not conda:
messagebox.showerror("conda missing", "`conda` was not found on PATH.")
return None
return [conda, "run", "--no-capture-output", "-n", target, "python"]
return None
# --------------------------- Env creation -------------------------- #
def _open_create_env_dialog(self) -> None:
"""Open the modal dialog that drives venv/conda environment creation."""
if self.process is not None:
messagebox.showinfo("Busy", "A command is already running.")
return
if not REQUIREMENTS_FILE.is_file():
messagebox.showerror(
"requirements.txt missing",
f"Cannot find {REQUIREMENTS_FILE}",
)
return
dlg = Toplevel(self.root)
dlg.title("Create new environment")
dlg.transient(self.root)
dlg.grab_set()
ttk.Label(dlg, text="Type:").grid(row=0, column=0, sticky="w", padx=8, pady=6)
type_combo = ttk.Combobox(
dlg,
state="readonly",
values=["venv (Python virtualenv)", "conda environment"],
width=32,
)
type_combo.current(0)
type_combo.grid(row=0, column=1, padx=8, pady=6, sticky="ew")
ttk.Label(dlg, text="Name / path:").grid(
row=1, column=0, sticky="w", padx=8, pady=6
)
name_var = StringVar(value=str(REPO_ROOT / ".venv"))
name_entry = ttk.Entry(dlg, textvariable=name_var, width=44)
name_entry.grid(row=1, column=1, padx=8, pady=6, sticky="ew")
browse_btn = ttk.Button(
dlg,
text="Browse…",
command=lambda: self._pick_create_location(type_combo, name_var),
)
browse_btn.grid(row=1, column=2, padx=4, pady=6)
ttk.Label(dlg, text="Python version:").grid(
row=2, column=0, sticky="w", padx=8, pady=6
)
py_var = StringVar(value="3.11")
ttk.Entry(dlg, textvariable=py_var, width=10).grid(
row=2, column=1, sticky="w", padx=8, pady=6
)
hint = ttk.Label(
dlg,
text=(
"venv: a directory will be created (uses current Python).\n"
"conda: a named env will be created (Python version above)."
),
foreground="#555",
)
hint.grid(row=3, column=0, columnspan=3, padx=8, pady=(0, 4), sticky="w")
def on_type_change(_e=None) -> None:
if type_combo.current() == 0: # venv
name_var.set(str(REPO_ROOT / ".venv"))
browse_btn.configure(state=NORMAL)
else: # conda
name_var.set("qumphy")
browse_btn.configure(state=DISABLED)
type_combo.bind("<<ComboboxSelected>>", on_type_change)
def do_create() -> None:
kind = "venv" if type_combo.current() == 0 else "conda"
target = name_var.get().strip()
py_ver = py_var.get().strip() or "3.11"
if not target:
messagebox.showwarning(
"Missing target", "Please provide a name or path."
)
return
dlg.destroy()
self._start_env_creation(kind, target, py_ver)
btns = ttk.Frame(dlg)
btns.grid(row=4, column=0, columnspan=3, pady=(6, 10))
ttk.Button(btns, text="Create", command=do_create).pack(side=LEFT, padx=4)
ttk.Button(btns, text="Cancel", command=dlg.destroy).pack(side=LEFT, padx=4)
dlg.columnconfigure(1, weight=1)
self.root.wait_window(dlg)
def _pick_create_location(
self, type_combo: ttk.Combobox, name_var: StringVar
) -> None:
"""Browse for a parent directory and seed ``<dir>/.venv`` into ``name_var``."""
if type_combo.current() != 0:
return
path = filedialog.askdirectory(
title="Pick parent directory for the new venv",
initialdir=str(REPO_ROOT),
)
if path:
name_var.set(str(Path(path) / ".venv"))
def _start_env_creation(self, kind: str, target: str, py_ver: str) -> None:
"""Kick off venv or conda creation on a background thread.
Parameters
----------
kind : str
``"venv"`` or ``"conda"``.
target : str
Directory (venv) or environment name (conda).
py_ver : str
Python version string passed to ``conda create`` (ignored for venv).
"""
if self.process is not None:
messagebox.showinfo("Busy", "A command is already running.")
return
if kind == "venv":
steps = self._venv_creation_steps(Path(target))
on_success_mode = "venv"
success_target = target
elif kind == "conda":
conda = shutil.which("conda") or shutil.which("mamba")
if not conda:
messagebox.showerror("conda missing", "`conda` was not found on PATH.")
return
steps = self._conda_creation_steps(conda, target, py_ver)
on_success_mode = "conda"
success_target = target
else:
return
self.status.set(f"Creating {kind} environment…")
self._set_running(True)
threading.Thread(
target=self._run_steps_thread,
args=(steps, on_success_mode, success_target),
daemon=True,
).start()
def _venv_creation_steps(self, venv_dir: Path) -> list[list[str]]:
"""Return the ordered shell commands that build a venv at ``venv_dir``."""
py = venv_python(venv_dir)
return [
[sys.executable, "-m", "venv", str(venv_dir)],
[str(py), "-m", "pip", "install", "--upgrade", "pip"],
[str(py), "-m", "pip", "install", "-r", str(REQUIREMENTS_FILE)],
[str(py), "-m", "pip", "install", "-e", str(REPO_ROOT)],
]
def _conda_creation_steps(
self, conda: str, name: str, py_ver: str
) -> list[list[str]]:
"""Return the ordered shell commands that build conda env ``name``."""
run_prefix = [conda, "run", "--no-capture-output", "-n", name]
return [
[conda, "create", "-n", name, f"python={py_ver}", "pip", "-y"],
run_prefix + ["pip", "install", "-r", str(REQUIREMENTS_FILE)],
run_prefix + ["pip", "install", "-e", str(REPO_ROOT)],
]
def _run_steps_thread(
self,
steps: list[list[str]],
success_mode: str,
success_target: str,
) -> None:
"""Run ``steps`` sequentially, streaming output, stopping on first failure."""
rc = 0
for cmd in steps:
self.output_queue.put(f"$ {shlex.join(cmd)}\n")
try:
proc = subprocess.Popen(
cmd,
cwd=str(REPO_ROOT),
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=1,
text=True,
)
except OSError as exc:
self.output_queue.put(f"Failed to start: {exc}\n")
rc = -1
break
self.process = proc
assert proc.stdout is not None
for line in proc.stdout:
self.output_queue.put(line)
rc = proc.wait()
self.process = None
self.output_queue.put(f"[step exited with code {rc}]\n")
if rc != 0:
break
self.root.after(0, self._on_steps_done, rc, success_mode, success_target)
def _on_steps_done(self, rc: int, success_mode: str, success_target: str) -> None:
"""Re-enable controls and, on success, auto-select the newly created env."""
self._set_running(False)
if rc == 0:
self.status.set("Environment ready.")
# Auto-select the new env
for idx, (key, _label) in enumerate(ENV_MODES):
if key == success_mode:
self.mode_combo.current(idx)
break
self._on_env_mode_change() # resets target
self.env_target.set(success_target)
else:
self.status.set(f"Environment creation failed (exit {rc}).")
# ---------------------------- Config tree -------------------------- #
def _populate_tree(self) -> None:
"""Rebuild the config Treeview from the contents of ``app/configs/``."""
self.tree.delete(*self.tree.get_children())
if not CONFIGS_DIR.is_dir():
self.status.set(f"Configs dir not found: {CONFIGS_DIR}")
return
self._insert_dir("", CONFIGS_DIR)
def _insert_dir(self, parent: str, directory: Path) -> None:
"""Recursively insert ``directory`` and its YAML files into the tree."""
try:
entries = sorted(
directory.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower())
)
except OSError:
return
for entry in entries:
if entry.name.startswith("."):
continue
if entry.is_dir():
node = self.tree.insert(parent, END, text=entry.name + "/", open=False)
self._insert_dir(node, entry)
elif entry.suffix.lower() in CONFIG_EXTENSIONS:
self.tree.insert(parent, END, text=entry.name, values=(str(entry),))
def _on_tree_select(self, _event) -> None:
"""Sync the selected-config field with the user's tree selection."""
sel = self.tree.selection()
if not sel:
return
values = self.tree.item(sel[0], "values")
if values:
self.selected_config.set(values[0])
def _browse_config(self) -> None:
"""Open a file picker to choose a YAML config from anywhere on disk."""
path = filedialog.askopenfilename(
title="Select config file",
initialdir=str(CONFIGS_DIR if CONFIGS_DIR.is_dir() else APP_DIR),
filetypes=[("YAML config", "*.yaml *.yml"), ("All files", "*.*")],
)
if path:
self.selected_config.set(path)
# ----------------------------- Editor ------------------------------ #
def _on_config_path_changed(self, *_args) -> None:
"""Called whenever self.selected_config changes — load file into editor."""
path = self.selected_config.get().strip()
if not path:
self._editor_clear()
return
p = Path(path)
if not p.is_file():
self._editor_clear()
self.editor_status.set(f"File not found: {p}")
return
if self._editor_is_dirty() and self._editor_loaded_path != str(p):
if not messagebox.askyesno(
"Discard changes?",
"The config editor has unsaved changes.\n"
"Loading another config will discard them.\n\nContinue?",
):
# Revert the path back to whatever was previously loaded.
prev = self._editor_loaded_path or ""
# Avoid recursion via trace by using set() directly — it
# would re-enter, but the dirty-check will then be false
# because we're about to reset the editor. Simplest: set
# and accept a benign re-entry that no-ops.
self.selected_config.set(prev)
return
try:
content = p.read_text(encoding="utf-8")
except OSError as exc:
messagebox.showerror("Read failed", f"Cannot read {p}:\n{exc}")
self._editor_clear()
return
self._editor_load_content(str(p), content)
def _editor_load_content(self, path: str, content: str) -> None:
"""Replace the editor buffer with ``content`` and remember the source ``path``."""
self._editor_disk_content = content
self._editor_loaded_path = path
self.editor.configure(state=NORMAL)
self.editor.delete("1.0", END)
self.editor.insert("1.0", content)
self.editor.edit_reset()
self.editor.edit_modified(False)
self.editor_save_button.configure(state=NORMAL)
self.editor_reload_button.configure(state=NORMAL)
self.editor_status.set(f"Loaded: {path}")
def _editor_clear(self) -> None:
"""Empty and disable the editor and reset its on-disk reference."""
self._editor_disk_content = ""
self._editor_loaded_path = None
self.editor.configure(state=NORMAL)
self.editor.delete("1.0", END)
self.editor.edit_modified(False)
self.editor.configure(state=DISABLED)
self.editor_save_button.configure(state=DISABLED)
self.editor_reload_button.configure(state=DISABLED)
self.editor_status.set("No config loaded.")
def _editor_current_text(self) -> str:
"""Return the editor contents without the trailing Tk-added newline."""
# Tk's Text always appends a trailing newline; strip it for comparison.
return self.editor.get("1.0", "end-1c")
def _editor_is_dirty(self) -> bool:
"""Return True if the editor buffer differs from the on-disk content."""
if self._editor_loaded_path is None:
return False
return self._editor_current_text() != self._editor_disk_content
def _on_editor_modified(self, _event) -> None:
"""Refresh the editor status line whenever the buffer changes."""
# The <<Modified>> event only fires once until edit_modified(False) is called.
if not self.editor.edit_modified():
return
self.editor.edit_modified(False)
if self._editor_loaded_path is None:
return
if self._editor_is_dirty():
self.editor_status.set(f"Modified: {self._editor_loaded_path}")
else:
self.editor_status.set(f"Loaded: {self._editor_loaded_path}")
def _editor_save(self) -> None:
"""Write the editor buffer back to the loaded config file."""
if self._editor_loaded_path is None:
return
text = self._editor_current_text()
if not self._validate_yaml(text):
return
try:
Path(self._editor_loaded_path).write_text(text, encoding="utf-8")
except OSError as exc:
messagebox.showerror("Save failed", str(exc))
return
self._editor_disk_content = text
self.editor_status.set(f"Saved: {self._editor_loaded_path}")
def _editor_reload(self) -> None:
"""Discard unsaved edits (after confirmation) and re-read the file from disk."""
if self._editor_loaded_path is None:
return
if self._editor_is_dirty():
if not messagebox.askyesno(
"Discard changes?",
"Discard unsaved changes and reload from disk?",
):
return
try:
content = Path(self._editor_loaded_path).read_text(encoding="utf-8")
except OSError as exc:
messagebox.showerror("Read failed", str(exc))
return
self._editor_load_content(self._editor_loaded_path, content)
def _validate_yaml(self, text: str) -> bool:
"""Return True if text is valid YAML (or yaml isn't available)."""
if yaml is None:
return True
try:
yaml.safe_load(text)
except yaml.YAMLError as exc:
if not messagebox.askyesno(
"YAML parse error",
f"The editor content does not parse as valid YAML:\n\n{exc}\n\nContinue anyway?",
):
return False
return True
def _materialize_config_for_run(self) -> str | None:
"""Return the path to pass to train.py. Writes a temp file if dirty."""
path = self.selected_config.get().strip()
if not path:
return None
if not self._editor_is_dirty() or self._editor_loaded_path != path:
return path
text = self._editor_current_text()
if not self._validate_yaml(text):
return None
suffix = Path(path).suffix or ".yaml"
prefix = Path(path).stem + "."
fd, tmp_path = tempfile.mkstemp(prefix=prefix, suffix=suffix, text=True)
try:
with os.fdopen(fd, "w", encoding="utf-8") as fh:
fh.write(text)
except OSError as exc:
messagebox.showerror("Temp file failed", str(exc))
return None
self._tempfiles.append(Path(tmp_path))
self._append_output(f"[using edited config from temp file: {tmp_path}]\n")
return tmp_path
# ------------------------------ Run -------------------------------- #
def _run(self) -> None:
"""Spawn ``<env-python> train.py --config <selected>`` and stream its output."""
if self.process is not None:
messagebox.showinfo("Busy", "A command is already running.")
return
config = self.selected_config.get().strip()
if not config:
messagebox.showwarning(
"No config selected", "Please select a config file first."
)
return
if not Path(config).is_file():
messagebox.showerror("Invalid config", f"File does not exist:\n{config}")
return
if not TRAIN_SCRIPT.is_file():
messagebox.showerror("train.py missing", f"Cannot find {TRAIN_SCRIPT}")
return
py_cmd = self._resolve_python_cmd()
if py_cmd is None:
return
run_config = self._materialize_config_for_run()
if run_config is None:
return
cmd = py_cmd + [str(TRAIN_SCRIPT), "--config", run_config]
self._append_output(f"$ {shlex.join(cmd)}\n")
self.status.set("Running…")
self._set_running(True)
try:
self.process = subprocess.Popen(
cmd,
cwd=str(APP_DIR),
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=1,
text=True,
)
except OSError as exc:
self._append_output(f"Failed to start: {exc}\n")
self._set_running(False)
self.status.set("Failed to start.")
return
threading.Thread(target=self._train_reader_thread, daemon=True).start()
def _train_reader_thread(self) -> None:
"""Background reader: drain the train subprocess' stdout into the output queue."""
assert self.process is not None
assert self.process.stdout is not None
for line in self.process.stdout:
self.output_queue.put(line)
rc = self.process.wait()
self.process = None
self.output_queue.put(f"\n[process exited with code {rc}]\n")
self.root.after(0, self._on_train_done, rc)
def _on_train_done(self, rc: int) -> None:
"""Update the status bar when the training subprocess has exited."""
self._set_running(False)
self.status.set(f"Finished (exit code {rc}).")
def _set_running(self, running: bool) -> None:
"""Toggle Run/Stop buttons and the env mode combo for the running state."""
state = DISABLED if running else NORMAL
self.run_button.configure(state=state)
self.mode_combo.configure(state="disabled" if running else "readonly")
self.stop_button.configure(state=NORMAL if running else DISABLED)
def _stop(self) -> None:
"""Send SIGTERM to the running subprocess (no-op if nothing is running)."""
if self.process is None:
return
self._append_output("\n[terminating process…]\n")
try:
self.process.terminate()
except OSError as exc:
self._append_output(f"terminate failed: {exc}\n")
# ----------------------------- Output ------------------------------ #
def _drain_output(self) -> None:
"""Tk-main-loop tick that flushes queued output lines into the output panel."""
try:
while True:
line = self.output_queue.get_nowait()
self._append_output(line)
except queue.Empty:
pass
self.root.after(100, self._drain_output)
def _append_output(self, text: str) -> None:
"""Append ``text`` to the (read-only) output panel and scroll to the bottom."""
self.output.configure(state=NORMAL)
self.output.insert(END, text)
self.output.see(END)
self.output.configure(state=DISABLED)
def _on_close(self) -> None:
"""Window-close handler: confirm cleanup of running tasks and temp files."""
if self.process is not None:
if not messagebox.askyesno(
"Quit", "A command is in progress. Terminate it and quit?"
):
return
try:
self.process.terminate()
except OSError:
pass
if self._editor_is_dirty():
if not messagebox.askyesno(
"Quit", "The config editor has unsaved changes. Discard and quit?"
):
return
for tmp in self._tempfiles:
try:
tmp.unlink(missing_ok=True)
except OSError:
pass
self.root.destroy()
[docs]
def main() -> int:
"""Launch the train GUI and run the Tk main loop until the window closes.
Returns
-------
int
Process exit code (always ``0`` from the launcher itself).
"""
if os.name == "nt" and sys.stdout is None: # pragma: no cover
sys.stdout = open(os.devnull, "w")
sys.stderr = sys.stdout
root = Tk()
try:
style = ttk.Style()
if "clam" in style.theme_names():
style.theme_use("clam")
except Exception:
pass
TrainGUI(root)
root.mainloop()
return 0
if __name__ == "__main__":
sys.exit(main())