Module neuralops.runners.monitor.endpoints

HTTP endpoint handlers and flattened routing for NeuralOps monitor server.

Functions

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