Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 18 additions & 3 deletions Egret/egret_mcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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(
Expand Down
5 changes: 3 additions & 2 deletions pandapower/panda_mcp.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from typing import Dict, List, Optional, Tuple, Any, Union
import copy
import sys
from pathlib import Path

Expand Down Expand Up @@ -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
Expand All @@ -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:
Expand Down
69 changes: 69 additions & 0 deletions tests/test_egret_unrated_branches.py
Original file line number Diff line number Diff line change
@@ -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
41 changes: 41 additions & 0 deletions tests/test_pandapower_contingency.py
Original file line number Diff line number Diff line change
@@ -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()
Loading