"""HTML dashboard for SalienceRuntime using the standard library."""
from __future__ import annotations
import argparse
import json
import threading
import time
from collections import deque
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Deque, Dict, List, Optional, Tuple
from urllib.parse import urlparse
from ...conversation.session import ConversationConfig, ConversationSession, ConversationSnapshot
from ...proto_lm.trainer import TrainingConfig
from ...runtime.orchestrator import RuntimeMetrics
from ...training.cot_curriculum.loader import iter_examples
from ...telemetry import (
BUS,
ParameterEvent,
SpatialEvent,
TelemetryEvent,
render_ingestion_event,
render_parameter_event,
render_spatial_event,
render_training_event,
)
BASE_DIR = Path(__file__).resolve().parents[2]
def _run_cot_curriculum(session: ConversationSession, *, limit: Optional[int] = None) -> Dict[str, int]:
curriculum_root = BASE_DIR / "training" / "cot_curriculum"
count = 0
for example in iter_examples(str(curriculum_root)):
trace_lines = [
f"Task: {example.task}",
"Trace:",
*example.reasoning_trace,
f"Answer: {example.answer}",
f"Meta: {example.meta_lesson}",
]
payload = "\n".join(trace_lines)
segments_processed, _ = session.ingest_text(
payload,
source="cot_curriculum",
max_chars=768,
)
count += segments_processed
if limit is not None and count >= limit:
break
return {"examples": count}
def _chunk_text(text: str, *, max_chars: int = 768) -> List[str]:
lines = text.splitlines()
chunks: List[str] = []
buffer: List[str] = []
length = 0
for line in lines:
line = line.strip()
if not line:
continue
line_len = len(line) + 1
if length + line_len > max_chars and buffer:
chunks.append("\n".join(buffer))
buffer = []
length = 0
buffer.append(line)
length += line_len
if buffer:
chunks.append("\n".join(buffer))
return chunks
def _ingest_food_corpus(session: ConversationSession) -> Dict[str, int]:
food_root = BASE_DIR / "food"
files = 0
chunks = 0
for path in sorted(food_root.rglob("*")):
if not path.is_file():
continue
if path.suffix.lower() not in {".txt", ".md", ".log", ".json"}:
continue
try:
content = path.read_text(encoding="utf-8", errors="ignore")
except Exception:
continue
content = content.strip()
if not content:
continue
for chunk in _chunk_text(content):
processed, _ = session.ingest_text(
chunk,
source=f"food::{path.name}",
max_chars=768,
)
chunks += processed
files += 1
return {"files": files, "chunks": chunks}
_SPARKLINE_BARS = "▁▂▃▄▅▆▇█"
def _render_sparkline(values: Deque[float], *, width: int = 32) -> str:
if not values:
return "—"
tail = list(values)[-width:]
minimum = min(tail)
maximum = max(tail)
if maximum - minimum < 1e-6:
return _SPARKLINE_BARS[0] * len(tail)
span = maximum - minimum
scale = len(_SPARKLINE_BARS) - 1
bars = []
for value in tail:
index = int((value - minimum) / span * scale)
index = max(0, min(scale, index))
bars.append(_SPARKLINE_BARS[index])
return "".join(bars)
def _format_salience(salience_raw: Dict[str, float]) -> Dict[str, object]:
if not salience_raw:
return {"top": [], "raw": {}}
positive_total = sum(max(value, 0.0) for value in salience_raw.values()) or 1.0
top = sorted(salience_raw.items(), key=lambda item: item[1], reverse=True)[:8]
return {
"top": [
{
"dimension": key,
"value": float(value),
"percent": max(value, 0.0) / positive_total,
}
for key, value in top
],
"raw": {key: float(value) for key, value in salience_raw.items()},
}
HTML_TEMPLATE = """
Salience Runtime Observatory
Learning + Filters
"""
class DashboardState:
"""Thread-safe container for conversation-aware snapshots and telemetry."""
def __init__(self, session: ConversationSession) -> None:
self._session = session
self._lock = threading.Lock()
self._latest_metrics: Optional[RuntimeMetrics] = None
self._latest_snapshot: Optional[ConversationSnapshot] = None
self._telemetry: List[Dict[str, object]] = []
self._max_telemetry = 250
self._cached_payload: Dict[str, object] = self._build_default_payload()
self._score_history: Deque[float] = deque(maxlen=96)
self._budget_history: Deque[float] = deque(maxlen=96)
self._accept_history: Deque[float] = deque(maxlen=96)
self._verification_history: Deque[float] = deque(maxlen=96)
def _build_default_payload(self) -> Dict[str, object]:
return {
"meta_report": "(awaiting conversation)",
"overview": [],
"decision_rows": [],
"status": {"budget_ratio_numeric": 0.0},
"status_rows": [],
"scratchpad": [""],
"salience": {"top": [], "raw": {}},
"yearning": {},
"todos": [{"Rank": "—", "Todo": "", "Score": "0.00"}],
"maintenance": [{"Category": "—", "Count": "0"}],
"experiments": [{"Experiment": "—", "Conclusion": ""}],
"history": [],
"sparklines": {
"score": "—",
"budget": "—",
"acceptance": "—",
"verification": "—",
},
"training_rows": [],
"training": {"learning_enabled": False, "runtime_active": False},
"filter": {"enabled": False, "thresholds": {}},
"spatial": {"summary": "no path", "ascii": "", "points": []},
}
def refresh_from_session(
self,
*,
metrics: Optional[RuntimeMetrics] = None,
snapshot: Optional[ConversationSnapshot] = None,
) -> None:
with self._lock:
if metrics is not None:
self._latest_metrics = metrics
if snapshot is not None:
self._latest_snapshot = snapshot
metrics_obj = self._latest_snapshot.metrics if self._latest_snapshot else self._latest_metrics
payload = self._build_default_payload()
if metrics_obj is not None:
decision = metrics_obj.decision
scheduler = dict(metrics_obj.scheduler_snapshot or {})
budget_ratio = float(scheduler.get("budget_ratio", 0.0))
verification = metrics_obj.verification_passed
verification_value = (
1.0
if verification is True
else 0.0
if verification is False
else 0.5
)
self._score_history.append(float(decision.score))
self._budget_history.append(budget_ratio)
self._accept_history.append(float(metrics_obj.idea_acceptances))
self._verification_history.append(verification_value)
payload["meta_report"] = metrics_obj.meta_report
overview_rows = [
{"label": "Step", "value": str(metrics_obj.step)},
{"label": "Budget left", "value": f"{metrics_obj.budget_left:.1f}"},
]
generator_desc = getattr(
self._latest_snapshot,
"generator_description",
"",
)
if generator_desc:
overview_rows.append({"label": "Generator", "value": generator_desc})
if metrics_obj.episode_recorded:
overview_rows.append({"label": "Episode", "value": metrics_obj.episode_recorded})
payload["overview"] = overview_rows
payload["decision_rows"] = [
{"label": "Action", "value": str(decision.action)},
{"label": "Score", "value": f"{decision.score:.2f}"},
{"label": "Hysteresis Δ", "value": f"{decision.hysteresis_delta:.2f}"},
{"label": "Cooldown", "value": str(decision.cooldown_steps)},
]
events = ", ".join(str(evt) for evt in scheduler.get("events", ())) or "—"
verification_label = (
"Passed" if verification is True else "Failed" if verification is False else "Pending"
)
payload["status"] = {
"step": metrics_obj.step,
"budget_left": metrics_obj.budget_left,
"idea_acceptances": metrics_obj.idea_acceptances,
"verification": verification_label,
"events": events,
"budget_ratio_numeric": budget_ratio,
}
payload["status_rows"] = [
{"label": "Budget left", "value": f"{metrics_obj.budget_left:.1f}"},
{"label": "Idea accepts", "value": str(metrics_obj.idea_acceptances)},
{"label": "Verification", "value": verification_label},
{"label": "Scheduler events", "value": events},
]
maintenance = metrics_obj.maintenance_report or {}
payload["maintenance"] = [
{"Category": key, "Count": str(value)}
for key, value in sorted(maintenance.items())
] or [{"Category": "—", "Count": "0"}]
experiments = metrics_obj.experiment_reports or ()
payload["experiments"] = [
{
"Experiment": report.get("name", "—"),
"Conclusion": report.get("conclusion", ""),
}
for report in experiments
] or [{"Experiment": "—", "Conclusion": ""}]
payload["salience"] = _format_salience(dict(metrics_obj.salience_raw or {}))
payload["yearning"] = dict(metrics_obj.yearning_snapshot or {})
scratchpad = getattr(self._session.runtime, "scratchpad", None)
if scratchpad is not None:
scratch_lines = list(scratchpad.current_trace[-20:])
if not scratch_lines:
scratch_lines = [""]
summary = scratchpad.summarize(max_traces=5)
if summary:
scratch_lines.append("")
scratch_lines.append("Summary: " + summary)
payload["scratchpad"] = scratch_lines
latest_path = (
scratchpad.latest_four_d_path() if hasattr(scratchpad, "latest_four_d_path") else None
)
if latest_path:
path_dict = latest_path.to_dict()
payload["spatial"] = {
"summary": latest_path.summary(),
"ascii": latest_path.ascii_projection(),
"points": path_dict.get("points", []),
}
else:
payload["spatial"] = {"summary": "no path", "ascii": "", "points": []}
else:
payload["scratchpad"] = [""]
payload["spatial"] = {"summary": "unavailable", "ascii": "", "points": []}
if self._latest_snapshot is not None:
todos_seq = list(self._latest_snapshot.todos)
payload["todos"] = [
{
"Rank": f"#{idx + 1}",
"Todo": getattr(item, "text", str(item)),
"Score": f"{getattr(item, 'score', 0.0):.2f}",
}
for idx, item in enumerate(todos_seq)
] or [{"Rank": "—", "Todo": "", "Score": "0.00"}]
payload["history"] = [
{"speaker": speaker, "text": text}
for speaker, text in list(self._session.history)[-64:]
]
if payload["overview"]:
payload["overview"].append(
{
"label": "Response length",
"value": str(len(self._latest_snapshot.response or "")),
}
)
else:
payload["history"] = []
payload["sparklines"] = {
"score": _render_sparkline(self._score_history),
"budget": _render_sparkline(self._budget_history),
"acceptance": _render_sparkline(self._accept_history),
"verification": _render_sparkline(self._verification_history),
}
training_rows, training_state = self._build_training_payload()
payload["training_rows"] = training_rows
payload["training"] = training_state
payload["filter"] = self._build_filter_payload()
self._cached_payload = payload
def _build_training_payload(self) -> Tuple[List[Dict[str, str]], Dict[str, object]]:
config = getattr(self._session, "config", None)
learning_enabled = bool(getattr(config, "learning_enabled", False))
buffer_items = len(getattr(self._session, "_learning_buffer", []))
buffer_chars = int(getattr(self._session, "_learning_accumulated_chars", 0))
last_flush = int(getattr(self._session, "_learning_last_flush_step", 0))
proto = getattr(self._session, "proto_lm", None)
model_step = int(getattr(proto, "step", 0))
runtime_active = bool(getattr(self._session.runtime.action_context, "training_active", False))
rows = [
{"label": "Learning enabled", "value": "yes" if learning_enabled else "no"},
{"label": "Runtime active", "value": "yes" if runtime_active else "no"},
{"label": "Buffer items", "value": str(buffer_items)},
{"label": "Buffered chars", "value": str(buffer_chars)},
{"label": "Last flush step", "value": str(last_flush)},
{"label": "Model step", "value": str(model_step)},
]
state = {
"learning_enabled": learning_enabled,
"runtime_active": runtime_active,
"buffer_items": buffer_items,
"buffer_chars": buffer_chars,
"last_flush_step": last_flush,
"model_step": model_step,
}
return rows, state
def _build_filter_payload(self) -> Dict[str, object]:
enabled = False
thresholds: Dict[str, Optional[float]] = {}
if hasattr(self._session, "is_filter_enabled"):
enabled = bool(self._session.is_filter_enabled())
filter_obj = getattr(self._session, "_filter", None)
if filter_obj is not None:
thresholds_obj = getattr(filter_obj, "thresholds", None)
if thresholds_obj is not None:
thresholds = {
"min_uncertainty": getattr(thresholds_obj, "min_uncertainty", None),
"min_novelty": getattr(thresholds_obj, "min_novelty", None),
"max_drag": getattr(thresholds_obj, "max_drag", None),
}
return {"enabled": enabled, "thresholds": thresholds}
def snapshot(self) -> Dict[str, object]:
with self._lock:
return dict(self._cached_payload)
def record_telemetry(self, event: TelemetryEvent) -> None:
if isinstance(event, ParameterEvent):
rendered = render_parameter_event(event)
elif isinstance(event, SpatialEvent):
rendered = render_spatial_event(event)
elif event.type == "training/step":
rendered = "\n".join(render_training_event(event))
elif event.type == "ingestion/chunk":
rendered = render_ingestion_event(event)
else:
rendered = json.dumps(event.payload)
entry = {
"type": event.type,
"rendered": rendered,
"timestamp": time.time(),
"payload": event.payload,
}
with self._lock:
self._telemetry.append(entry)
if len(self._telemetry) > self._max_telemetry:
self._telemetry = self._telemetry[-self._max_telemetry :]
def telemetry(self) -> Dict[str, object]:
with self._lock:
return {"entries": list(self._telemetry)}
STATE: Optional[DashboardState] = None
def telemetry_sink(event: TelemetryEvent) -> None:
if STATE is not None:
STATE.record_telemetry(event)
class DashboardHandler(BaseHTTPRequestHandler):
state: Optional[DashboardState] = None
session: Optional[ConversationSession] = None
def do_GET(self) -> None: # pragma: no cover - IO heavy
parsed = urlparse(self.path)
path = parsed.path
if path in {"/", "/index.html"}:
self._send_html(HTML_TEMPLATE)
return
if path == "/metrics":
if self.__class__.state is not None:
self.__class__.state.refresh_from_session()
payload = self.__class__.state.snapshot()
else:
payload = {"error": "dashboard not initialized"}
self._send_json(payload if payload else {"error": "snapshot not ready"})
return
if path == "/telemetry":
payload = self.__class__.state.telemetry() if self.__class__.state else {"entries": []}
self._send_json(payload)
return
self.send_error(HTTPStatus.NOT_FOUND, "Unknown endpoint")
def do_POST(self) -> None: # pragma: no cover - IO heavy
parsed = urlparse(self.path)
path = parsed.path
if path == "/utter":
self._handle_utter()
return
if path == "/ingest":
self._handle_ingest()
return
if path == "/think":
self._handle_think()
return
if path == "/run_gym":
self._handle_run_gym()
return
if path == "/eat_food":
self._handle_eat_food()
return
if path == "/toggle_learning":
self._handle_toggle_learning()
return
if path == "/flush_learning":
self._handle_flush_learning()
return
if path == "/training_active":
self._handle_training_active()
return
if path == "/toggle_filter":
self._handle_toggle_filter()
return
self.send_error(HTTPStatus.NOT_FOUND, "Unknown endpoint")
def _handle_utter(self) -> None:
if self.__class__.state is None or self.__class__.session is None:
self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, "Dashboard not initialized")
return
length = int(self.headers.get("Content-Length", "0"))
raw = self.rfile.read(length)
try:
payload = json.loads(raw.decode("utf-8")) if raw else {}
except json.JSONDecodeError:
self.send_error(HTTPStatus.BAD_REQUEST, "Invalid JSON")
return
text = str(payload.get("text", "")).strip()
if not text:
self.send_error(HTTPStatus.BAD_REQUEST, "Text is required")
return
session = self.__class__.session
try:
user_metrics = session.process_user_input(text)
self.__class__.state.refresh_from_session(metrics=user_metrics)
snapshot = session.generate_response()
self.__class__.state.refresh_from_session(snapshot=snapshot)
except Exception as exc: # pragma: no cover - runtime safety
self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, f"Conversation error: {exc}")
return
response = {
"response": snapshot.response,
"meta_report": snapshot.metrics.meta_report,
"status": "ok",
}
self._send_json(response)
def _handle_ingest(self) -> None:
if self.__class__.state is None or self.__class__.session is None:
self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, "Dashboard not initialized")
return
content_type = self.headers.get("Content-Type", "")
if "multipart/form-data" not in content_type:
self.send_error(HTTPStatus.BAD_REQUEST, "Expected multipart/form-data upload")
return
boundary_token = "boundary="
boundary_idx = content_type.find(boundary_token)
if boundary_idx == -1:
self.send_error(HTTPStatus.BAD_REQUEST, "Missing multipart boundary")
return
boundary = content_type[boundary_idx + len(boundary_token) :]
if boundary.startswith('"') and boundary.endswith('"'):
boundary = boundary[1:-1]
boundary_bytes = ("--" + boundary).encode("utf-8")
length = int(self.headers.get("Content-Length", "0"))
payload = self.rfile.read(length)
segments_total = 0
files_processed: List[str] = []
allow_duplicates = False
buffered_files: List[Tuple[str, str]] = []
parts = payload.split(boundary_bytes)
for part in parts:
if not part or part in {b"--\r\n", b"--"}:
continue
header_section, _, body = part.partition(b"\r\n\r\n")
if not body:
continue
headers = header_section.decode("utf-8", errors="ignore")
if "Content-Disposition" not in headers:
continue
name_marker = "name=\""
name_start = headers.find(name_marker)
name_end = headers.find("\"", name_start + len(name_marker)) if name_start != -1 else -1
field_name = headers[name_start + len(name_marker) : name_end] if name_start != -1 and name_end != -1 else ""
if "filename=" not in headers:
if field_name == "allow_duplicates":
value = body.rstrip(b"\r\n--").decode("utf-8", errors="ignore").strip()
allow_duplicates = value in {"1", "true", "True"}
continue
filename_marker = "filename="
start = headers.find(filename_marker)
end = headers.find("\r\n", start)
filename = headers[start + len(filename_marker) : end].strip().strip('"')
text = body.rstrip(b"\r\n--").decode("utf-8", errors="ignore")
buffered_files.append((filename or "upload", text))
session = self.__class__.session
for filename, text in buffered_files:
processed, metrics = session.ingest_text(
text,
source=filename,
allow_duplicates=allow_duplicates,
)
segments_total += processed
if metrics is not None:
snapshot = ConversationSnapshot(
metrics=metrics,
response="",
meta_report=metrics.meta_report,
todos=tuple(),
generator_description="ingest",
)
self.__class__.state.refresh_from_session(snapshot=snapshot)
files_processed.append(filename)
self.__class__.state.refresh_from_session()
self._send_json({
"segments": segments_total,
"files": files_processed,
"status": f"Ingested {segments_total} segment(s)",
})
def _handle_think(self) -> None:
if self.__class__.state is None or self.__class__.session is None:
self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, "Dashboard not initialized")
return
runtime = self.__class__.session.runtime
runtime.introspection.request_reflection(boost=1.0)
self._send_json({"status": "Reflection queued"})
def _handle_run_gym(self) -> None:
if self.__class__.state is None or self.__class__.session is None:
self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, "Dashboard not initialized")
return
session = self.__class__.session
result = _run_cot_curriculum(session)
self.__class__.state.refresh_from_session()
self._send_json(result | {"status": f"Processed {result.get('examples', 0)} examples"})
def _handle_eat_food(self) -> None:
if self.__class__.state is None or self.__class__.session is None:
self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, "Dashboard not initialized")
return
session = self.__class__.session
result = _ingest_food_corpus(session)
self.__class__.state.refresh_from_session()
self._send_json(result | {"status": "Food corpus digested"})
def _handle_toggle_learning(self) -> None:
session = self.__class__.session
if self.__class__.state is None or session is None:
self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, "Dashboard not initialized")
return
current = bool(getattr(session.config, "learning_enabled", False))
session.config.learning_enabled = not current
status = "Learning enabled" if session.config.learning_enabled else "Learning disabled"
self.__class__.state.refresh_from_session()
self._send_json({"status": status})
def _handle_flush_learning(self) -> None:
session = self.__class__.session
if self.__class__.state is None or session is None:
self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, "Dashboard not initialized")
return
buffered = len(getattr(session, "_learning_buffer", []))
try:
session._maybe_train_on_buffer() # type: ignore[attr-defined]
except Exception as exc: # pragma: no cover - runtime safety
self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, f"Training error: {exc}")
return
self.__class__.state.refresh_from_session()
self._send_json({"status": f"Flushed {buffered} buffered chunk(s)"})
def _handle_training_active(self) -> None:
session = self.__class__.session
if self.__class__.state is None or session is None:
self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, "Dashboard not initialized")
return
runtime = session.runtime
current = bool(getattr(runtime.action_context, "training_active", False))
runtime.set_training_active(not current)
self.__class__.state.refresh_from_session()
status = "Training active" if not current else "Training idle"
self._send_json({"status": status})
def _handle_toggle_filter(self) -> None:
session = self.__class__.session
if self.__class__.state is None or session is None:
self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, "Dashboard not initialized")
return
current = bool(session.is_filter_enabled()) if hasattr(session, "is_filter_enabled") else False
toggle_target = not current
if hasattr(session, "_set_filter_enabled"):
session._set_filter_enabled(toggle_target) # type: ignore[attr-defined]
self.__class__.state.refresh_from_session()
status = "Ingest filter enabled" if toggle_target else "Ingest filter bypassed"
self._send_json({"status": status})
def log_message(self, format: str, *args: object) -> None: # pragma: no cover - silence
return
def _send_html(self, html: str) -> None:
encoded = html.encode("utf-8")
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(encoded)))
self.end_headers()
self.wfile.write(encoded)
def _send_json(self, payload: Dict[str, object]) -> None:
encoded = json.dumps(payload).encode("utf-8")
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Cache-Control", "no-store")
self.send_header("Content-Length", str(len(encoded)))
self.end_headers()
self.wfile.write(encoded)
def run_dashboard(
*,
host: str = "127.0.0.1",
port: int = 8765,
auto_save_path: Optional[str] = None,
checkpoint_path: Optional[str] = "storage/proto_lm/synthetic_baseline.pt",
device: str = "auto",
learning_enabled: bool = False,
archive_checkpoint_on_start: bool = False,
) -> None:
training_config = TrainingConfig()
training_config.device = device
if checkpoint_path is not None:
resolved = Path(checkpoint_path)
if not resolved.is_absolute():
resolved = BASE_DIR / resolved
training_config.checkpoint_path = str(resolved)
session_config = ConversationConfig(
auto_save_path=auto_save_path,
lm=training_config,
learning_enabled=learning_enabled,
archive_checkpoint_on_start=archive_checkpoint_on_start,
)
session = ConversationSession(config=session_config)
run_dashboard_from_session(
session=session,
host=host,
port=port,
)
def run_dashboard_from_session(
*,
session: ConversationSession,
host: str = "127.0.0.1",
port: int = 8765,
) -> None:
state = DashboardState(session)
session_snapshot = session.generate_response("") if session.history else None
if session_snapshot is not None:
state.refresh_from_session(snapshot=session_snapshot)
else:
state.refresh_from_session()
global STATE
STATE = state
DashboardHandler.state = state
DashboardHandler.session = session
unsubscribe = BUS.subscribe(telemetry_sink)
try:
with ThreadingHTTPServer((host, port), DashboardHandler) as server:
server.serve_forever()
except KeyboardInterrupt: # pragma: no cover - interactive
pass
finally:
unsubscribe()
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=8765)
parser.add_argument("--auto-save-path")
parser.add_argument("--checkpoint", default="storage/proto_lm/synthetic_baseline.pt")
parser.add_argument("--device", default="auto")
parser.add_argument("--learn", action="store_true")
parser.add_argument("--archive", action="store_true")
args = parser.parse_args()
run_dashboard(
host=args.host,
port=args.port,
auto_save_path=args.auto_save_path,
checkpoint_path=args.checkpoint,
device=args.device,
learning_enabled=args.learn,
archive_checkpoint_on_start=args.archive,
)
if __name__ == "__main__":
main()