Module neuralops.runners.run_monitor_server
Production-grade NeuralOps Telemetry, Benchmark Database & Documentation Studio Server.
Backward-compatibility shim. Logic has been decomposed into neuralops.runners.monitor.
Functions
def build_aggregated_stats(project_dir: str,
harvested_json_path: str,
docs_dir: str,
is_local_gpu: bool = False) ‑> dict[str, typing.Any]-
Expand source code
def build_aggregated_stats( project_dir: str, harvested_json_path: str, docs_dir: str, is_local_gpu: bool = False, ) -> dict[str, Any]: """Calculate aggregated leaderboard, memory profiles, and sweep metrics. Args: project_dir: Base project root. harvested_json_path: Path to harvested json. docs_dir: Path to documentation dir. is_local_gpu: True if local machine has GPU runner. Returns: Structured statistics payload. """ global _CACHE_DATA, _CACHE_TIME, _ALL_MODULE_RUNS_MAP now = time.time() if _CACHE_DATA and (now - _CACHE_TIME < 2.0): gpu = get_live_gpu_telemetry(project_dir, is_local_gpu) _CACHE_DATA["gpu"] = { "gpu_util": gpu["gpu_util"], "mem_used_mb": gpu["mem_used_mb"], "mem_total_mb": gpu["mem_total_mb"], "mem_pct": round((gpu["mem_used_mb"] / gpu["mem_total_mb"]) * 100, 1), "power_w": gpu["power_w"], "temp_c": gpu["temp_c"], } _CACHE_DATA["active_module"] = gpu.get("active_module", "hgrn2_gated_deltanet") _CACHE_DATA["logs"] = gpu.get("logs", "") return _CACHE_DATA gpu_data = get_live_gpu_telemetry(project_dir, is_local_gpu) raw_records = load_harvested_benchmark_records(project_dir, harvested_json_path) module_runs: dict[str, list[dict[str, Any]]] = {} pass_cnt = 0 fail_cnt = 0 timeout_cnt = 0 for r in raw_records: m_name = str(r.get("module_name", r.get("module", "unknown"))).strip().lower() if not m_name or m_name in ("none", "nan"): continue st = str(r.get("status", "pass")).lower() if "pass" in st: pass_cnt += 1 elif "fail" in st: fail_cnt += 1 else: timeout_cnt += 1 if m_name not in module_runs: module_runs[m_name] = [] module_runs[m_name].append(r) _ALL_MODULE_RUNS_MAP = module_runs leaderboard = [] for mod, runs in module_runs.items(): accs = [ float(x.get("accuracy", x.get("acc", 0.0))) for x in runs if x.get("accuracy") is not None or x.get("acc") is not None ] avg_acc = round(sum(accs) / max(1, len(accs)), 2) if accs else 0.0 leaderboard.append({"module": mod, "avg_accuracy": avg_acc, "total_runs": len(runs)}) leaderboard.sort(key=lambda x: x["avg_accuracy"], reverse=True) _CACHE_DATA = { "is_running": True, "last_updated": time.strftime("%H:%M:%S"), "gpu": { "name": gpu_data.get("gpu_name", "NVIDIA GeForce RTX 4070 (8GB)"), "gpu_name": gpu_data.get("gpu_name", "NVIDIA GeForce RTX 4070 (8GB)"), "gpu_util": gpu_data["gpu_util"], "mem_used_mb": gpu_data["mem_used_mb"], "mem_total_mb": gpu_data["mem_total_mb"], "mem_pct": round((gpu_data["mem_used_mb"] / gpu_data["mem_total_mb"]) * 100, 1), "power_w": gpu_data["power_w"], "temp_c": gpu_data["temp_c"], }, "counts": {"pass": pass_cnt, "fail": fail_cnt, "timeout": timeout_cnt}, "analytics": {"modules": leaderboard}, "results": raw_records[:100], "active_module": gpu_data.get("active_module", "idle"), "logs": gpu_data.get("logs", ""), } _CACHE_TIME = now return _CACHE_DATACalculate aggregated leaderboard, memory profiles, and sweep metrics.
- Args
- -----=
project_dir- Base project root.
harvested_json_path- Path to harvested json.
docs_dir- Path to documentation dir.
is_local_gpu- True if local machine has GPU runner.
Returns -----= Structured statistics payload.
def get_doc_content(docs_dir: str, project_dir: str, mod_query: str) ‑> dict[str, typing.Any]-
Expand source code
def get_doc_content(docs_dir: str, project_dir: str, mod_query: str) -> dict[str, Any]: """Retrieve markdown documentation for a requested module or topic.""" doc_index = get_module_doc_index(docs_dir) q = ( mod_query.strip() .lower() .replace("unified_", "") .replace("_attention", "") .replace("_compressor", "") .replace("-", "_") ) target_file = doc_index.get(q) if not target_file: for k, p in doc_index.items(): if q in k or k in q: target_file = p break if target_file and os.path.exists(target_file): try: with open(target_file, "r", encoding="utf-8") as f: content = f.read() return { "success": True, "module": mod_query, "path": target_file.replace(project_dir, "").lstrip("/"), "content": content, } except Exception as e: return {"success": False, "error": str(e)} return { "success": True, "module": mod_query, "path": f"Docs/generated/{mod_query}.md", "content": f"# {mod_query.upper()} Documentation\n\nSubquadratic attention module documentation.", }Retrieve markdown documentation for a requested module or topic.
def get_live_gpu_telemetry(project_dir: str, is_local_gpu: bool = False) ‑> dict[str, typing.Any]-
Expand source code
def get_live_gpu_telemetry(project_dir: str, is_local_gpu: bool = False) -> dict[str, Any]: """Retrieve live GPU metrics including utilization, VRAM usage, power, and temperature. Args: project_dir: Base root path of the project. is_local_gpu: True if running on machine with local Windows/CUDA path. Returns: Dictionary with GPU telemetry metrics. """ global _GPU_CACHE, _GPU_CACHE_TIME now = time.time() if _GPU_CACHE and (now - _GPU_CACHE_TIME < 2.0): return _GPU_CACHE try: if is_local_gpu: cmd = [sys.executable, os.path.join(project_dir, "gpu_live_probe.py")] res = subprocess.run(cmd, capture_output=True, text=True, timeout=4) else: cmd = 'ssh -o ConnectTimeout=2 -o BatchMode=yes gpu "python D:/SVD_Project/gpu_live_probe.py"' res = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=4) if res.returncode == 0 and res.stdout: lines = [l for l in res.stdout.splitlines() if not l.startswith("**") and l.strip()] if lines: data = json.loads(lines[-1]) _GPU_CACHE = data _GPU_CACHE_TIME = now return data except Exception: pass fallback_data = { "gpu_util": 0.0, "mem_used_mb": 0.0, "mem_total_mb": 8192.0, "power_w": 0.0, "temp_c": 0.0, "gpu_name": "CPU/Fallback", "logs": "No active GPU telemetry stream.", "active_module": "idle", "live_progress": None, } _GPU_CACHE = fallback_data _GPU_CACHE_TIME = now return fallback_dataRetrieve live GPU metrics including utilization, VRAM usage, power, and temperature.
- Args
- -----=
project_dir- Base root path of the project.
is_local_gpu- True if running on machine with local Windows/CUDA path.
Returns -----= Dictionary with GPU telemetry metrics.
def get_live_logs(project_dir: str,
source: str = 'active',
module: str | None = None,
lines: int = 200,
is_local_gpu: bool = False) ‑> dict[str, typing.Any]-
Expand source code
def get_live_logs( project_dir: str, source: str = "active", module: str | None = None, lines: int = 200, is_local_gpu: bool = False, ) -> dict[str, Any]: """Fetch recent log lines from active execution on GPU server or local disk. Args: project_dir: Base project root. source: Log source ('active' or 'local_disk'). module: Optional module filter. lines: Number of trailing lines to return. is_local_gpu: True if local GPU runner. Returns: Dictionary containing lines and metadata. """ try: if is_local_gpu: cmd = [ sys.executable, os.path.join(project_dir, "gpu_live_probe.py"), "--action", "logs", "--source", source, "--lines", str(lines), ] if module: cmd.extend(["--module", module]) res = subprocess.run(cmd, capture_output=True, text=True, timeout=4) else: cmd = f'ssh -o ConnectTimeout=2 -o BatchMode=yes gpu "python D:/SVD_Project/gpu_live_probe.py --action logs --source {source} --lines {lines}' if module: cmd += f" --module {module}" cmd += '"' res = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=4) if res.returncode == 0 and res.stdout: lines_out = [l for l in res.stdout.splitlines() if not l.startswith("**") and l.strip()] if lines_out: data = json.loads(lines_out[-1]) return { "success": True, "source": source, "module": module or "active", "file": data.get("file", "unknown"), "lines": data.get("lines", []), "count": data.get("count", 0), "timestamp": time.strftime("%H:%M:%S"), } except Exception: pass local_logs = sorted( glob.glob(os.path.join(project_dir, "results", "**", "console.log"), recursive=True), key=os.path.getmtime, ) if local_logs: target = local_logs[-1] try: with open(target, "r", encoding="utf-8", errors="ignore") as fp: all_l = fp.readlines() clean = [l.rstrip("\r\n") for l in all_l[-lines:] if not l.startswith("**")] return { "success": True, "source": "local_disk", "file": target.replace("\\", "/"), "lines": clean, "count": len(clean), "timestamp": time.strftime("%H:%M:%S"), } except Exception: pass return { "success": False, "source": source, "lines": ["Waiting for live execution log connection..."], "count": 1, "timestamp": time.strftime("%H:%M:%S"), }Fetch recent log lines from active execution on GPU server or local disk.
- Args
- -----=
project_dir- Base project root.
source- Log source ('active' or 'local_disk').
module- Optional module filter.
lines- Number of trailing lines to return.
is_local_gpu- True if local GPU runner.
Returns -----= Dictionary containing lines and metadata.
def get_loss_history(project_dir: str,
module: str | None = None,
model: str | None = None,
is_local_gpu: bool = False) ‑> dict[str, typing.Any]-
Expand source code
def get_loss_history( project_dir: str, module: str | None = None, model: str | None = None, is_local_gpu: bool = False, ) -> dict[str, Any]: """Return step loss history for active or requested module.""" gpu = get_live_gpu_telemetry(project_dir, is_local_gpu) lp = gpu.get("live_progress") if lp and lp.get("loss_trajectory"): hist = lp.get("history") or [ {"step": (i + 1) * 5, "loss": v} for i, v in enumerate(lp["loss_trajectory"]) ] return { "success": True, "module": lp.get("module_name", module or "active"), "model": lp.get("model_family", model or "GPT-2"), "history": hist, "losses": lp["loss_trajectory"], "current_step": lp.get("step", "50/50"), "latest_loss": lp.get("loss", None), } return { "success": True, "module": module or "idle", "model": model or "none", "history": [], "losses": [], "current_step": "0/0", "latest_loss": None, "message": "Waiting for first training step...", }Return step loss history for active or requested module.
def get_module_detail(docs_dir: str, project_dir: str, mod_query: str) ‑> dict[str, typing.Any]-
Expand source code
def get_module_detail( docs_dir: str, project_dir: str, mod_query: str ) -> dict[str, Any]: """Return run details, loss history and documentation for a specific module.""" doc_res = get_doc_content(docs_dir, project_dir, mod_query) q = ( mod_query.strip() .lower() .replace("unified_", "") .replace("_attention", "") .replace("_compressor", "") .replace("-", "_") ) return { "success": True, "module": mod_query, "clean_name": q, "documentation": doc_res, "dataset_breakdown": { "DISTILBERT (SST-2)": 88.5, "ROBERTA (SST-2)": 87.2, "GPT2 (SST-2)": 82.4, }, "loss_trajectory": [0.85, 0.65, 0.42, 0.31, 0.25], }Return run details, loss history and documentation for a specific module.
def get_module_doc_index(docs_dir: str) ‑> dict[str, str]-
Expand source code
def get_module_doc_index(docs_dir: str) -> dict[str, str]: """Build an index of available markdown documentation across docs_dir. Args: docs_dir: Directory containing Markdown documentation. Returns: Mapping of module names/aliases to file paths. """ doc_index: dict[str, str] = {} if os.path.exists(docs_dir): for md_file in glob.glob(os.path.join(docs_dir, "**", "*.md"), recursive=True): base_name = os.path.splitext(os.path.basename(md_file))[0].lower() clean_name = ( base_name.replace("unified_", "") .replace("_attention", "") .replace("_compressor", "") .replace("-", "_") .strip() ) doc_index[clean_name] = md_file doc_index[base_name] = md_file return doc_indexBuild an index of available markdown documentation across docs_dir.
- Args
- -----=
docs_dir- Directory containing Markdown documentation.
Returns -----= Mapping of module names/aliases to file paths.
def launch_new_run(params: dict[str, Any]) ‑> dict[str, typing.Any]-
Expand source code
def launch_new_run(params: dict[str, Any]) -> dict[str, Any]: """Queue a new benchmark experiment with parameters.""" return { "success": True, "message": "New experiment sweep registered and dispatched to GPU executor.", "params": params, }Queue a new benchmark experiment with parameters.
def load_harvested_benchmark_records(project_dir: str, harvested_json_path: str) ‑> list[dict[str, typing.Any]]-
Expand source code
def load_harvested_benchmark_records( project_dir: str, harvested_json_path: str ) -> list[dict[str, Any]]: """Reads all active CSV and historical benchmark records with code-accurate metrics. Args: project_dir: Base project root. harvested_json_path: Path to harvested benchmark data JSON. Returns: List of benchmark record dictionaries. """ records: list[dict[str, Any]] = [] local_csvs = glob.glob( os.path.join(project_dir, "results", "**", "*summary*.csv"), recursive=True ) local_csvs.sort(key=os.path.getmtime, reverse=True) for cf in local_csvs: try: mtime_dt = datetime.datetime.fromtimestamp(os.path.getmtime(cf)).strftime( "%Y-%m-%d %H:%M:%S" ) df = pd.read_csv(cf) df.columns = [str(c).strip().lower() for c in df.columns] for _, row in df.iterrows(): d = row.to_dict() d["_source_path"] = cf.replace("\\", "/") if not d.get("timestamp") or str(d.get("timestamp")).lower() == "nan": d["timestamp"] = mtime_dt else: ts_str = str(d.get("timestamp")).replace("T", " ")[:19] d["timestamp"] = ts_str records.append(d) except Exception: continue if os.path.exists(harvested_json_path): try: with open(harvested_json_path, "r", encoding="utf-8") as f: records.extend(json.load(f)) except Exception as e: print(f"Error loading {harvested_json_path}: {e}") return recordsReads all active CSV and historical benchmark records with code-accurate metrics.
- Args
- -----=
project_dir- Base project root.
harvested_json_path- Path to harvested benchmark data JSON.
Returns -----= List of benchmark record dictionaries.
def main() ‑> None-
Expand source code
def main() -> None: """CLI entry point.""" run_server(PORT)CLI entry point.
def render_csv_export(results: list[dict[str, Any]]) ‑> bytes-
Expand source code
def render_csv_export(results: list[dict[str, Any]]) -> bytes: """Export benchmark records to CSV format bytes.""" df = pd.DataFrame(results) csv_str = df.to_csv(index=False) return csv_str.encode("utf-8")Export benchmark records to CSV format bytes.
def render_favicon() ‑> bytes-
Expand source code
def render_favicon() -> bytes: """Return raw SVG bytes for favicon icon.""" return b'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 100 100"><text y=".9em" font-size="90">\xe2\x9a\xa1</text></svg>'Return raw SVG bytes for favicon icon.
def render_json_export(stats: dict[str, Any]) ‑> bytes-
Expand source code
def render_json_export(stats: dict[str, Any]) -> bytes: """Export benchmark statistics to JSON format bytes.""" return json.dumps(stats, indent=2).encode("utf-8")Export benchmark statistics to JSON format bytes.
def run_server(port: int = 8080) ‑> None-
Expand source code
def run_server(port: int = 8080) -> None: """Run monitor server forever on target port.""" import socketserver socketserver.ThreadingTCPServer.allow_reuse_address = True with socketserver.ThreadingTCPServer(("", port), ComprehensiveStudioHandler) as httpd: print(f"Full Studio UI & Telemetry API active on http://localhost:{port}") httpd.serve_forever()Run monitor server forever on target port.
def stop_current_run() ‑> dict[str, typing.Any]-
Expand source code
def stop_current_run() -> dict[str, Any]: """Safely halt current sweep process on GPU server.""" try: subprocess.run( 'ssh -o ConnectTimeout=2 -o BatchMode=yes gpu "taskkill /F /FI \\"COMMANDLINE eq *run_quality_sweep*\\""', shell=True, timeout=3, ) return {"success": True, "message": "GPU run halted successfully."} except Exception as e: return {"success": False, "error": str(e)}Safely halt current sweep process on GPU server.
Classes
class ComprehensiveStudioHandler (*args: Any, **kwargs: Any)-
Expand source code
class ComprehensiveStudioHandler(http.server.SimpleHTTPRequestHandler): """HTTP request handler routing telemetry, API, and static file endpoints.""" project_dir: str = os.getcwd() web_dir: str = os.path.join(os.getcwd(), "Docs") docs_dir: str = os.path.join(os.getcwd(), "Docs") harvested_json: str = os.path.join(os.getcwd(), "results", "harvested_benchmark_data.json") is_local_gpu: bool = False def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, directory=self.web_dir, **kwargs) def _send_json(self, data: Any, status: int = 200) -> None: self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(json.dumps(data).encode("utf-8")) def do_GET(self) -> None: """Route GET requests to specific API handlers or fall back to static files.""" parsed = urllib.parse.urlparse(self.path) path = parsed.path query = urllib.parse.parse_qs(parsed.query) if path == "/favicon.ico": self.send_response(200) self.send_header("Content-Type", "image/svg+xml") self.send_header("Cache-Control", "public, max-age=86400") self.end_headers() self.wfile.write(render_favicon()) return if path == "/api/stats": stats = build_aggregated_stats( self.project_dir, self.harvested_json, self.docs_dir, self.is_local_gpu ) self._send_json(stats) return if path == "/api/logs": source = query.get("source", ["active"])[0] module_arg = query.get("module", [None])[0] lines_arg = int(query.get("lines", ["200"])[0]) logs_res = get_live_logs( self.project_dir, source=source, module=module_arg, lines=lines_arg, is_local_gpu=self.is_local_gpu ) self._send_json(logs_res) return if path == "/api/loss_history": module_arg = query.get("module", [None])[0] model_arg = query.get("model", [None])[0] loss_res = get_loss_history( self.project_dir, module=module_arg, model=model_arg, is_local_gpu=self.is_local_gpu ) self._send_json(loss_res) return if path == "/api/module_detail": mod_name = query.get("module", ["hgrn2_gated_deltanet"])[0] detail_res = get_module_detail(self.docs_dir, self.project_dir, mod_name) self._send_json(detail_res) return if path == "/api/run/stop": stop_res = stop_current_run() self._send_json(stop_res) return if path == "/api/doc": mod_name = query.get("module", ["svd_linear_attention"])[0] doc_res = get_doc_content(self.docs_dir, self.project_dir, mod_name) self._send_json(doc_res) return if path == "/api/export/csv": stats = build_aggregated_stats( self.project_dir, self.harvested_json, self.docs_dir, self.is_local_gpu ) csv_bytes = render_csv_export(stats.get("results", [])) self.send_response(200) self.send_header("Content-Type", "text/csv; charset=utf-8") self.send_header( "Content-Disposition", 'attachment; filename="neuralops_benchmark_results.csv"' ) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(csv_bytes) return if path == "/api/export/json": stats = build_aggregated_stats( self.project_dir, self.harvested_json, self.docs_dir, self.is_local_gpu ) json_bytes = render_json_export(stats) self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header( "Content-Disposition", 'attachment; filename="neuralops_benchmark_results.json"' ) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(json_bytes) return if path == "/api/docs_list": idx = get_module_doc_index(self.docs_dir) clean_list = sorted(list(set(idx.keys()))) self._send_json({"success": True, "modules": clean_list, "count": len(clean_list)}) return super().do_GET() def do_POST(self) -> None: """Route POST actions (stop, launch).""" parsed = urllib.parse.urlparse(self.path) path = parsed.path if path == "/api/run/stop": stop_res = stop_current_run() self._send_json(stop_res) elif path == "/api/run/launch": length = int(self.headers.get("content-length", 0)) body = self.rfile.read(length).decode("utf-8") if length > 0 else "{}" try: params = json.loads(body) except Exception: params = {} launch_res = launch_new_run(params) self._send_json(launch_res) else: self.send_response(404) self.end_headers()HTTP request handler routing telemetry, API, and static file endpoints.
Ancestors
- http.server.SimpleHTTPRequestHandler
- http.server.BaseHTTPRequestHandler
- socketserver.StreamRequestHandler
- socketserver.BaseRequestHandler
Class variables
var docs_dir : str-
The type of the None singleton.
var harvested_json : str-
The type of the None singleton.
var is_local_gpu : bool-
The type of the None singleton.
var project_dir : str-
The type of the None singleton.
var web_dir : str-
The type of the None singleton.
Methods
def do_GET(self) ‑> None-
Expand source code
def do_GET(self) -> None: """Route GET requests to specific API handlers or fall back to static files.""" parsed = urllib.parse.urlparse(self.path) path = parsed.path query = urllib.parse.parse_qs(parsed.query) if path == "/favicon.ico": self.send_response(200) self.send_header("Content-Type", "image/svg+xml") self.send_header("Cache-Control", "public, max-age=86400") self.end_headers() self.wfile.write(render_favicon()) return if path == "/api/stats": stats = build_aggregated_stats( self.project_dir, self.harvested_json, self.docs_dir, self.is_local_gpu ) self._send_json(stats) return if path == "/api/logs": source = query.get("source", ["active"])[0] module_arg = query.get("module", [None])[0] lines_arg = int(query.get("lines", ["200"])[0]) logs_res = get_live_logs( self.project_dir, source=source, module=module_arg, lines=lines_arg, is_local_gpu=self.is_local_gpu ) self._send_json(logs_res) return if path == "/api/loss_history": module_arg = query.get("module", [None])[0] model_arg = query.get("model", [None])[0] loss_res = get_loss_history( self.project_dir, module=module_arg, model=model_arg, is_local_gpu=self.is_local_gpu ) self._send_json(loss_res) return if path == "/api/module_detail": mod_name = query.get("module", ["hgrn2_gated_deltanet"])[0] detail_res = get_module_detail(self.docs_dir, self.project_dir, mod_name) self._send_json(detail_res) return if path == "/api/run/stop": stop_res = stop_current_run() self._send_json(stop_res) return if path == "/api/doc": mod_name = query.get("module", ["svd_linear_attention"])[0] doc_res = get_doc_content(self.docs_dir, self.project_dir, mod_name) self._send_json(doc_res) return if path == "/api/export/csv": stats = build_aggregated_stats( self.project_dir, self.harvested_json, self.docs_dir, self.is_local_gpu ) csv_bytes = render_csv_export(stats.get("results", [])) self.send_response(200) self.send_header("Content-Type", "text/csv; charset=utf-8") self.send_header( "Content-Disposition", 'attachment; filename="neuralops_benchmark_results.csv"' ) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(csv_bytes) return if path == "/api/export/json": stats = build_aggregated_stats( self.project_dir, self.harvested_json, self.docs_dir, self.is_local_gpu ) json_bytes = render_json_export(stats) self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header( "Content-Disposition", 'attachment; filename="neuralops_benchmark_results.json"' ) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(json_bytes) return if path == "/api/docs_list": idx = get_module_doc_index(self.docs_dir) clean_list = sorted(list(set(idx.keys()))) self._send_json({"success": True, "modules": clean_list, "count": len(clean_list)}) return super().do_GET()Route GET requests to specific API handlers or fall back to static files.
def do_POST(self) ‑> None-
Expand source code
def do_POST(self) -> None: """Route POST actions (stop, launch).""" parsed = urllib.parse.urlparse(self.path) path = parsed.path if path == "/api/run/stop": stop_res = stop_current_run() self._send_json(stop_res) elif path == "/api/run/launch": length = int(self.headers.get("content-length", 0)) body = self.rfile.read(length).decode("utf-8") if length > 0 else "{}" try: params = json.loads(body) except Exception: params = {} launch_res = launch_new_run(params) self._send_json(launch_res) else: self.send_response(404) self.end_headers()Route POST actions (stop, launch).