backend/core/alphageometry_adapter.py

"""
AlphaGeometry Adapter: Run AlphaGeometry proofs from Python with advanced features.
Features:
- Async execution, timeouts, resource limits
- Logging, error handling, compliance
- Batch/parallel runs, result parsing, provenance
- Plugin system, benchmarking, test harness
"""

import subprocess
import os
import asyncio
import concurrent.futures
import logging
import time
from typing import List, Optional, Callable, Dict, Any

class AlphaGeometryResult:
    def __init__(self, output: str, success: bool, elapsed: float, provenance: Optional[Dict[str, Any]] = None):
        self.output: str = output
        self.success: bool = success
        self.elapsed: float = elapsed
        self.provenance: Dict[str, Any] = provenance or {}

    def parse(self) -> Dict[str, Any]:
        # Example: parse output for key results (stub)
        lines: List[str] = self.output.splitlines()
        result: Dict[str, Any] = {"lines": lines, "success": self.success, "elapsed": self.elapsed}
        if any("QED" in l for l in lines):
            result["proved"] = True
        return result

def run_alphageometry(
    input_file: str,
    alphageometry_dir: str = "external/alphageometry",
    timeout: int = 60,
    plugins: Optional[List[Callable[[AlphaGeometryResult], None]]] = None
) -> AlphaGeometryResult:
    """
    Runs AlphaGeometry on the given input file and returns a structured result.
    """
    exe_path = os.path.join(alphageometry_dir, "main.py")
    if not os.path.exists(exe_path):
        raise FileNotFoundError(f"AlphaGeometry not found at {exe_path}")
    start = time.time()
    try:
        result = subprocess.run([
            "python", exe_path, input_file
        ], capture_output=True, text=True, check=True, timeout=timeout)
        elapsed = time.time() - start
        ag_result = AlphaGeometryResult(result.stdout, True, elapsed)
    except subprocess.TimeoutExpired as e:
        logging.error(f"AlphaGeometry timeout: {e}")
        ag_result = AlphaGeometryResult(f"Timeout: {e}", False, timeout)
    except Exception as e:
        logging.error(f"AlphaGeometry error: {e}", exc_info=True)
        ag_result = AlphaGeometryResult(f"AlphaGeometry error: {e}", False, time.time() - start)
    # Plugin post-processing
    if plugins:
        for plugin in plugins:
            plugin(ag_result)
    return ag_result

async def run_alphageometry_async(
    input_file: str,
    alphageometry_dir: str = "external/alphageometry",
    timeout: int = 60
) -> AlphaGeometryResult:
    loop = asyncio.get_event_loop()
    with concurrent.futures.ThreadPoolExecutor() as pool:
        return await loop.run_in_executor(pool, run_alphageometry, input_file, alphageometry_dir, timeout)

def run_alphageometry_batch(
    input_files: List[str],
    alphageometry_dir: str = "external/alphageometry",
    timeout: int = 60,
    parallel: int = 4
) -> List[AlphaGeometryResult]:
    """Run AlphaGeometry on a batch of input files in parallel."""
    with concurrent.futures.ThreadPoolExecutor(max_workers=parallel) as executor:
        futures: List[concurrent.futures.Future[AlphaGeometryResult]] = [executor.submit(run_alphageometry, f, alphageometry_dir, timeout) for f in input_files]
        return [f.result() for f in futures]

def benchmark_alphageometry(
    input_file: str,
    alphageometry_dir: str = "external/alphageometry",
    n_iter: int = 5
) -> None:
    times: List[float] = []
    for _ in range(n_iter):
        start = time.time()
        _ = run_alphageometry(input_file, alphageometry_dir)
        times.append(float(time.time() - start))
    if times:
        mean: float = sum(times) / len(times)
        std: float = float((sum((t - mean) ** 2 for t in times) / len(times)) ** 0.5)
        print(f"[Benchmark] Mean: {mean:.4f}s, Std: {std:.4f}s")
    else:
        print("[Benchmark] No runs completed.")

# --- Plugin Example ---
class QEDPlugin:
    def __call__(self, result: AlphaGeometryResult) -> None:
        if "QED" in result.output:
            result.provenance["proved"] = True

# --- Test Harness ---
def test_alphageometry_adapter() -> None:
    # Dummy test: expects a dummy input file and AlphaGeometry stub
    input_file = "dummy_input.txt"
    with open(input_file, "w") as f:
        f.write("A B C = triangle A B C\n")
    result = run_alphageometry(input_file, timeout=2, plugins=[QEDPlugin()])
    print("Result:", result.parse())
    os.remove(input_file)

if __name__ == "__main__":
    test_alphageometry_adapter()