diff --git a/ANDES/README.md b/ANDES/README.md index c39c736..63f67e8 100644 --- a/ANDES/README.md +++ b/ANDES/README.md @@ -35,15 +35,39 @@ Configure in your MCP client (e.g., Cursor, Claude Desktop): ## Available Tools -- **run_power_flow(file_path: str)**: Run power flow analysis on a power system case file. +- **run_power_flow(file_path: str, dyr_path: Optional[str] = None)**: Run power flow analysis on a power system case file. + - `dyr_path`: optional path to a PSS/E `.dyr` dynamic-model file (generators, exciters, governors) to attach to a PSS/E `.raw` case via ANDES's `addfile` loading. Without it, `run_time_domain_simulation`/`run_eigenvalue_analysis` have no real dynamics to work with beyond static topology. + - Adds two fields to the returned `power_flow` dict: + - `dynamic_models_loaded`: `True` when `dyr_path` was supplied, else `False`. + - `n_dynamic_generators`: count of dynamic generator models attached (`ss.groups["SynGen"].n`), `0` when no `.dyr` was loaded. + - Purely additive: omitting `dyr_path` behaves exactly as before. - **run_time_domain_simulation(step_size: float = 0.01, t_end: float = 10.0)**: Run time domain simulation on the currently loaded power system. -- **run_eigenvalue_analysis(file_path: str)**: Run eigenvalue analysis on a power system case. +- **run_eigenvalue_analysis(file_path: str)**: Run eigenvalue (small-signal) analysis on a power system case. Reloads the case fresh from `file_path` and returns: + - `n_modes`: number of eigenvalues/modes. + - `modes`: a list, one entry per mode, each with: + - `eigenvalue`: `[real, imag]` parts of the eigenvalue. + - `frequency_hz`: oscillation frequency in Hz (`0.0` for non-oscillatory/real modes). + - `damping_ratio_pct`: damping ratio as a percentage. + - `is_oscillatory`: whether the mode has a non-zero imaginary part. + + The list is sorted **least-damped (most concerning) first**. + - `participation_factors`: the raw participation-factor matrix from ANDES. + - `state_names`: state variable labels, one per mode/row. + - `success`: whether ANDES's `EIG.run()` reported success. - **get_system_info()**: Get information about the currently loaded power system. +- **load_network_from_json(network_json: str, out_path: str)**: Stage a powerio JSON transport string as a MATPOWER file for ANDES (pass `out_path` to `run_power_flow`). +- **load_network_from_any(file_path: str, out_path: str, source_format: Optional[str] = None)**: Stage any powerio-readable case (MATPOWER `.m`, PSS/E `.raw` v33, PowerWorld `.aux`, PowerModels JSON, egret JSON) as a MATPOWER file for ANDES. + +## License note + +[ANDES](https://github.com/curent/andes) is GPL-3.0; it is installed only as an optional pip extra (`andes = ["andes"]` in `pyproject.toml`), never vendored into this MIT-licensed repo. The raw+dyr example/test case (`ieee14.raw`/`ieee14.dyr`) is likewise referenced at call time via `andes.get_case(...)` from the installed `andes` package's own bundled `andes/cases/` directory -- never vendored into this repo either. ## Prompt Example Could you run power flow on the Kundur case at `yourpath\PowerMCP\ANDES\kundur_full.json` using ANDES and summarize the results? Then call `get_system_info` to show the system details. +Or, with a PSS/E raw+dyr case: run power flow on `ieee14.raw` with `dyr_path` set to `ieee14.dyr` (e.g. via `andes.get_case("ieee14/ieee14.raw")` / `andes.get_case("ieee14/ieee14.dyr")` for ANDES's own bundled example), confirm `dynamic_models_loaded` and `n_dynamic_generators` in the result, then run a time-domain simulation against the loaded dynamics. + ## Resources - [ANDES Documentation](https://andes.readthedocs.io/) diff --git a/ANDES/andes_mcp.py b/ANDES/andes_mcp.py index c85a146..70c4dc7 100644 --- a/ANDES/andes_mcp.py +++ b/ANDES/andes_mcp.py @@ -5,6 +5,7 @@ import sys import shutil import json +import numpy as np from pathlib import Path from contextlib import redirect_stdout, redirect_stderr from mcp.server.fastmcp import FastMCP @@ -59,12 +60,17 @@ def _ensure_file_logging(): system_state: Dict[str, Any] = {} @mcp.tool() -def run_power_flow(file_path: str) -> Dict[str, Any]: +def run_power_flow(file_path: str, dyr_path: Optional[str] = None) -> Dict[str, Any]: """Run power flow analysis on a power system case - + Args: file_path: Path to the case file - + dyr_path: Optional path to a PSS/E .dyr dynamic-model file (generators, + exciters, governors) to attach to a PSS/E .raw case. When given, + it is loaded alongside file_path via ANDES's addfile mechanism, + enabling run_time_domain_simulation/run_eigenvalue_analysis to + operate on real dynamics instead of static topology only. + Returns: Dict containing power flow results and output information """ @@ -78,36 +84,62 @@ def run_power_flow(file_path: str) -> Dict[str, Any]: "message": f"Input file not found: {abs_file_path}" } + # Resolve and validate the optional .dyr file before any run-dir/chdir + # work happens, same pattern as the main input file check above. + abs_dyr_path = None + if dyr_path is not None: + abs_dyr_path = os.path.abspath(dyr_path) + if not os.path.exists(abs_dyr_path): + return { + "status": "error", + "message": f"Dynamic model file not found: {abs_dyr_path}" + } + # Create a unique directory for this run run_dir = os.path.join(_andes_runs_dir(), f"pf_{Path(abs_file_path).stem}") os.makedirs(run_dir, exist_ok=True) - + # Copy input file to run directory input_file = os.path.join(run_dir, os.path.basename(abs_file_path)) shutil.copy2(abs_file_path, input_file) - + + # Copy the .dyr file into run_dir alongside the main input, and use + # the copied path as addfile -- keeps everything this run touched + # under output_dir. + dyr_file = None + if abs_dyr_path is not None: + dyr_file = os.path.join(run_dir, os.path.basename(abs_dyr_path)) + shutil.copy2(abs_dyr_path, dyr_file) + # Save current directory and change to run directory original_dir = os.getcwd() os.chdir(run_dir) - + try: # Capture stdout/stderr f_out = io.StringIO() f_err = io.StringIO() - + with redirect_stdout(f_out), redirect_stderr(f_err): - # Run power flow with minimal output - ss = andes.run(input_file, no_output=True, verbose=50) - + # Run power flow with minimal output. addfile is only passed + # when a .dyr was supplied, so the no-dyr call path is + # byte-identical to before. + run_kwargs = {"no_output": True, "verbose": 50} + if dyr_file is not None: + run_kwargs["addfile"] = dyr_file + ss = andes.run(input_file, **run_kwargs) + # Store system state for other tools system_state['current_system'] = ss - + # Extract key power flow results pflow_results = { "converged": ss.PFlow.converged, "iterations": ss.PFlow.niter if hasattr(ss.PFlow, 'niter') else 0, "max_mis": float(ss.PFlow.mis[-1]) if hasattr(ss.PFlow, 'mis') and len(ss.PFlow.mis) > 0 else 0.0, - "time": float(ss.PFlow.t) if hasattr(ss.PFlow, 't') else 0.0 + "time": float(ss.PFlow.t) if hasattr(ss.PFlow, 't') else 0.0, + "dynamic_models_loaded": dyr_path is not None, + "n_dynamic_generators": int(getattr(ss.groups.get("SynGen"), "n", 0)) if hasattr(ss, "groups") else 0, } # Get list of output files @@ -255,15 +287,34 @@ def run_eigenvalue_analysis(file_path: str) -> Dict[str, Any]: # Run eigenvalue analysis success = ss.EIG.run() - - # Extract eigenvalue results + + # Extract eigenvalue results. ss.EIG.mu holds the eigenvalues + # (complex array); frequency and damping ratio are derived + # from mu using the same formula ANDES's own EIG.post_process() + # uses internally for its text report: + # freq_hz = |Im(mu)| / (2*pi) + # damping_pct = -100 * Re(mu) / |mu| + modes = [] + for mu in ss.EIG.mu: + if mu.imag == 0: + freq_hz, damping_pct = 0.0, 0.0 + else: + freq_hz = abs(mu.imag) / (2 * np.pi) + damping_pct = -100.0 * mu.real / abs(mu) + modes.append({ + "eigenvalue": [float(mu.real), float(mu.imag)], + "frequency_hz": freq_hz, + "damping_ratio_pct": damping_pct, + "is_oscillatory": bool(mu.imag != 0), + }) + modes.sort(key=lambda m: m["damping_ratio_pct"]) # least-damped (most concerning) first + eig_results = { - "n_eigenvalues": len(ss.EIG.mu) if hasattr(ss.EIG, 'mu') else 0, - "eigenvalues": ss.EIG.mu.tolist() if hasattr(ss.EIG, 'mu') else [], - "eigenvectors": ss.EIG.vectors.tolist() if hasattr(ss.EIG, 'vectors') else [], - "participation_factors": ss.EIG.pfactors.tolist() if hasattr(ss.EIG, 'pfactors') else [], - "state_variables": ss.EIG.state_desc if hasattr(ss.EIG, 'state_desc') else [], - "success": success + "n_modes": len(modes), + "modes": modes, + "participation_factors": ss.EIG.pfactors.tolist() if getattr(ss.EIG, "pfactors", None) is not None else [], + "state_names": list(ss.EIG.x_name) if getattr(ss.EIG, "x_name", None) is not None else [], + "success": success, } # Get list of output files diff --git a/powermcp/README.md b/powermcp/README.md index 291ebe7..08ea375 100644 --- a/powermcp/README.md +++ b/powermcp/README.md @@ -36,7 +36,7 @@ pip install "powermcp[all]" # everything | Extra | Tool(s) | Notes | |---|---|---| | *(none / core)* | pandapower, PyPSA | always installed | -| `andes` | ANDES | | +| `andes` | ANDES | GPL-3.0; installed only as an optional pip extra, never vendored | | `egret` | Egret | + needs an external solver (ipopt/Gurobi) | | `opendss` | OpenDSS | | | `surge` | surge | **Python 3.12–3.14 only** | diff --git a/tests/conftest.py b/tests/conftest.py index 5c8c481..b188920 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,6 +2,8 @@ from __future__ import annotations +import sys + import pytest @@ -14,3 +16,22 @@ def isolated_config(tmp_path, monkeypatch): if var.startswith("POWERMCP_") and var != "POWERMCP_HOME": monkeypatch.delenv(var, raising=False) return tmp_path + + +@pytest.fixture() +def andes_mcp(): + """Import andes_mcp from the registry-resolved server dir, skipping if + andes is not installed. + + Shared by test_powerio_server.py (the powerio/pandapower bridge tools) + and test_andes_server.py (the ANDES engine tools themselves). + """ + pytest.importorskip("andes") + from powermcp.registry import TOOLS + + andes_dir = str(TOOLS["andes"].resolve_server_dir()) + if andes_dir not in sys.path: + sys.path.insert(0, andes_dir) + import andes_mcp as _andes_mcp + + return _andes_mcp diff --git a/tests/test_andes_server.py b/tests/test_andes_server.py new file mode 100644 index 0000000..3ca839c --- /dev/null +++ b/tests/test_andes_server.py @@ -0,0 +1,172 @@ +"""Tests for the ANDES MCP server's simulation tools (power flow, time-domain, +and eigenvalue/small-signal analysis). + +andes is an optional pip extra (`pip install "powermcp[andes]"`), not a core +dependency, so every test in this module goes through the `andes_mcp` fixture +in conftest.py, which calls `pytest.importorskip("andes")`. In this repo's +base CI job (.github/workflows/test.yml installs no extras), these tests +therefore show as SKIPPED, not run -- verified locally by installing andes +into a venv and running this file directly (`pip install andes && pytest +tests/test_andes_server.py -v`), matching the existing ANDES-bridge tests' +posture in test_powerio_server.py. + +ANDES/kundur_full.json is the repo's only ANDES fixture: a Kundur two-area +four-machine system with its own embedded dynamic models (GENROU, exciters, +governors), so it exercises power flow, time-domain simulation, and +eigenvalue analysis all in one case with no fixture authoring needed here. + +The raw+dyr dynamic-model-loading tests further down this file use ANDES's +own bundled ieee14 PSS/E case (raw + dyr), resolved via andes.get_case(...) +at test-call time -- see the comment above _ieee14_raw_dyr_paths() for why. +""" + +from __future__ import annotations + +from pathlib import Path + +KUNDUR_FULL = Path(__file__).resolve().parents[1] / "ANDES" / "kundur_full.json" + + +def test_run_power_flow_converges(andes_mcp): + r = andes_mcp.run_power_flow(str(KUNDUR_FULL)) + assert r["status"] == "success", r + assert r["power_flow"]["converged"] is True + + +def test_get_system_info_reports_plausible_counts(andes_mcp): + # get_system_info reads from the module-level system_state, which the + # power flow run above populates. + r = andes_mcp.run_power_flow(str(KUNDUR_FULL)) + assert r["status"] == "success", r + + info = andes_mcp.get_system_info() + assert info["status"] == "success", info + assert info["num_buses"] > 0 + assert info["num_generators"] > 0 + + +def test_run_time_domain_simulation_completes(andes_mcp): + pf = andes_mcp.run_power_flow(str(KUNDUR_FULL)) + assert pf["status"] == "success", pf + + r = andes_mcp.run_time_domain_simulation(step_size=0.01, t_end=1.0) + assert r["status"] == "success", r + sim = r["simulation"] + assert sim["success"] is True + assert sim["status"] == "completed" + assert sim["t_end"] == 1.0 + assert sim["step_size"] == 0.01 + # Not asserting on t_array's shape here: this installed ANDES version + # returns ss.dae.t as a 0-d array (final sim time) rather than a full + # per-step series, which is a pre-existing quirk of + # run_time_domain_simulation orthogonal to the eigenvalue-analysis fix + # this PR targets, so it's left alone (out of scope). + assert "t_array" in sim + + +def test_run_eigenvalue_analysis_returns_modes(andes_mcp): + r = andes_mcp.run_eigenvalue_analysis(str(KUNDUR_FULL)) + assert r["status"] == "success", r + analysis = r["analysis"] + assert analysis["success"] is True + assert analysis["n_modes"] > 0 + + modes = analysis["modes"] + assert len(modes) == analysis["n_modes"] + for mode in modes: + assert isinstance(mode["frequency_hz"], (int, float)) + assert isinstance(mode["damping_ratio_pct"], (int, float)) + assert isinstance(mode["is_oscillatory"], bool) + assert len(mode["eigenvalue"]) == 2 + + # Regression check for the original bug: run_eigenvalue_analysis used to + # read ss.EIG.vectors/state_desc, attributes that don't exist on the real + # EIG object, so hasattr() guards silently returned [] for these fields. + # The real attribute is x_name (state labels); it must be non-empty now. + assert len(analysis["state_names"]) > 0 + assert len(analysis["participation_factors"]) > 0 + + # Modes are sorted least-damped (most concerning) first. + damping_values = [m["damping_ratio_pct"] for m in modes] + assert damping_values == sorted(damping_values) + + +def test_run_eigenvalue_analysis_missing_file(andes_mcp, tmp_path): + r = andes_mcp.run_eigenvalue_analysis(str(tmp_path / "nope.json")) + assert r["status"] == "error" + assert "not found" in r["message"].lower() + + +# --------------------------------------------------------------------------- +# PSS/E raw+dyr dynamic-model loading (fungible-farm/PowerMCP#2) +# +# These tests use ANDES's own bundled ieee14 example case (raw + dyr), +# resolved at test-call time via andes.get_case(...), which reads from the +# installed andes package's own andes/cases/ directory (andes declares these +# as package-data in its own pyproject.toml). No .raw/.dyr fixture is +# authored or vendored into this repo: ieee14.raw/.dyr are GPL-3.0 ANDES +# files, and this repo is MIT, so referencing the installed dependency's own +# copy avoids any vendoring question entirely. +# +# andes must not be imported at module scope (it may not be installed), so +# the get_case() calls happen inside a helper invoked from within each test +# body, after the andes_mcp fixture parameter has already run +# pytest.importorskip("andes"). +# --------------------------------------------------------------------------- + + +def _ieee14_raw_dyr_paths(): + import andes + + raw_path = andes.get_case("ieee14/ieee14.raw") + dyr_path = andes.get_case("ieee14/ieee14.dyr") + return raw_path, dyr_path + + +def test_run_power_flow_without_dyr_has_no_dynamic_models(andes_mcp): + raw_path, _ = _ieee14_raw_dyr_paths() + + r = andes_mcp.run_power_flow(raw_path) + assert r["status"] == "success", r + pf = r["power_flow"] + assert pf["converged"] is True + assert pf["dynamic_models_loaded"] is False + assert pf["n_dynamic_generators"] == 0 + + +def test_run_power_flow_with_dyr_attaches_dynamic_models(andes_mcp): + raw_path, dyr_path = _ieee14_raw_dyr_paths() + + r = andes_mcp.run_power_flow(raw_path, dyr_path=dyr_path) + assert r["status"] == "success", r + pf = r["power_flow"] + assert pf["converged"] is True + assert pf["dynamic_models_loaded"] is True + # Not asserting an exact count (e.g. == 5): stay robust to upstream ANDES + # case-file changes across versions -- only assert that dynamics + # actually attached. + assert pf["n_dynamic_generators"] > 0 + + +def test_run_power_flow_with_dyr_enables_time_domain_simulation(andes_mcp): + # The actual motivating scenario from the issue: without a .dyr there are + # no real dynamics to simulate. + raw_path, dyr_path = _ieee14_raw_dyr_paths() + + pf = andes_mcp.run_power_flow(raw_path, dyr_path=dyr_path) + assert pf["status"] == "success", pf + assert pf["power_flow"]["n_dynamic_generators"] > 0 + + r = andes_mcp.run_time_domain_simulation(step_size=0.01, t_end=1.0) + assert r["status"] == "success", r + sim = r["simulation"] + assert sim["success"] is True + assert sim["status"] == "completed" + + +def test_run_power_flow_missing_dyr_file(andes_mcp, tmp_path): + raw_path, _ = _ieee14_raw_dyr_paths() + + r = andes_mcp.run_power_flow(raw_path, dyr_path=str(tmp_path / "nope.dyr")) + assert r["status"] == "error" + assert "not found" in r["message"].lower() diff --git a/tests/test_powerio_server.py b/tests/test_powerio_server.py index 81c99cd..88f536d 100644 --- a/tests/test_powerio_server.py +++ b/tests/test_powerio_server.py @@ -467,21 +467,13 @@ def test_wrong_schema_json_maps_cleanly(): # --------------------------------------------------------------------------- # ANDES bridge tests +# +# The andes_mcp fixture (shared with test_andes_server.py) lives in +# conftest.py; it skips via pytest.importorskip("andes") when andes isn't +# installed. # --------------------------------------------------------------------------- -def _load_andes_mcp(): - """Import andes_mcp from the registry-resolved server dir, skipping if - andes or powerio are not installed.""" - pytest.importorskip("andes") - andes_dir = str(TOOLS["andes"].resolve_server_dir()) - if andes_dir not in sys.path: - sys.path.insert(0, andes_dir) - import andes_mcp # noqa: E402 - return andes_mcp - - -def test_andes_load_network_from_any(tmp_path): - andes_mcp = _load_andes_mcp() +def test_andes_load_network_from_any(tmp_path, andes_mcp): out = tmp_path / "case9.m" r = andes_mcp.load_network_from_any(str(CASE9), str(out)) assert r["status"] == "success", r @@ -490,8 +482,7 @@ def test_andes_load_network_from_any(tmp_path): assert r["info"]["buses"] == 9 -def test_andes_load_network_from_json(tmp_path): - andes_mcp = _load_andes_mcp() +def test_andes_load_network_from_json(tmp_path, andes_mcp): transport = powerio_mcp.parse(path=str(CASE9))["json"] out = tmp_path / "case9_from_json.m" r = andes_mcp.load_network_from_json(transport, str(out)) @@ -500,8 +491,7 @@ def test_andes_load_network_from_json(tmp_path): assert r["info"]["buses"] == 9 -def test_andes_load_missing_file(tmp_path): - andes_mcp = _load_andes_mcp() +def test_andes_load_missing_file(tmp_path, andes_mcp): r = andes_mcp.load_network_from_any("/nope/missing.m", str(tmp_path / "x.m")) assert r["status"] == "error" assert "not found" in r["message"].lower()