diff --git a/OpenDSS/opendss_tools/configuration.py b/OpenDSS/opendss_tools/configuration.py index 48a9a00..cd724b6 100644 --- a/OpenDSS/opendss_tools/configuration.py +++ b/OpenDSS/opendss_tools/configuration.py @@ -18,23 +18,22 @@ def compile_opendss_file(dss_file: str, force_recompile: bool = False) -> Dict[str, Any]: - """Compile a master DSS file (ClearAll + Compile). + """Compile a master DSS file and verify the resulting circuit is usable. - If the same resolved path is already loaded, skips compile unless ``force_recompile`` is True - (e.g. user wants to recompile the model to clean changes). A different ``dss_file`` loads a new model and always compiles. - After ``clear_all_opendss_memory``, the next call compiles as usual. - - Returns dss_file, circuit_readiness, circuit_loaded, and whether the compile was skipped. + OpenDSS may report a failed Redirect/Compile in diagnostics yet leave a + partial circuit. A successful return from the wrapper therefore must also + pass readiness and model-integrity checks before this session marks the + circuit as loaded. """ try: dss_file = checked_path(dss_file, purpose="dss_file") - # A DSS model may Redirect/Compile sibling files. Preflight the whole - # project tree when containment is enabled; preserve the historical - # single-file behavior when the operator has not configured roots. + # A DSS model may Redirect/Compile sibling files. Preflight the whole + # project tree when containment is enabled. if allowed_roots(): checked_read_tree(str(Path(dss_file).parent), purpose="DSS input tree") except PathNotAllowed as exc: return _err(str(exc)) + resolved = str(Path(dss_file).resolve()) if ( state.circuit_loaded @@ -42,21 +41,34 @@ def compile_opendss_file(dss_file: str, force_recompile: bool = False) -> Dict[s and resolved == state.last_compiled_dss_file and not force_recompile ): - readiness = dss_tools.configuration.circuit_readiness() - return _ok( - { - "dss_file": dss_file, - "resolved_dss_file": resolved, - "skipped": True, - "already_compiled": True, - "circuit_readiness": readiness, - "circuit_loaded": True, - "solution_available": state.solution_available, - } - ) + try: + readiness = dss_tools.configuration.circuit_readiness() + validation_error = _validate_compiled_circuit(readiness) + if validation_error: + _reset_compile_state() + return _err(validation_error) + return _ok( + { + "dss_file": dss_file, + "resolved_dss_file": resolved, + "skipped": True, + "already_compiled": True, + "circuit_readiness": readiness, + "circuit_loaded": True, + "solution_available": state.solution_available, + } + ) + except Exception as exc: + _reset_compile_state() + return _err(f"Unable to verify the previously compiled circuit: {exc}") + try: dss_tools.configuration.compile_dss(dss_file) readiness = dss_tools.configuration.circuit_readiness() + validation_error = _validate_compiled_circuit(readiness) + if validation_error: + _reset_compile_state() + return _err(validation_error) state.circuit_loaded = True state.solution_available = False state.last_compiled_dss_file = resolved @@ -70,25 +82,67 @@ def compile_opendss_file(dss_file: str, force_recompile: bool = False) -> Dict[s "solution_available": False, } ) - except Exception as e: - state.circuit_loaded = False - state.solution_available = False - state.last_compiled_dss_file = None - return _err(str(e)) + except Exception as exc: + _reset_compile_state() + return _err(str(exc)) + + +def _validate_compiled_circuit(readiness: Any) -> str | None: + """Return an error when engine readiness or model counts reject the circuit.""" + if isinstance(readiness, dict): + for key in ("success", "ready", "is_ready", "circuit_ready", "converged"): + value = readiness.get(key) + if value is False: + return f"OpenDSS circuit readiness reports {key}=false." + for key in ("error", "errors", "compile_error", "redirect_errors"): + value = readiness.get(key) + if value: + return f"OpenDSS circuit readiness reports {key}: {value}" + + # The 13-bus failure mode leaves a compile command without the expected + # model. Inspect summary records only when this toolkit version exposes a + # bus count, keeping compatibility with versions that return other shapes. + try: + summary = dss_tools.model._summary_model_records + if callable(summary): + summary = summary() + if isinstance(summary, dict): + counts = summary.get("counts", summary) + if isinstance(counts, dict): + buses = counts.get("buses", counts.get("bus_count")) + if isinstance(buses, (int, float)) and buses <= 0: + return "OpenDSS compiled no buses; check Redirect/Compile paths and engine diagnostics." + elif isinstance(counts, (list, tuple)) and counts: + # Some toolkit versions expose summary records as a column + # mapping, others as a sequence of rows with a bus count. + for row in counts: + if isinstance(row, dict): + buses = row.get("buses", row.get("bus_count")) + if isinstance(buses, (int, float)) and buses <= 0: + return "OpenDSS compiled no buses; check Redirect/Compile paths and engine diagnostics." + except (AttributeError, TypeError, ValueError): + # Optional metadata varies by toolkit version; don't reject a circuit + # solely because this extra diagnostic is unavailable. + pass + return None + + +def _reset_compile_state() -> None: + state.circuit_loaded = False + state.solution_available = False + state.last_compiled_dss_file = None def clear_all_opendss_memory() -> Dict[str, Any]: - """Clear OpenDSS engine memory (ClearAll); resets circuit_loaded, solution_available, and last compiled path.""" + """Clear OpenDSS engine memory and reset circuit state.""" try: engine.ensure_engine().text("ClearAll") - state.circuit_loaded = False - state.solution_available = False - state.last_compiled_dss_file = None + _reset_compile_state() return _ok( {"cleared": True, "circuit_loaded": False, "solution_available": False} ) - except Exception as e: - return _err(str(e)) + except Exception as exc: + return _err(str(exc)) def register_configuration_tools(mcp: FastMCP) -> None: diff --git a/tests/test_opendss_compile_validation.py b/tests/test_opendss_compile_validation.py new file mode 100644 index 0000000..aaaaa08 --- /dev/null +++ b/tests/test_opendss_compile_validation.py @@ -0,0 +1,127 @@ +"""Regression tests for OpenDSS compile-result validation.""" + +from __future__ import annotations + +import importlib +import sys +import types +from pathlib import Path + +import pytest + + +@pytest.fixture() +def configuration_module(monkeypatch): + opendss_dir = Path(__file__).resolve().parents[1] / "OpenDSS" + monkeypatch.syspath_prepend(str(opendss_dir)) + + fake_config = types.SimpleNamespace( + compile_dss=lambda _path: None, + circuit_readiness=lambda: {"ready": True}, + ) + fake_model = types.SimpleNamespace( + _summary_model_records={"counts": {"buses": 13}} + ) + fake_tools = types.SimpleNamespace( + configuration=fake_config, + model=fake_model, + update_dss=lambda _dss: None, + ) + monkeypatch.setitem( + sys.modules, "py_dss_toolkit", types.SimpleNamespace(dss_tools=fake_tools) + ) + sandbox = types.ModuleType("powermcp.sandbox") + sandbox.PathNotAllowed = type("PathNotAllowed", (Exception,), {}) + sandbox.allowed_roots = lambda: [] + sandbox.checked_path = lambda value, **_kwargs: str(value) + sandbox.checked_read_tree = lambda *_args, **_kwargs: None + monkeypatch.setitem(sys.modules, "powermcp.sandbox", sandbox) + + for name in ( + "core", "core.state", "core.engine", "opendss_tools", + "opendss_tools.configuration", "utils", "utils.responses", + ): + monkeypatch.delitem(sys.modules, name, raising=False) + + core = types.ModuleType("core") + core.__path__ = [] + monkeypatch.setitem(sys.modules, "core", core) + + state = types.ModuleType("core.state") + state.circuit_loaded = False + state.solution_available = False + state.last_compiled_dss_file = None + monkeypatch.setitem(sys.modules, "core.state", state) + core.state = state + + engine = types.ModuleType("core.engine") + engine.ensure_engine = lambda: types.SimpleNamespace(text=lambda _cmd: None) + monkeypatch.setitem(sys.modules, "core.engine", engine) + core.engine = engine + + utils = types.ModuleType("utils") + utils.__path__ = [] + monkeypatch.setitem(sys.modules, "utils", utils) + responses = types.ModuleType("utils.responses") + responses._err = lambda message: {"success": False, "error": message} + responses._ok = lambda payload=None: { + "success": True, + **({"payload": payload} if payload is not None else {}), + } + monkeypatch.setitem(sys.modules, "utils.responses", responses) + + module = importlib.import_module("opendss_tools.configuration") + module.dss_tools = fake_tools + module.state = state + return module, fake_tools, state + + +def test_compile_rejects_explicit_unready_state(configuration_module, tmp_path): + module, tools, state = configuration_module + tools.configuration.circuit_readiness = lambda: {"ready": False} + + result = module.compile_opendss_file(str(tmp_path / "master.dss")) + + assert result["success"] is False + assert "ready=false" in result["error"] + assert state.circuit_loaded is False + assert state.last_compiled_dss_file is None + + +def test_compile_rejects_empty_model_when_summary_exposes_bus_count(configuration_module, tmp_path): + module, tools, state = configuration_module + tools.model._summary_model_records = {"counts": {"buses": 0}} + + result = module.compile_opendss_file(str(tmp_path / "master.dss")) + + assert result["success"] is False + assert "compiled no buses" in result["error"] + assert state.circuit_loaded is False + assert state.solution_available is False + + +def test_compile_accepts_nonempty_model(configuration_module, tmp_path): + module, tools, state = configuration_module + tools.configuration.circuit_readiness = lambda: {"ready": True} + tools.model._summary_model_records = {"counts": {"buses": 13}} + + result = module.compile_opendss_file(str(tmp_path / "master.dss")) + + assert result["success"] is True + assert result["payload"]["circuit_loaded"] is True + assert state.circuit_loaded is True + assert state.last_compiled_dss_file == str((tmp_path / "master.dss").resolve()) + + +def test_rechecking_cached_compile_invalidates_stale_state(configuration_module, tmp_path): + module, tools, state = configuration_module + path = tmp_path / "master.dss" + state.circuit_loaded = True + state.last_compiled_dss_file = str(path.resolve()) + tools.model._summary_model_records = {"counts": {"buses": 0}} + + result = module.compile_opendss_file(str(path)) + + assert result["success"] is False + assert state.circuit_loaded is False + assert state.last_compiled_dss_file is None