Class CppCoreBackend

src/materialdft/backends/cpp_core.py:184–1804

class CppCoreBackend:
    name = "cpp_core"

    def __init__(
        self,
        *,
        default_potcar_root: Path | None = None,
        default_potcar_set: str | None = None,
    ):
        self._module: Any | None = None
        self._import_error: Exception | None = None
        self.default_potcar_root = default_potcar_root.resolve() if default_potcar_root is not None else None
        self.default_potcar_set = default_potcar_set
        try:
            import materialdft_cpp_core as cpp_core

            self._module = cpp_core
        except Exception as exc:  # pragma: no cover - import failure path is exercised via runtime backend selection
            self._import_error = exc

    def _ensure_module(self) -> Any:
        if self._module is None:
            msg = "cpp_core backend is unavailable: pybind11 module import failed"
            if self._import_error is not None:
                msg = f"{msg}: {self._import_error}"
            raise BackendExecutionError(msg, "RUNTIME_ERROR", "PREPARE")
        return self._module

    def _input_obj(self, spec: SimulationSpec) -> dict[str, Any]:
        if isinstance(spec.input, dict):
            return spec.input
        return {}

    def _runtime_policy(self, spec: SimulationSpec) -> dict[str, Any]:
        if isinstance(spec.runtime_policy, dict):
            return spec.runtime_policy
        return {}

    def _resources(self, spec: SimulationSpec) -> dict[str, Any]:
        if isinstance(spec.resources, dict):
            return spec.resources
        return {}

    def _dump_cpp_core_payload_enabled(self, spec: SimulationSpec) -> bool:
        runtime_policy = self._runtime_policy(spec)
        if "dump_cpp_core_payload" in runtime_policy:
            return _as_bool(runtime_policy.get("dump_cpp_core_payload"), False)
        return _as_bool(os.environ.get("MaterialDFT_DUMP_CPP_CORE_PAYLOAD"), False)

    def _gate_min_nelm(self, spec: SimulationSpec) -> int | None:
        gate_profile = self._runtime_policy(spec).get("gate_profile")
        if not isinstance(gate_profile, dict):
            return None
        min_nelm = _as_int(gate_profile.get("min_nelm"), 0)
        if min_nelm <= 0:
            return None
        return min_nelm