Module neuralops.runners.monitor

NeuralOps Telemetry and Documentation Studio Server package.

Sub-modules

neuralops.runners.monitor.endpoints

HTTP endpoint handlers and flattened routing for NeuralOps monitor server.

neuralops.runners.monitor.rendering

HTML rendering and response serialization utilities for monitor server.

neuralops.runners.monitor.server

HTTP Server and request handler implementation for NeuralOps monitor.

neuralops.runners.monitor.stats

Aggregated statistical calculation and benchmark record loading for monitor server.

neuralops.runners.monitor.telemetry

GPU and execution telemetry collection for NeuralOps monitor server …

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_DATA

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.

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_data

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.

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_index

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.

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 records

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.

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).