From ce731c22fe28ed6a8a0889f0fdb1f3f375f9676a Mon Sep 17 00:00:00 2001 From: sbblack-qualus Date: Thu, 15 Oct 2026 21:48:12 -0600 Subject: [PATCH] PSSE auto discovery and STDIO fix --- PSSE/psse_mcp.py | 233 ++++++++++++++++++++++++++++++++++++++++++++--- config.json | 3 +- 2 files changed, 221 insertions(+), 15 deletions(-) diff --git a/PSSE/psse_mcp.py b/PSSE/psse_mcp.py index 608b285c..f4431a1f 100644 --- a/PSSE/psse_mcp.py +++ b/PSSE/psse_mcp.py @@ -1,25 +1,189 @@ import sys import os +import re import json import io +import importlib from pathlib import Path sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) from mcp.server.fastmcp import FastMCP -from typing import Dict, List, Optional, Any +from typing import Dict, List, Optional, Any, Tuple # Initialize MCP server mcp = FastMCP("PSSE 35+ Positive Sequence Load Flow Program") + +def _version_key(version_str: str) -> Tuple[int, ...]: + """Turn a version string like '36.4.0' or '35.4' into a sortable tuple of ints.""" + parts = re.findall(r"\d+", version_str or "") + return tuple(int(p) for p in parts) if parts else (0,) + + +def _is_valid_psspy_dir(psspy_path: str, exe_path: str) -> bool: + """ + Confirm a candidate psspy directory actually contains the psspy module + (as "psspy.pyc" or "psspy.py") rather than being an empty/incomplete + install folder left behind by a partial installation. + """ + if not (psspy_path and exe_path and os.path.isdir(psspy_path) and os.path.isdir(exe_path)): + return False + return any( + os.path.isfile(os.path.join(psspy_path, f"psspy{ext}")) + for ext in (".pyc", ".py", ".pyd") + ) + + +def _psse_candidates_from_registry(allowed_majors, pyver: str): + """ + Enumerate PSSE installations from the Windows registry. + + PSSE's installer writes entries under HKLM\\SOFTWARE\\PTI\\PSSE \\\\Product Paths + with values such as PsseInstallPath, PsseExePath (PSSBIN) and one + PsseLocalPsspyPath value per supported Python version (casing varies, + e.g. "Psselocalpsspy313Path" vs "PsseLocalPsspy314Path"), so lookups are + done case-insensitively. + """ + candidates = [] + try: + import winreg + except ImportError: + return candidates # not running on Windows + + psspy_value_name = f"psselocalpsspy{pyver}path" + + for major in allowed_majors: + root_key = rf"SOFTWARE\PTI\PSSE {major}" + try: + pkey = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, root_key) + except OSError: + continue + + with pkey: + i = 0 + while True: + try: + build_name = winreg.EnumKey(pkey, i) + except OSError: + break + i += 1 + + try: + bkey = winreg.OpenKey(pkey, rf"{build_name}\Product Paths") + except OSError: + continue + + with bkey: + values = {} + j = 0 + while True: + try: + name, data, _ = winreg.EnumValue(bkey, j) + except OSError: + break + values[name.lower()] = data + j += 1 + + psspy_path = values.get(psspy_value_name) + exe_path = values.get("psseexepath") + version_str = values.get("version") or f"{major}.{build_name}" + if _is_valid_psspy_dir(psspy_path, exe_path): + candidates.append((_version_key(version_str), major, version_str, psspy_path, exe_path)) + + return candidates + + +def _psse_candidates_from_filesystem(allowed_majors, pyver: str): + """Fallback discovery by scanning default PTI install directories.""" + candidates = [] + for major in allowed_majors: + for root in (r"C:\Program Files\PTI", r"C:\Program Files (x86)\PTI"): + psse_dir = os.path.join(root, f"PSSE{major}") + if not os.path.isdir(psse_dir): + continue + for entry in os.listdir(psse_dir): + version_dir = os.path.join(psse_dir, entry) + if not os.path.isdir(version_dir): + continue + psspy_path = os.path.join(version_dir, f"PSSPY{pyver}") + exe_path = os.path.join(version_dir, "PSSBIN") + if _is_valid_psspy_dir(psspy_path, exe_path): + candidates.append((_version_key(entry), major, entry, psspy_path, exe_path)) + return candidates + + +def _discover_psse_paths() -> Tuple[int, str, str, str]: + """ + Locate an installed PSSE version (35 or 36) whose bundled psspy module + matches the Python interpreter currently running this MCP server. + + Resolution order: + 1. Manual override via PSSE_PSSPY_PATH / PSSE_EXE_PATH env vars (with + optional PSSE_VERSION to pick which "psseXX" module to import). + 2. Windows registry (HKLM\\SOFTWARE\\PTI\\PSSE 35|36), populated by the + PSSE installer with exact paths per supported Python version. + 3. Filesystem scan of the default "C:\\Program Files\\PTI\\PSSE35|36" + install locations, merged in to cover installs the registry doesn't + (yet) know about. + + Among all matches, the highest PSSE version is selected. Set PSSE_VERSION + to "35" or "36" to restrict discovery to a single major version. + + Returns (major_version, version_str, psspy_path, exe_path). + """ + pyver = f"{sys.version_info.major}{sys.version_info.minor}" # e.g. "311" + + forced_major = os.environ.get("PSSE_VERSION") + allowed_majors = [int(forced_major)] if forced_major else [36, 35] + + # 1. Manual override ----------------------------------------------------- + override_psspy = os.environ.get("PSSE_PSSPY_PATH") + override_exe = os.environ.get("PSSE_EXE_PATH") + if override_psspy and override_exe: + major = int(forced_major) if forced_major else 36 + return major, "manual-override", override_psspy, override_exe + + # 2. Windows registry, merged with 3. filesystem scan --------------------- + # The registry is queried first since it's authoritative and fast, but it + # can be stale/incomplete (e.g. missing an entry for a version that was + # installed without updating that key), so filesystem results are merged + # in rather than used only as an all-or-nothing fallback. Candidates are + # de-duplicated by psspy path, keeping the version with the most info. + candidates_by_path = {} + for candidate in _psse_candidates_from_registry(allowed_majors, pyver): + candidates_by_path[os.path.normcase(candidate[3])] = candidate + for candidate in _psse_candidates_from_filesystem(allowed_majors, pyver): + candidates_by_path.setdefault(os.path.normcase(candidate[3]), candidate) + candidates = list(candidates_by_path.values()) + + if not candidates: + raise RuntimeError( + f"Could not find a PSSE 35/36 installation with Python {pyver} support " + f"(interpreter: {sys.executable}). Checked the Windows registry " + r"(HKLM\SOFTWARE\PTI\PSSE 35|36) and default install directories " + r"(C:\Program Files\PTI\PSSE35|36). If PSSE is installed elsewhere, " + "set the PSSE_PSSPY_PATH and PSSE_EXE_PATH environment variables " + "to the PSSPY and PSSBIN directories respectively." + ) + + candidates.sort(key=lambda c: c[0], reverse=True) + _, major, version_str, psspy_path, exe_path = candidates[0] + return major, version_str, psspy_path, exe_path + + # Import and initialize PSSE Python library -# You need to change the path to the PSSE Python library and version to the path on your computer. -sys.path.append(r'C:\Program Files\PTI\PSSE36\36.2\PSSPY311') -sys.path.append(r'C:\Program Files\PTI\PSSE36\36.2\PSSBIN') -os.environ['PATH'] = r'C:\Program Files\PTI\PSSE36\36.2\PSSPY311;' + os.environ['PATH'] -os.environ['PATH'] = r'C:\Program Files\PTI\PSSE36\36.2\PSSBIN;' + os.environ['PATH'] -import psse36 +# Path is auto-discovered (registry, then filesystem scan) to find the highest +# installed PSSE version whose psspy module matches this Python interpreter's +# version. See _discover_psse_paths() for details and override env vars. +_psse_major, _psse_version_str, _psse_psspy_path, _psse_exe_path = _discover_psse_paths() +sys.path.append(_psse_psspy_path) +sys.path.append(_psse_exe_path) +os.environ['PATH'] = _psse_psspy_path + os.pathsep + os.environ['PATH'] +os.environ['PATH'] = _psse_exe_path + os.pathsep + os.environ['PATH'] + +importlib.import_module(f"psse{_psse_major}") import psspy psspy.psseinit(50) @@ -64,8 +228,32 @@ def _build_kwargs(spec, provided_args): # --------------------------------------------------------------------------- -# One handler per return_type +# Native stdout protection # --------------------------------------------------------------------------- +# +# Many psspy calls (fdns, nsol, gendsp_2, etc.) write solution/progress text +# to PSSE's "standard destination" (islct=1) as a side effect, which in this +# headless context lands on the process's real stdout -- the same stream +# FastMCP's stdio transport uses for JSON-RPC framing. If that text is not +# intercepted, it corrupts the protocol stream mid-message and permanently +# wedges the server (every subsequent request times out, even ones that +# never touch psspy, since the transport itself is desynced) until it is +# restarted. Wrap every psspy call site in a throwaway stdout capture so +# native output can never reach the real stream. Handlers that intentionally +# want the captured text (_handle_error_only_report/_listing) perform their +# own inner redirect-and-capture, which nests safely inside this one. +import contextlib + + +@contextlib.contextmanager +def _suppress_native_stdout(): + old_stdout = sys.stdout + sys.stdout = io.StringIO() + try: + yield + finally: + sys.stdout = old_stdout + def _handle_error_only(spec, args): """ierr = func(...)""" @@ -368,10 +556,11 @@ def open_case(case: str) -> Dict[str, Any]: Dict with status and case information """ try: - ierr = psspy.case(case) - err, bus_data = psspy.abuscount(flag=2) - err, branch_data = psspy.abrncount(flag=4) - err, gen_data = psspy.amachcount(flag=4) + with _suppress_native_stdout(): + ierr = psspy.case(case) + err, bus_data = psspy.abuscount(flag=2) + err, branch_data = psspy.abrncount(flag=4) + err, gen_data = psspy.amachcount(flag=4) if err == 0: return { @@ -397,7 +586,8 @@ def solve_case() -> Dict[str, Any]: Dict with status and result code """ try: - ierr = psspy.nsol() + with _suppress_native_stdout(): + ierr = psspy.nsol() if ierr == 0: return {'status': 'success', 'ierr': 0} # Load nsol error codes from JSON if available @@ -449,7 +639,22 @@ def run_psspy_command(function_name: str, arguments: Optional[Dict[str, Any]] = return {"status": "error", "message": f"Unknown return_type '{return_type}' for '{function_name}'"} try: - result = handler(spec, arguments) + buf = io.StringIO() + old_stdout = sys.stdout + sys.stdout = buf + try: + result = handler(spec, arguments) + finally: + sys.stdout = old_stdout + # Some psspy functions (e.g. diff, accc_with_trip_2) print their real + # output as a console side effect rather than returning it as a value. + # Surface anything captured that the handler itself didn't already + # return (handlers with return_type error_only_report/_listing put + # it under "report" themselves), so nothing is silently lost -- while + # still keeping it off the real stdout/JSON-RPC stream. + captured = buf.getvalue() + if captured and "report" not in result: + result["_console_output"] = captured # Attach function metadata for AI context result["_function"] = function_name result["_return_type"] = return_type diff --git a/config.json b/config.json index 728cbbeb..edffa413 100644 --- a/config.json +++ b/config.json @@ -52,7 +52,8 @@ "command": "cmd", "args": [ "/c", - "yourpath\\python.exe", + "py", + "-3.11", "yourpath\\PowerMCP\\PSSE\\psse_mcp.py" ] },