ANDES: fix run_eigenvalue_analysis + add PSS/E raw+dyr dynamic-model loading

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/)

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

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** |

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

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()