| from __future__ import annotations
|
|
|
| import subprocess
|
| import sys
|
| from pathlib import Path
|
|
|
| if __package__ in {None, ""}:
|
| sys.path.append(str(Path(__file__).resolve().parents[1]))
|
| from openpeer_trainer.benchmarks import run_benchmark_suite
|
| from openpeer_trainer.hardware import collect_hardware_specs, hardware_table_rows
|
| else:
|
| from .benchmarks import run_benchmark_suite
|
| from .hardware import collect_hardware_specs, hardware_table_rows
|
|
|
|
|
| def run_app() -> None:
|
| try:
|
| import streamlit as st
|
| except ImportError as exc:
|
| raise RuntimeError("Streamlit is required for the runtime GUI. Install with `pip install -e \".[gui]\"`") from exc
|
|
|
| st.set_page_config(page_title="OpenPeer NTK Trainer", page_icon="🧠", layout="wide")
|
|
|
| specs = collect_hardware_specs()
|
| st.title("OpenPeer NTK Trainer Runtime GUI")
|
| st.caption("Live benchmark dashboard with hardware specs, target-accuracy training, and OpenBB-backed charts when available.")
|
|
|
| with st.sidebar:
|
| st.header("Runtime Controls")
|
| step_text = st.text_input("Step schedule", value="10, 25, 50")
|
| batch_size = st.slider("Batch size", min_value=8, max_value=256, value=64, step=8)
|
| seed = st.number_input("Seed", min_value=0, max_value=10_000, value=0, step=1)
|
| target_accuracy = st.slider("Target accuracy", min_value=0.90, max_value=0.999, value=0.99, step=0.001, format="%.3f")
|
| output_dir = st.text_input("Output directory", value="artifacts/runtime_gui")
|
| run_label = st.button("Run benchmark")
|
|
|
| st.divider()
|
| st.subheader("Hardware Specs")
|
| st.write(f"Hostname: {specs.hostname}")
|
| st.write(f"Platform: {specs.platform}")
|
| st.write(f"CPU: {specs.cpu_model}")
|
| st.write(f"Cores: {specs.physical_cores} physical / {specs.logical_cores} logical")
|
| st.write(f"Memory: {specs.memory_available_gb:.2f} GB free of {specs.memory_total_gb:.2f} GB")
|
| st.write(f"Disk: {specs.disk_free_gb:.2f} GB free of {specs.disk_total_gb:.2f} GB")
|
| st.write(f"Python: {specs.python_version}")
|
| st.write(f"CUDA: {'yes' if specs.cuda_available else 'no'}")
|
|
|
| if run_label:
|
| step_counts = [int(part.strip()) for part in step_text.split(",") if part.strip()]
|
| result = run_benchmark_suite(
|
| step_counts=step_counts,
|
| batch_size=batch_size,
|
| seed=int(seed),
|
| output_dir=output_dir,
|
| target_accuracy=float(target_accuracy),
|
| )
|
| st.success(f"Saved benchmark artifacts to {result.csv_path}")
|
| st.session_state["latest_result"] = result
|
| st.session_state["latest_output_dir"] = output_dir
|
|
|
| output_dir_path = Path(st.session_state.get("latest_output_dir", output_dir))
|
| dashboard_path = output_dir_path / "benchmark_dashboard.html"
|
| csv_path = output_dir_path / "gate_benchmarks.csv"
|
|
|
| cols = st.columns([1.1, 1.1, 1.1])
|
| cols[0].metric("Hostname", specs.hostname)
|
| cols[1].metric("CPU Cores", f"{specs.physical_cores}/{specs.logical_cores}")
|
| cols[2].metric("Memory Free GB", f"{specs.memory_available_gb:.2f}")
|
|
|
| st.subheader("Current Hardware")
|
| st.table(hardware_table_rows(specs))
|
|
|
| if csv_path.exists():
|
| import pandas as pd
|
|
|
| df = pd.read_csv(csv_path)
|
| st.subheader("Benchmark Data")
|
| st.dataframe(df, use_container_width=True)
|
| else:
|
| st.info("Run a benchmark to populate the table and dashboard.")
|
|
|
| if dashboard_path.exists():
|
| st.subheader("Dashboard Preview")
|
| st.components.v1.html(dashboard_path.read_text(encoding="utf-8"), height=1200, scrolling=True)
|
|
|
|
|
| def launch_runtime_gui() -> int:
|
| app_path = Path(__file__).resolve()
|
| command = [sys.executable, "-m", "streamlit", "run", str(app_path)]
|
| print("Launching runtime GUI at http://localhost:8501")
|
| completed = subprocess.run(command, check=False)
|
| return int(completed.returncode)
|
|
|
|
|
| if __name__ == "__main__":
|
| run_app()
|
|
|
