File size: 4,136 Bytes
3f2dde4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | 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: # pragma: no cover - optional GUI dependency
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()
|