feat: introduce Node.js runtime management and Pydantic schemas for Next.js dashboard data serialization.

Author: KennyGraham1

floatcsep/postprocess/nextjs/runtime.py

@@ -0,0 +1,188 @@
+"""Node.js runtime management for floatCSEP Next.js dashboard."""
+
+import logging
+import os
+import platform
+import re
+import shutil
+import stat
+import subprocess
+import tarfile
+import zipfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import List, Optional
+from urllib import request
+
+logger = logging.getLogger(__name__)
+
+MIN_NODE_VERSION = (18, 17, 0)
+BUNDLED_NODE_VERSION = "20.11.1"
+
+
+@dataclass
+class NodeRuntime:
+    """Represents a runnable Node.js installation."""
+
+    node_path: Path
+    npm_path: Path
+    bin_dir: Path
+    source: str
+
+    def apply_to_env(self, env: dict) -> dict:
+        """Return a copy of the environment with this runtime prepended to PATH."""
+
+        current_path = env.get("PATH", "")
+        updated = env.copy()
+        updated["PATH"] = (
+            f"{self.bin_dir}{os.pathsep}{current_path}"
+            if current_path
+            else str(self.bin_dir)
+        )
+        return updated
+
+
+def parse_node_version(raw: str) -> Optional[tuple[int, int, int]]:
+    match = re.match(r"v?(\d+)\.(\d+)\.(\d+)", raw.strip())
+    if not match:
+        return None
+    return tuple(int(part) for part in match.groups())
+
+
+def get_system_node_runtime() -> Optional[NodeRuntime]:
+    node_cmd = shutil.which("node")
+    npm_cmd = shutil.which("npm")
+    if not node_cmd or not npm_cmd:
+        return None
+    try:
+        result = subprocess.run(
+            [node_cmd, "--version"], capture_output=True, text=True, check=True
+        )
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        return None
+    version = parse_node_version(result.stdout)
+    if not version or version < MIN_NODE_VERSION:
+        return None
+    return NodeRuntime(
+        node_path=Path(node_cmd),
+        npm_path=Path(npm_cmd),
+        bin_dir=Path(node_cmd).parent,
+        source="system",
+    )
+
+
+def _node_dist_name() -> tuple[str, str]:
+    system = platform.system().lower()
+    machine = platform.machine().lower()
+    if system == "linux":
+        if machine in ("x86_64", "amd64"):
+            return "linux-x64", ".tar.xz"
+        if machine in ("aarch64", "arm64"):
+            return "linux-arm64", ".tar.xz"
+    elif system == "darwin":
+        if machine == "arm64":
+            return "darwin-arm64", ".tar.xz"
+        if machine in ("x86_64", "amd64"):
+            return "darwin-x64", ".tar.xz"
+    elif system == "windows":
+        if machine in ("x86_64", "amd64"):
+            return "win-x64", ".zip"
+    raise RuntimeError(
+        f"Unsupported platform '{platform.system()} {platform.machine()}'. "
+        "Please install Node.js 20+ manually."
+    )
+
+
+def _download_node_archive(target: Path, url: str) -> None:
+    logger.info("Downloading Node.js runtime from %s", url)
+    target.parent.mkdir(parents=True, exist_ok=True)
+    with request.urlopen(url) as response, open(target, "wb") as handle:
+        shutil.copyfileobj(response, handle)
+
+
+def _extract_node_archive(archive: Path, destination: Path) -> Path:
+    logger.info("Extracting Node.js runtime to %s", destination)
+    destination.mkdir(parents=True, exist_ok=True)
+    if archive.suffix == ".zip":
+        with zipfile.ZipFile(archive) as zf:
+            zf.extractall(destination)
+    else:
+        # Handles .tar.xz
+        with tarfile.open(archive, mode="r:*") as tf:
+            tf.extractall(destination)
+    # Find the extracted directory (node-vXX-<platform>)
+    for child in destination.iterdir():
+        if child.is_dir() and child.name.startswith(f"node-v{BUNDLED_NODE_VERSION}"):
+            return child
+    raise RuntimeError("Failed to locate extracted Node.js runtime")
+
+
+def ensure_bundled_node(nextjs_dir: Path) -> NodeRuntime:
+    platform_tag, archive_ext = _node_dist_name()
+    cache_dir = nextjs_dir / ".cache" / "node-runtime"
+    extract_root = cache_dir / f"node-v{BUNDLED_NODE_VERSION}-{platform_tag}"
+    if extract_root.exists():
+        logger.info("Using cached Node.js runtime at %s", extract_root)
+    else:
+        archive_name = f"node-v{BUNDLED_NODE_VERSION}-{platform_tag}{archive_ext}"
+        download_url = (
+            f"https://nodejs.org/dist/v{BUNDLED_NODE_VERSION}/{archive_name}"
+        )
+        archive_path = cache_dir / archive_name
+        _download_node_archive(archive_path, download_url)
+        extracted = _extract_node_archive(archive_path, cache_dir)
+        extracted.rename(extract_root)
+        archive_path.unlink(missing_ok=True)
+
+    if platform.system().lower() == "windows":
+        node_path = extract_root / "node.exe"
+        npm_path = extract_root / "npm.cmd"
+        bin_dir = extract_root
+    else:
+        bin_dir = extract_root / "bin"
+        node_path = bin_dir / "node"
+        npm_path = bin_dir / "npm"
+    for path in (node_path, npm_path):
+        if not path.exists():
+            raise RuntimeError(f"Bundled Node.js binary missing: {path}")
+        mode = path.stat().st_mode
+        path.chmod(mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
+    return NodeRuntime(
+        node_path=node_path, npm_path=npm_path, bin_dir=bin_dir, source="bundled"
+    )
+
+
+def ensure_node_runtime(nextjs_dir: Path) -> NodeRuntime:
+    runtime = get_system_node_runtime()
+    if runtime:
+        logger.info("Detected Node.js %s from system PATH", runtime.node_path)
+        return runtime
+    logger.warning(
+        "Node.js %s or newer not found. Downloading a scoped runtime (v%s).",
+        ".".join(str(part) for part in MIN_NODE_VERSION),
+        BUNDLED_NODE_VERSION,
+    )
+    return ensure_bundled_node(nextjs_dir)
+
+
+def ensure_nextjs_dependencies(
+    nextjs_dir: Path, npm_cmd: List[str], env: dict
+) -> None:
+    """Install Node dependencies if needed."""
+    node_modules = nextjs_dir / "node_modules"
+    if node_modules.exists():
+        return
+    logger.info("Installing Next.js dependencies (this may take a few minutes)...")
+    try:
+        subprocess.run(
+            npm_cmd + ["install"],
+            cwd=nextjs_dir,
+            check=True,
+            env=env,
+        )
+    except subprocess.CalledProcessError as exc:
+        logger.error("Failed to install dependencies: %s", exc)
+        raise RuntimeError(
+            "Could not install Next.js dependencies automatically. "
+            "Please ensure network access is available or install them manually."
+        )

floatcsep/postprocess/nextjs/schemas.py

@@ -0,0 +1,131 @@
+"""Pydantic schemas for floatCSEP Next.js dashboard."""
+
+from typing import Any, Dict, List, Optional, Tuple
+from pathlib import Path
+from pydantic import BaseModel, ConfigDict, field_validator, Field
+
+
+def serialize_value_recursive(value: Any) -> Any:
+    """Recursively convert values to JSON-serializable types."""
+    if isinstance(value, Path):
+        return str(value)
+    elif isinstance(value, dict):
+        return {k: serialize_value_recursive(v) for k, v in value.items()}
+    elif isinstance(value, (list, tuple)):
+        return [serialize_value_recursive(v) for v in value]
+    elif hasattr(value, "__dict__"):
+        return serialize_value_recursive(vars(value))
+    else:
+        return value
+
+
+class ManifestModel(BaseModel):
+    """
+    Pydantic model for the Experiment Manifest.
+    Validates and serializes the Manifest dataclass from floatcsep.
+    """
+
+    model_config = ConfigDict(from_attributes=True, populate_by_name=True)
+
+    # --- Existing fields ---
+    name: str
+    start_date: str
+    end_date: str
+    authors: Optional[str] = None
+    doi: Optional[str] = None
+    journal: Optional[str] = None
+    manuscript_doi: Optional[str] = None
+    exp_time: Optional[str] = None
+    floatcsep_version: Optional[str] = None
+    pycsep_version: Optional[str] = None
+    last_run: Optional[str] = None
+    catalog_doi: Optional[str] = None
+    license: Optional[str] = None
+    date_range: str
+    magnitudes: List[float]
+    
+    # Region is typically an object in the dataclass
+    region: Optional[Dict[str, Any]] = None
+
+    models: List[Dict[str, Any]]
+    tests: List[Dict[str, Any]]
+    time_windows: List[str]
+
+    catalog: Dict[str, Any]
+    results_main: Dict[str, str]  # Key will be converted to string pipe-delimited
+    results_model: Dict[str, str]
+
+    app_root: Optional[str] = None
+
+    # --- Metadata fields ---
+    exp_class: str
+    n_intervals: int
+    horizon: Optional[str] = None
+    offset: Optional[str] = None
+    growth: Optional[str] = None
+
+    mag_min: Optional[float] = None
+    mag_max: Optional[float] = None
+    mag_bin: Optional[float] = None
+    depth_min: Optional[float] = None
+    depth_max: Optional[float] = None
+
+    run_mode: Optional[str] = None
+    run_dir: Optional[str] = None
+    config_file: Optional[str] = None
+    # Rename to avoid conflict with Pydantic's model_config
+    model_config_path: Optional[str] = Field(None, validation_alias="model_config", serialization_alias="model_config")
+    test_config: Optional[str] = None
+
+    @field_validator("region", mode="before")
+    def serialize_region(cls, v: Any) -> Optional[Dict[str, Any]]:
+        if v is None:
+            return None
+        if isinstance(v, dict):
+            return v
+        # Attempt to extract attributes from Region object
+        return {
+            "name": getattr(v, "name", None),
+            "bbox": list(v.get_bbox()) if hasattr(v, "get_bbox") else None,
+            "dh": float(v.dh) if hasattr(v, "dh") else None,
+            "origins": v.origins().tolist() if hasattr(v, "origins") else None,
+        }
+
+    @field_validator("models", "tests", "catalog", mode="before")
+    def serialize_generic_structures(cls, v: Any) -> Any:
+        return serialize_value_recursive(v)
+    
+    @field_validator("results_main", mode="before")
+    def serialize_results_main(cls, v: Any) -> Dict[str, str]:
+        # transform Dict[Tuple[str, str], str] -> Dict[str, str]
+        if isinstance(v, dict):
+            new_dict = {}
+            for key, val in v.items():
+                if isinstance(key, tuple):
+                    new_key = f"{key[0]}|{key[1]}"
+                else:
+                    new_key = str(key)
+                new_dict[new_key] = serialize_value_recursive(val)
+            return new_dict
+        return v
+
+    @field_validator("results_model", mode="before")
+    def serialize_results_model(cls, v: Any) -> Dict[str, str]:
+        # transform Dict[Tuple[str, str, str], str] -> Dict[str, str]
+        if isinstance(v, dict):
+            new_dict = {}
+            for key, val in v.items():
+                if isinstance(key, tuple):
+                    new_key = f"{key[0]}|{key[1]}|{key[2]}"
+                else:
+                    new_key = str(key)
+                new_dict[new_key] = serialize_value_recursive(val)
+            return new_dict
+        return v
+
+    @field_validator("app_root", "run_dir", "config_file", "model_config_path", "test_config", mode="before")
+    def serialize_paths(cls, v: Any) -> Optional[str]:
+        if isinstance(v, Path):
+            return str(v)
+        return v
+