| """Solver adapter contract (decision 44, section 7 'Solver contract'). |
| |
| Every adapter declares supported elements, system sizes, boundary conditions, observables, units, |
| and known limitations; records engine/version, input deck, parameters, seeds, logs, exit code and |
| output checksums; makes convergence explicit; and returns NOT_SUPPORTED / NEEDS_METHOD rather than |
| inventing a value. |
| """ |
| from __future__ import annotations |
|
|
| import time |
| from abc import ABC, abstractmethod |
| from dataclasses import dataclass, field |
| from pathlib import Path |
| from typing import Any, Callable, Optional |
|
|
| from core.errors import ErrorCode |
| from core.representation import MatterStructure |
|
|
|
|
| class CancelledByUser(Exception): |
| pass |
|
|
|
|
| @dataclass |
| class SolverCapabilities: |
| solver_id: str |
| family: str |
| engine: str |
| engine_version: str |
| supported_elements: Optional[set[str]] |
| max_atoms: int |
| boundary_conditions: list[str] |
| observables: list[str] |
| units: dict[str, str] |
| limitations: list[str] |
| status: str = "implemented" |
| cost_model: str = "" |
|
|
| def to_dict(self) -> dict: |
| d = dict(self.__dict__) |
| d["supported_elements"] = None if self.supported_elements is None else sorted(self.supported_elements) |
| return d |
|
|
|
|
| @dataclass |
| class RunContext: |
| workdir: Path |
| seed: int = 0 |
| wall_time_s: float = 120.0 |
| checkpoint_path: Optional[Path] = None |
| checkpoint_every: int = 5 |
| cancel_check: Callable[[], bool] = lambda: False |
| started: float = field(default_factory=time.time) |
|
|
| def check(self): |
| if self.cancel_check(): |
| raise CancelledByUser() |
| if time.time() - self.started > self.wall_time_s: |
| raise TimeoutError("wall time exceeded inside solver") |
|
|
|
|
| @dataclass |
| class SolverOutput: |
| status: str |
| error_code: Optional[str] = None |
| observables: dict[str, Any] = field(default_factory=dict) |
| convergence: dict[str, Any] = field(default_factory=lambda: {"status": "unknown"}) |
| final_structure: Optional[MatterStructure] = None |
| logs: str = "" |
| input_deck: str = "" |
| artifacts: list[tuple[str, bytes, str]] = field(default_factory=list) |
| uncertainty: list[dict] = field(default_factory=list) |
| electronic: dict[str, Any] = field(default_factory=dict) |
| sanity: dict[str, Any] = field(default_factory=dict) |
| resumed_from_checkpoint: bool = False |
|
|
|
|
| def not_supported(reason: str, code: ErrorCode = ErrorCode.UNSUPPORTED_METHOD) -> SolverOutput: |
| return SolverOutput(status="NOT_SUPPORTED", error_code=code.value, logs=reason, |
| convergence={"status": "not_applicable", "reason": reason}) |
|
|
|
|
| class SolverAdapter(ABC): |
| solver_id: str = "abstract" |
| family: str = "abstract" |
|
|
| def __init__(self, parameter_set: dict | None = None): |
| self.parameter_set = parameter_set or {} |
|
|
| @abstractmethod |
| def capabilities(self) -> SolverCapabilities: ... |
|
|
| def available(self) -> tuple[bool, str]: |
| return True, "in-process python engine" |
|
|
| def compatible(self, ms: MatterStructure, observable: str) -> tuple[bool, list[str]]: |
| cap = self.capabilities() |
| reasons = [] |
| ok, why = self.available() |
| if not ok: |
| reasons.append(f"engine unavailable: {why}") |
| if observable not in cap.observables: |
| reasons.append(f"observable '{observable}' not provided") |
| if cap.supported_elements is not None: |
| bad = set(ms.symbols) - cap.supported_elements |
| if bad: |
| reasons.append(f"unsupported elements {sorted(bad)}") |
| if ms.n_atoms > cap.max_atoms: |
| reasons.append(f"{ms.n_atoms} atoms exceeds max {cap.max_atoms}") |
| if ms.boundary_profile not in cap.boundary_conditions: |
| reasons.append(f"boundary profile '{ms.boundary_profile}' unsupported") |
| return (not reasons), reasons |
|
|
| def estimate_cost(self, ms: MatterStructure, observable: str, params: dict | None = None) -> dict[str, float]: |
| return {"cpu_s": 0.01 * ms.n_atoms, "ram_mb": 50.0, "model": "default linear"} |
|
|
| @abstractmethod |
| def run(self, ms: MatterStructure, observable: str, params: dict, ctx: RunContext) -> SolverOutput: ... |
|
|
| def version(self) -> str: |
| return self.capabilities().engine_version |
|
|
