diff --git a/Egret/egret_mcp.py b/Egret/egret_mcp.py index 210fb56d..8b0e2f57 100644 --- a/Egret/egret_mcp.py +++ b/Egret/egret_mcp.py @@ -36,6 +36,21 @@ # Create an MCP server mcp = FastMCP("Egret Power System Analysis Server") +# MATPOWER marks an unlimited branch with rateA/B/C = 0. Egret's own parser +# stores that as None, which its models read as "no thermal limit", but they +# index these keys directly, so a converter that leaves them out (powerio 0.11.5 +# does) fails with KeyError: 'rating_long_term'. +_BRANCH_RATING_KEYS = ("rating_long_term", "rating_short_term", "rating_emergency") + + +def _read_model_data(case_file: str) -> ModelData: + """Read a case and give unrated branches Egret's explicit None ratings.""" + md = ModelData.read(case_file) + for branch in md.data.get("elements", {}).get("branch", {}).values(): + for key in _BRANCH_RATING_KEYS: + branch.setdefault(key, None) + return md + @mcp.tool() def solve_unit_commitment_problem( case_file: str, @@ -65,7 +80,7 @@ def solve_unit_commitment_problem( with redirect_stdout(f_out), redirect_stderr(f_err): # Load the case file - md = ModelData.read(case_file) + md = _read_model_data(case_file) # Solve the unit commitment problem with solver_tee=False to silence solver output md_sol = solve_unit_commitment( @@ -121,7 +136,7 @@ def solve_ac_opf( with redirect_stdout(f_out), redirect_stderr(f_err): # Load the case file - md = ModelData.read(case_file) + md = _read_model_data(case_file) # Solve AC OPF with solver_tee=False to silence solver output md_sol, results = solve_acopf( @@ -178,7 +193,7 @@ def solve_dc_opf( with redirect_stdout(f_out), redirect_stderr(f_err): # Load the case file - md = ModelData.read(case_file) + md = _read_model_data(case_file) # Solve DC OPF with solver_tee=False to silence solver output md_sol, results = solve_dcopf( diff --git a/pandapower/panda_mcp.py b/pandapower/panda_mcp.py index f75f243c..ea2fc548 100644 --- a/pandapower/panda_mcp.py +++ b/pandapower/panda_mcp.py @@ -1,4 +1,5 @@ from typing import Dict, List, Optional, Tuple, Any, Union +import copy import sys from pathlib import Path @@ -265,7 +266,7 @@ def run_contingency_analysis(contingency_type: str = "N-1", net = _get_network() # Store original state - orig_net = net.deepcopy() + orig_net = copy.deepcopy(net) results = [] # Define elements to analyze @@ -276,7 +277,7 @@ def run_contingency_analysis(contingency_type: str = "N-1", for element_type in elements: for idx in net[element_type].index: # Create contingency by taking element out of service - contingency_net = orig_net.deepcopy() + contingency_net = copy.deepcopy(orig_net) contingency_net[element_type].at[idx, 'in_service'] = False try: diff --git a/tests/test_egret_unrated_branches.py b/tests/test_egret_unrated_branches.py new file mode 100644 index 00000000..69440240 --- /dev/null +++ b/tests/test_egret_unrated_branches.py @@ -0,0 +1,69 @@ +"""Egret OPF must solve cases whose branches have no thermal rating (issue #128). + +MATPOWER writes rateA = 0 for an unlimited branch. powerio leaves the Egret +rating keys out for those branches, and Egret's OPF models index them directly. +""" + +import json +from pathlib import Path +import sys + +import pytest + +_CASE14 = Path(__file__).resolve().parents[1] / "Egret" / "pglib_opf_case14_ieee.m" + + +@pytest.fixture() +def egret_mcp(tmp_path, monkeypatch): + pytest.importorskip("egret") + monkeypatch.setenv("POWERMCP_HOME", str(tmp_path / "powermcp-home")) + from powermcp.registry import TOOLS + + egret_dir = str(TOOLS["egret"].resolve_server_dir()) + if egret_dir not in sys.path: + sys.path.insert(0, egret_dir) + import egret_mcp as _egret_mcp + + return _egret_mcp + + +def _unrate_every_other_branch(target: Path) -> int: + """Copy IEEE 14-bus with rateA/B/C = 0 on alternate branches; return how many.""" + head, rest = _CASE14.read_text().split("mpc.branch = [", 1) + body, tail = rest.split("];", 1) + rows = [row.split() for row in body.strip().splitlines()] + for cols in rows[::2]: + cols[5:8] = ["0", "0", "0"] + table = "\n".join("\t" + "\t".join(cols) for cols in rows) + target.write_text(f"{head}mpc.branch = [\n{table}\n];{tail}") + return len(rows[::2]) + + +def test_dc_opf_solves_case_with_unrated_branches(egret_mcp, tmp_path): + pytest.importorskip("highspy") + case = tmp_path / "case14_unrated.m" + unrated = _unrate_every_other_branch(case) + + loaded = egret_mcp.load_model_from_any(str(case)) + assert loaded["status"] == "success", loaded + + solved = egret_mcp.solve_dc_opf(loaded["case_file"], solver="highs") + + assert solved["status"] == "success", solved + branches = solved["solution"]["elements"]["branch"] + assert sum(1 for b in branches.values() if b["rating_long_term"] is None) == unrated + + +def test_missing_rating_keys_read_as_unlimited(egret_mcp, tmp_path): + case = tmp_path / "case.json" + case.write_text(json.dumps({ + "elements": {"branch": {"1": {"from_bus": "1", "to_bus": "2"}, + "2": {"from_bus": "2", "to_bus": "3", "rating_long_term": 100.0}}}, + "system": {"baseMVA": 100.0}, + })) + + branches = egret_mcp._read_model_data(str(case)).data["elements"]["branch"] + + assert branches["1"]["rating_long_term"] is None + assert branches["1"]["rating_emergency"] is None + assert branches["2"]["rating_long_term"] == 100.0 diff --git a/tests/test_pandapower_contingency.py b/tests/test_pandapower_contingency.py new file mode 100644 index 00000000..f34509f7 --- /dev/null +++ b/tests/test_pandapower_contingency.py @@ -0,0 +1,41 @@ +"""run_contingency_analysis must work with current pandapower (issue #127). + +pandapowerNet has no ``deepcopy`` method, so every call failed before the tool +switched to ``copy.deepcopy``. +""" + +import importlib.util +from pathlib import Path +import sys + +import pandapower.networks as pn + +_SERVER_PATH = Path(__file__).resolve().parents[1] / "pandapower" / "panda_mcp.py" + + +def _load_server_like_runner(): + # The runner puts only the server's own directory on sys.path. + server_dir = str(_SERVER_PATH.parent) + sys.path.insert(0, server_dir) + try: + spec = importlib.util.spec_from_file_location("powermcp_pandapower_server_contingency", _SERVER_PATH) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + finally: + sys.path.remove(server_dir) + + +def test_n1_contingency_runs_on_case14_and_leaves_the_network_untouched(monkeypatch): + server = _load_server_like_runner() + net = pn.case14() + monkeypatch.setattr(server, "_current_net", net) + + result = server.run_contingency_analysis() + + assert result["status"] == "success", result + assert len(result["results"]) == len(net.line) + len(net.trafo) + assert any(r["converged"] for r in result["results"]) + assert net.line.in_service.all() and net.trafo.in_service.all()