Source code for scpn_fusion.core.quantum_bridge

# SPDX-License-Identifier: AGPL-3.0-or-later
# Commercial license available
# © Concepts 1996–2026 Miroslav Šotek. All rights reserved.
# © Code 2020–2026 Miroslav Šotek. All rights reserved.
# ORCID: 0009-0009-3560-0851
# Contact: www.anulum.li | protoscience@anulum.li
# SCPN Fusion Core — Quantum Bridge
"""Runtime integration wrapper for the optional local Quantum Lab scripts."""

from __future__ import annotations

import logging
import math
import subprocess
import sys
from pathlib import Path

logger = logging.getLogger(__name__)

_QUANTUM_SCRIPT_NAMES = (
    "14_quantum_plasma_simulation.py",
    "15_vqe_grad_shafranov.py",
    "16_knm_vqe_fusion.py",
)
_QUANTUM_STEP_LABELS = (
    "[1] Quantum Transport Simulation (Trotterization)",
    "[2] Quantum Equilibrium Solver (VQE)",
    "[3] Physics-Informed Knm-VQE (Topology Ansatz)",
)
_QUANTUM_SCRIPT_TIMEOUT_SECONDS = 1800.0


def _resolve_quantum_lab_path(base_path: str | Path | None = None) -> Path:
    """Resolve the Quantum Lab root directory for suite execution."""
    if base_path is not None:
        return Path(base_path).expanduser().resolve()
    return Path(__file__).resolve().parents[4] / "QUANTUM_LAB"


def _normalize_script_timeout_seconds(timeout_s: float) -> float:
    """Validate and normalise a finite, positive script timeout in seconds."""
    timeout = float(timeout_s)
    if not math.isfinite(timeout) or timeout <= 0.0:
        raise ValueError("script_timeout_seconds must be finite and > 0.")
    return timeout




def run_quantum_suite(
    *,
    base_path: str | Path | None = None,
    script_timeout_seconds: float = _QUANTUM_SCRIPT_TIMEOUT_SECONDS,
) -> dict[str, object]:
    """Execute the staged Quantum Lab workflow scripts in sequence.

    Parameters
    ----------
    base_path
        Optional explicit path to the Quantum Lab directory. If omitted, the
        default repository-local path is used.
    script_timeout_seconds
        Per-script timeout in seconds. Must be finite and strictly positive.

    Returns
    -------
    dict[str, object]
        Execution report with success flag, resolved base path, and script names.

    Raises
    ------
    FileNotFoundError
        If the Quantum Lab directory or required scripts are missing.
    RuntimeError
        If any script fails or times out.
    ValueError
        If the timeout value is invalid.
    """
    logger.info("SCPN quantum fusion bridge start")
    logger.info("Executing staged Quantum Lab plasma-physics scripts")

    timeout_seconds = _normalize_script_timeout_seconds(script_timeout_seconds)
    lab_path = _resolve_quantum_lab_path(base_path)
    if not lab_path.is_dir():
        raise FileNotFoundError(f"Quantum Lab not found at {lab_path}")

    script_paths = [lab_path / name for name in _QUANTUM_SCRIPT_NAMES]
    missing = [p for p in script_paths if not p.is_file()]
    if missing:
        missing_text = ", ".join(p.name for p in missing)
        raise FileNotFoundError(f"Quantum Lab missing required scripts: {missing_text}")

    for step_label, script_path in zip(_QUANTUM_STEP_LABELS, script_paths):
        logger.info("Quantum bridge step: %s", step_label)
        try:
            subprocess.run(
                [sys.executable, str(script_path)],
                check=True,
                timeout=timeout_seconds,
            )
        except subprocess.TimeoutExpired as exc:
            raise RuntimeError(
                f"Quantum script timed out: {script_path.name} (timeout={timeout_seconds:.1f}s)"
            ) from exc
        except subprocess.CalledProcessError as exc:
            raise RuntimeError(
                f"Quantum script failed: {script_path.name} (exit={exc.returncode})"
            ) from exc

    logger.info("Quantum bridge integration complete")
    return {
        "ok": True,
        "base_path": str(lab_path),
        "scripts": [p.name for p in script_paths],
    }



if __name__ == "__main__":
    run_quantum_suite()