diff --git a/.gitignore b/.gitignore index b6148a0..7076dc7 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,150 @@ -.venv/ -**/__pycache__/** \ No newline at end of file +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +pip-wheel-metadata/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +.python-version + +# pipenv +Pipfile.lock + +# PEP 582 +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# IDEs +.vscode/ +.idea/ +*.swp +*.swo +*~ +.cursor/ + +# OS +.DS_Store +Thumbs.db + +# Power system specific files (temporary outputs) +*.log +*.out +*.tmp +*.nc +*.h5 + +# UV package manager +.uv/ +uv.lock + +# MCP/Gradio +flagged/ +*.gradio/ \ No newline at end of file diff --git a/MANIFEST.in b/MANIFEST.in new file mode 100644 index 0000000..ce24bcd --- /dev/null +++ b/MANIFEST.in @@ -0,0 +1,24 @@ +# Include documentation +include README.md +include LICENSE +include config.json + +# Include all data files in subdirectories +recursive-include ANDES *.json *.dyd +recursive-include Egret *.m *.json +recursive-include OpenDSS *.dss *.csv +recursive-include pandapower *.json +recursive-include PowerWorld *.pwb *.pwd +recursive-include PSLF *.sav *.otg *.cntl *.dycr *.dyd +recursive-include PSSE *.dyr *.sav *.con *.mon *.sub +recursive-include PSSE35 *.dyr *.sav *.con *.mon *.sub +recursive-include PyLTSpice *.txt +recursive-include PyPSA *.nc *.txt + +# Include all README files +recursive-include * README.md + +# Exclude test files and cache +recursive-exclude tests * +recursive-exclude * __pycache__ +recursive-exclude * *.py[co] diff --git a/PyPSA/test_case.nc b/PyPSA/test_case.nc deleted file mode 100644 index 06ff00d..0000000 Binary files a/PyPSA/test_case.nc and /dev/null differ diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..23163a7 --- /dev/null +++ b/__init__.py @@ -0,0 +1,85 @@ +""" +PowerMCP - Power System Analysis MCP Server Collection + +An open-source collection of MCP (Model Context Protocol) servers +for power system simulation and analysis software. + +Supported tools: +- pandapower: Power system modeling and analysis +- PyPSA: Power system optimization +- ANDES: Power system dynamics simulation +- OpenDSS: Distribution system analysis +- And more... +""" + +__version__ = "0.1.0" + +# Import availability flags and key functions from submodules +from pandapower_tools import ( + PANDAPOWER_AVAILABLE, + create_empty_network as pandapower_create_empty_network, + create_test_network as pandapower_create_test_network, + load_network as pandapower_load_network, + run_power_flow as pandapower_run_power_flow, + run_dc_power_flow as pandapower_run_dc_power_flow, + get_network_info as pandapower_get_network_info, + add_bus as pandapower_add_bus, + add_line as pandapower_add_line, + add_load as pandapower_add_load, + add_generator as pandapower_add_generator, + add_ext_grid as pandapower_add_ext_grid, + run_contingency_analysis as pandapower_run_contingency_analysis, + get_available_std_types as pandapower_get_available_std_types, +) + +from pypsa_tools import ( + PYPSA_AVAILABLE, + create_network as pypsa_create_network, + get_network_info as pypsa_get_network_info, + add_bus as pypsa_add_bus, + add_generator as pypsa_add_generator, + add_load as pypsa_add_load, + add_line as pypsa_add_line, + run_power_flow as pypsa_run_power_flow, + run_optimal_power_flow as pypsa_run_optimal_power_flow, + load_network as pypsa_load_network, + save_network as pypsa_save_network, +) + +# Convenience imports for submodules +from . import pandapower_tools +from . import pypsa_tools + +__all__ = [ + '__version__', + # Pandapower + 'PANDAPOWER_AVAILABLE', + 'pandapower_create_empty_network', + 'pandapower_create_test_network', + 'pandapower_load_network', + 'pandapower_run_power_flow', + 'pandapower_run_dc_power_flow', + 'pandapower_get_network_info', + 'pandapower_add_bus', + 'pandapower_add_line', + 'pandapower_add_load', + 'pandapower_add_generator', + 'pandapower_add_ext_grid', + 'pandapower_run_contingency_analysis', + 'pandapower_get_available_std_types', + # PyPSA + 'PYPSA_AVAILABLE', + 'pypsa_create_network', + 'pypsa_get_network_info', + 'pypsa_add_bus', + 'pypsa_add_generator', + 'pypsa_add_load', + 'pypsa_add_line', + 'pypsa_run_power_flow', + 'pypsa_run_optimal_power_flow', + 'pypsa_load_network', + 'pypsa_save_network', + # Submodules + 'pandapower_tools', + 'pypsa_tools', +] diff --git a/pandapower/README.md b/pandapower_tools/README.md similarity index 100% rename from pandapower/README.md rename to pandapower_tools/README.md diff --git a/pandapower_tools/__init__.py b/pandapower_tools/__init__.py new file mode 100644 index 0000000..4088fba --- /dev/null +++ b/pandapower_tools/__init__.py @@ -0,0 +1,37 @@ +"""PowerMCP Pandapower Tools - Power system analysis using pandapower.""" + +from .tools import ( + PANDAPOWER_AVAILABLE, + create_empty_network, + create_test_network, + get_available_networks, + load_network, + run_power_flow, + run_dc_power_flow, + get_network_info, + add_bus, + add_line, + add_load, + add_generator, + add_ext_grid, + run_contingency_analysis, + get_available_std_types, +) + +__all__ = [ + 'PANDAPOWER_AVAILABLE', + 'create_empty_network', + 'create_test_network', + 'get_available_networks', + 'load_network', + 'run_power_flow', + 'run_dc_power_flow', + 'get_network_info', + 'add_bus', + 'add_line', + 'add_load', + 'add_generator', + 'add_ext_grid', + 'run_contingency_analysis', + 'get_available_std_types', +] diff --git a/pandapower/panda_mcp.py b/pandapower_tools/panda_mcp.py similarity index 100% rename from pandapower/panda_mcp.py rename to pandapower_tools/panda_mcp.py diff --git a/pandapower/test_case.json b/pandapower_tools/test_case.json similarity index 100% rename from pandapower/test_case.json rename to pandapower_tools/test_case.json diff --git a/pandapower_tools/tools.py b/pandapower_tools/tools.py new file mode 100644 index 0000000..a343055 --- /dev/null +++ b/pandapower_tools/tools.py @@ -0,0 +1,695 @@ +""" +PowerMCP Pandapower Module +Provides power system analysis tools using pandapower. +""" + +from typing import Dict, List, Optional, Any +import json +import inspect +import copy + +try: + import pandapower as pp + import pandapower.networks as pp_networks + PANDAPOWER_AVAILABLE = True +except ImportError: + PANDAPOWER_AVAILABLE = False + pp = None + pp_networks = None + +# Global variable to store the current network +_current_net = None + + +def _get_available_networks() -> Dict[str, Any]: + """Dynamically discover all available network functions in pandapower.networks. + + Returns: + Dict mapping network names to their callable functions + """ + if not PANDAPOWER_AVAILABLE: + return {} + + network_functions = {} + + # Get all members of pp.networks module + for name, obj in inspect.getmembers(pp_networks): + # Skip private/internal items + if name.startswith('_'): + continue + + # Check if it's a callable (function) that could create a network + if callable(obj): + # Try to determine if this function creates a network + # Most network functions either: + # 1. Start with 'case' (IEEE cases) + # 2. Start with 'create_' (CIGRE networks, etc.) + # 3. Are known network names (iceland, GBnetwork, etc.) + # 4. Return a pandapower network + + # Get function signature to check if it can be called with no required args + try: + sig = inspect.signature(obj) + # Check if all parameters have defaults (can be called without args) + can_call_without_args = all( + p.default != inspect.Parameter.empty + for p in sig.parameters.values() + ) + except (ValueError, TypeError): + can_call_without_args = False + + # Include functions that look like network creators + if (name.startswith('case') or + name.startswith('create_') or + name in ['iceland', 'GBnetwork', 'GBreducednetwork', + 'simple_four_bus_system', 'simple_mv_open_ring_net', + 'mv_oberrhein', 'panda_four_load_branch', + 'four_loads_with_branches_out', 'example_simple', + 'example_multivoltage', 'kb_extrem_landnetz_trafo', + 'kb_extrem_landnetz_freileitung', 'kb_extrem_vorstadtnetz_trafo', + 'kb_extrem_vorstadtnetz_kabel'] or + can_call_without_args): + network_functions[name] = obj + + return network_functions + + +# Cache the available networks to avoid repeated inspection +_NETWORK_FUNCTIONS_CACHE = None + + +def _get_network_functions(): + """Get cached network functions or build cache.""" + global _NETWORK_FUNCTIONS_CACHE + if _NETWORK_FUNCTIONS_CACHE is None: + _NETWORK_FUNCTIONS_CACHE = _get_available_networks() + return _NETWORK_FUNCTIONS_CACHE + + +def _get_network(): + """Get the current pandapower network instance.""" + global _current_net + if _current_net is None: + raise RuntimeError("No pandapower network is currently loaded. Please create or load a network first.") + return _current_net + + +def _set_network(net): + """Set the current pandapower network instance.""" + global _current_net + _current_net = copy.deepcopy(net) + +def create_empty_network() -> Dict[str, Any]: + """Create an empty pandapower network. + + Returns: + Dict containing status and network information + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + global _current_net + try: + _current_net = pp.create_empty_network() + return { + "status": "success", + "message": "Empty network created successfully", + "network_info": { + "buses": len(_current_net.bus), + "lines": len(_current_net.line), + "transformers": len(_current_net.trafo), + "generators": len(_current_net.gen), + "loads": len(_current_net.load) + } + } + except Exception as e: + return {"status": "error", "message": f"Failed to create empty network: {str(e)}"} + + +def create_test_network(network_type: str = "case9") -> Dict[str, Any]: + """Create a standard IEEE test network or other built-in pandapower network. + + Args: + network_type: Type of test network. Use get_available_networks() to see all options. + Common examples: case4gs, case5, case6ww, case9, case14, case24_ieee_rts, + case30, case33bw, case39, case57, case89pegase, case118, case145, case300, + case1354pegase, case1888rte, case2848rte, case2869pegase, case3120sp, + case6470rte, case6495rte, case6515rte, case9241pegase, GBnetwork, + GBreducednetwork, iceland, create_cigre_network_hv, create_cigre_network_mv, + create_cigre_network_lv, mv_oberrhein, simple_four_bus_system, + simple_mv_open_ring_net, example_simple, example_multivoltage, and more. + + Returns: + Dict containing status and network information + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + global _current_net + try: + network_functions = _get_network_functions() + + if network_type not in network_functions: + # Try to find a close match (case-insensitive) + lower_type = network_type.lower() + for key in network_functions: + if key.lower() == lower_type: + network_type = key + break + else: + return { + "status": "error", + "message": f"Unknown network type: {network_type}", + "available_types": sorted(list(network_functions.keys())), + "hint": "Use get_available_networks() to see all available network types" + } + + _current_net = network_functions[network_type]() + + return { + "status": "success", + "message": f"Created {network_type} test network", + "network_info": { + "buses": len(_current_net.bus), + "lines": len(_current_net.line), + "transformers": len(_current_net.trafo), + "generators": len(_current_net.gen), + "loads": len(_current_net.load), + "ext_grids": len(_current_net.ext_grid), + "shunts": len(_current_net.shunt) if hasattr(_current_net, 'shunt') else 0 + } + } + except Exception as e: + return {"status": "error", "message": str(e)} + + +def get_available_networks() -> Dict[str, Any]: + """Get a list of all available pandapower test networks. + + Returns: + Dict containing list of available network types with descriptions + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + try: + network_functions = _get_network_functions() + + # Categorize networks + ieee_cases = [] + pegase_cases = [] + rte_cases = [] + cigre_networks = [] + other_networks = [] + + for name in sorted(network_functions.keys()): + if name.startswith('case') and 'pegase' in name.lower(): + pegase_cases.append(name) + elif name.startswith('case') and 'rte' in name.lower(): + rte_cases.append(name) + elif name.startswith('case'): + ieee_cases.append(name) + elif 'cigre' in name.lower(): + cigre_networks.append(name) + else: + other_networks.append(name) + + return { + "status": "success", + "total_networks": len(network_functions), + "categories": { + "ieee_cases": ieee_cases, + "pegase_cases": pegase_cases, + "rte_cases": rte_cases, + "cigre_networks": cigre_networks, + "other_networks": other_networks + }, + "all_networks": sorted(list(network_functions.keys())) + } + except Exception as e: + return {"status": "error", "message": str(e)} + + +def load_network(file_path: str) -> Dict[str, Any]: + """Load a pandapower network from a file. + + Args: + file_path: Path to the network file (.json or .p) + + Returns: + Dict containing status and network information + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + global _current_net + try: + if file_path.endswith('.json'): + _current_net = pp.from_json(file_path) + elif file_path.endswith('.p'): + _current_net = pp.from_pickle(file_path) + else: + return {"status": "error", "message": "Unsupported file format. Use .json or .p files."} + + return { + "status": "success", + "message": f"Network loaded successfully from {file_path}", + "network_info": { + "buses": len(_current_net.bus), + "lines": len(_current_net.line), + "transformers": len(_current_net.trafo), + "generators": len(_current_net.gen), + "loads": len(_current_net.load) + } + } + except FileNotFoundError: + return {"status": "error", "message": f"File not found: {file_path}"} + except Exception as e: + return {"status": "error", "message": f"Failed to load network: {str(e)}"} + + +def run_power_flow(algorithm: str = "nr", calculate_voltage_angles: bool = True, + max_iteration: int = 50, tolerance_mva: float = 1e-8) -> Dict[str, Any]: + """Run AC power flow analysis on the current network. + + Args: + algorithm: Power flow algorithm ('nr', 'bfsw', 'gs', 'fdbx', 'fdxb') + calculate_voltage_angles: Whether to calculate voltage angles + max_iteration: Maximum number of iterations + tolerance_mva: Convergence tolerance in MVA + + Returns: + Dict containing power flow results + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + try: + net = _get_network() + pp.runpp(net, algorithm=algorithm, calculate_voltage_angles=calculate_voltage_angles, + max_iteration=max_iteration, tolerance_mva=tolerance_mva) + + results = { + "status": "success", + "message": "Power flow converged successfully" if net.converged else "Power flow did not converge", + "converged": net.converged, + "bus_results": { + "vm_pu": net.res_bus["vm_pu"].to_dict(), + "va_degree": net.res_bus["va_degree"].to_dict(), + "p_mw": net.res_bus["p_mw"].to_dict(), + "q_mvar": net.res_bus["q_mvar"].to_dict() + }, + "line_results": { + "loading_percent": net.res_line["loading_percent"].to_dict(), + "p_from_mw": net.res_line["p_from_mw"].to_dict(), + "p_to_mw": net.res_line["p_to_mw"].to_dict(), + "pl_mw": net.res_line["pl_mw"].to_dict(), + "ql_mvar": net.res_line["ql_mvar"].to_dict() + }, + "total_losses": { + "p_mw": float(net.res_line["pl_mw"].sum()), + "q_mvar": float(net.res_line["ql_mvar"].sum()) + } + } + + if len(net.trafo) > 0: + results["transformer_results"] = { + "loading_percent": net.res_trafo["loading_percent"].to_dict() + } + + return results + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": f"Power flow calculation failed: {str(e)}"} + + +def run_dc_power_flow() -> Dict[str, Any]: + """Run DC power flow analysis on the current network. + + Returns: + Dict containing DC power flow results + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + try: + net = _get_network() + pp.rundcpp(net) + + return { + "status": "success", + "message": "DC power flow completed", + "bus_results": { + "va_degree": net.res_bus["va_degree"].to_dict(), + "p_mw": net.res_bus["p_mw"].to_dict() + }, + "line_results": { + "p_from_mw": net.res_line["p_from_mw"].to_dict(), + "p_to_mw": net.res_line["p_to_mw"].to_dict() + } + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": f"DC power flow calculation failed: {str(e)}"} + + +def get_network_info() -> Dict[str, Any]: + """Get information about the current network. + + Returns: + Dict containing network statistics + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + try: + net = _get_network() + + info = { + "status": "success", + "component_counts": { + "buses": len(net.bus), + "lines": len(net.line), + "transformers": len(net.trafo), + "generators": len(net.gen), + "static_generators": len(net.sgen), + "loads": len(net.load), + "external_grids": len(net.ext_grid), + "shunts": len(net.shunt) if hasattr(net, 'shunt') else 0, + "switches": len(net.switch) if hasattr(net, 'switch') else 0 + }, + "bus_data": net.bus.to_dict() if len(net.bus) <= 50 else f"Too large ({len(net.bus)} buses)", + "line_data": net.line.to_dict() if len(net.line) <= 50 else f"Too large ({len(net.line)} lines)", + "load_data": net.load.to_dict() if len(net.load) <= 50 else f"Too large ({len(net.load)} loads)", + "gen_data": net.gen.to_dict() if len(net.gen) <= 50 else f"Too large ({len(net.gen)} generators)" + } + + return info + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": f"Failed to get network information: {str(e)}"} + + +def add_bus(name: str, vn_kv: float, bus_type: str = "b", + in_service: bool = True, max_vm_pu: float = 1.1, + min_vm_pu: float = 0.9) -> Dict[str, Any]: + """Add a bus to the current network. + + Args: + name: Name of the bus + vn_kv: Nominal voltage in kV + bus_type: Bus type ('b' for PQ bus, 'n' for node) + in_service: Whether bus is in service + max_vm_pu: Maximum voltage magnitude in per unit + min_vm_pu: Minimum voltage magnitude in per unit + + Returns: + Dict with status and new bus index + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + try: + net = _get_network() + bus_idx = pp.create_bus(net, vn_kv=vn_kv, name=name, type=bus_type, + in_service=in_service, max_vm_pu=max_vm_pu, min_vm_pu=min_vm_pu) + return { + "status": "success", + "message": f"Bus '{name}' added successfully", + "bus_index": int(bus_idx), + "total_buses": len(net.bus) + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": str(e)} + + +def add_line(from_bus: int, to_bus: int, length_km: float, + std_type: str = "NAYY 4x50 SE", name: str = "") -> Dict[str, Any]: + """Add a line to the current network. + + Args: + from_bus: Index of the starting bus + to_bus: Index of the ending bus + length_km: Length of the line in km + std_type: Standard line type + name: Name of the line + + Returns: + Dict with status and new line index + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + try: + net = _get_network() + line_idx = pp.create_line(net, from_bus=from_bus, to_bus=to_bus, + length_km=length_km, std_type=std_type, name=name) + return { + "status": "success", + "message": f"Line from bus {from_bus} to bus {to_bus} added", + "line_index": int(line_idx), + "total_lines": len(net.line) + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": str(e)} + + +def add_load(bus: int, p_mw: float, q_mvar: float = 0.0, name: str = "") -> Dict[str, Any]: + """Add a load to the current network. + + Args: + bus: Index of the bus to connect the load + p_mw: Active power of the load in MW + q_mvar: Reactive power of the load in Mvar + name: Name of the load + + Returns: + Dict with status and new load index + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + try: + net = _get_network() + load_idx = pp.create_load(net, bus=bus, p_mw=p_mw, q_mvar=q_mvar, name=name) + return { + "status": "success", + "message": f"Load added at bus {bus}", + "load_index": int(load_idx), + "total_loads": len(net.load) + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": str(e)} + + +def add_generator(bus: int, p_mw: float, vm_pu: float = 1.0, + name: str = "", controllable: bool = True) -> Dict[str, Any]: + """Add a generator to the current network. + + Args: + bus: Index of the bus to connect the generator + p_mw: Active power output in MW + vm_pu: Voltage setpoint in per unit + name: Name of the generator + controllable: Whether the generator is controllable + + Returns: + Dict with status and new generator index + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + try: + net = _get_network() + gen_idx = pp.create_gen(net, bus=bus, p_mw=p_mw, vm_pu=vm_pu, + name=name, controllable=controllable) + return { + "status": "success", + "message": f"Generator added at bus {bus}", + "generator_index": int(gen_idx), + "total_generators": len(net.gen) + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": str(e)} + + +def add_ext_grid(bus: int, vm_pu: float = 1.0, va_degree: float = 0.0, + name: str = "External Grid") -> Dict[str, Any]: + """Add an external grid (slack bus) to the current network. + + Args: + bus: Index of the bus to connect the external grid + vm_pu: Voltage magnitude setpoint in per unit + va_degree: Voltage angle in degrees + name: Name of the external grid + + Returns: + Dict with status and new external grid index + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + try: + net = _get_network() + ext_grid_idx = pp.create_ext_grid(net, bus=bus, vm_pu=vm_pu, + va_degree=va_degree, name=name) + return { + "status": "success", + "message": f"External grid added at bus {bus}", + "ext_grid_index": int(ext_grid_idx), + "total_ext_grids": len(net.ext_grid) + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": str(e)} + + +def run_contingency_analysis(contingency_type: str = "line", + element_indices: Optional[List[int]] = None) -> Dict[str, Any]: + """Run N-1 contingency analysis on the current network. + + Args: + contingency_type: Type of contingency ('line', 'trafo', or 'gen') + element_indices: List of element indices to analyze (None for all) + + Returns: + Dict containing contingency analysis results + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + try: + net = _get_network() + + # Determine indices to analyze + if element_indices is None: + if contingency_type == "line": + indices = list(net.line.index) + elif contingency_type == "trafo": + indices = list(net.trafo.index) + elif contingency_type == "gen": + indices = list(net.gen.index) + else: + return {"status": "error", "message": f"Unknown contingency type: {contingency_type}"} + else: + indices = element_indices + + results = [] + base_case_converged = False + base_losses = 0.0 + + # Run base case first + try: + pp.runpp(net) + base_case_converged = net.converged + base_losses = float(net.res_line["pl_mw"].sum()) + except: + base_case_converged = False + + # Store original state + orig_net = copy.deepcopy(net) + + # Run contingency for each element + for idx in indices: + contingency_net = copy.deepcopy(orig_net) + + try: + contingency_net[contingency_type].at[idx, 'in_service'] = False + pp.runpp(contingency_net) + + # Check for violations + voltage_violations = contingency_net.res_bus[ + (contingency_net.res_bus.vm_pu < 0.95) | + (contingency_net.res_bus.vm_pu > 1.05) + ].index.tolist() + + loading_violations = contingency_net.res_line[ + contingency_net.res_line.loading_percent > 100 + ].index.tolist() + + results.append({ + "contingency": f"{contingency_type}_{idx}", + "converged": contingency_net.converged, + "voltage_violations": voltage_violations, + "loading_violations": loading_violations, + "max_loading_percent": float(contingency_net.res_line["loading_percent"].max()), + "min_voltage_pu": float(contingency_net.res_bus["vm_pu"].min()), + "max_voltage_pu": float(contingency_net.res_bus["vm_pu"].max()) + }) + except Exception as e: + results.append({ + "contingency": f"{contingency_type}_{idx}", + "converged": False, + "error": str(e) + }) + + # Restore original network + pp.runpp(orig_net) + + return { + "status": "success", + "message": f"Contingency analysis completed for {len(indices)} {contingency_type}(s)", + "base_case_converged": base_case_converged, + "base_case_losses_mw": base_losses, + "contingency_results": results + } + + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": f"Contingency analysis failed: {str(e)}"} + + +def get_available_std_types() -> Dict[str, Any]: + """Get available standard types for lines and transformers. + + Returns: + Dict with available standard types + """ + if not PANDAPOWER_AVAILABLE: + return {"status": "error", "message": "pandapower is not installed"} + + try: + empty_net = pp.create_empty_network() + line_types = list(pp.available_std_types(empty_net, "line").index)[:20] + trafo_types = list(pp.available_std_types(empty_net, "trafo").index)[:20] + + return { + "status": "success", + "line_std_types_sample": line_types, + "trafo_std_types_sample": trafo_types, + "note": "Showing first 20 of each type. Many more available." + } + except Exception as e: + return {"status": "error", "message": str(e)} + + +# Export all public functions +__all__ = [ + 'PANDAPOWER_AVAILABLE', + 'create_empty_network', + 'create_test_network', + 'load_network', + 'run_power_flow', + 'run_dc_power_flow', + 'get_network_info', + 'add_bus', + 'add_line', + 'add_load', + 'add_generator', + 'add_ext_grid', + 'run_contingency_analysis', + 'get_available_std_types', +] diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..b47f3be --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,145 @@ +[build-system] +requires = ["setuptools>=65.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "powermcp" +version = "0.1.0" +description = "Open-source collection of MCP servers for power system software" +readme = "README.md" +requires-python = ">=3.10" +license = {text = "MIT"} +authors = [ + {name = "Qian Zhang"}, + {name = "Muhy Eddin Za'ter"}, + {name = "Stephen Jenkins"}, + {name = "Maanas Goel"}, +] +keywords = ["mcp", "power-systems", "simulation", "ai", "llm"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Science/Research", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Scientific/Engineering", + "Topic :: Software Development :: Libraries :: Python Modules", +] + +dependencies = [ + "fastmcp>=2.0.1", + "mcp>=1.0.0", +] + +[project.optional-dependencies] +# Individual power system tools +andes = [ + "andes>=1.9.0", +] +egret = [ + "egret", +] +opendss = [ + "opendssdirect.py>=0.8.0", +] +pandapower = [ + "pandapower>=2.13.0", +] +powerworld = [ + "pywin32>=306; sys_platform == 'win32'", +] +pslf = [ + # PSLF requires proprietary installation +] +psse = [ + # PSSE requires proprietary installation +] +pypsa = [ + "pypsa>=0.25.0", + "highspy>=1.5.0", + "networkx>=3.0", + "cartopy>=0.21.0", +] +pyltspice = [ + "PyLTSpice>=1.0.0", + "matplotlib>=3.5.0", +] + +# Development dependencies +dev = [ + "pytest>=7.0.0", + "pytest-cov>=4.0.0", + "pytest-asyncio>=0.21.0", + "pytest-mock>=3.10.0", + "pytest-timeout>=2.1.0", + "black>=23.0.0", + "flake8>=6.0.0", + "mypy>=1.0.0", + "pylint>=2.17.0", + "isort>=5.12.0", +] + +# Testing dependencies +test = [ + "pytest>=7.0.0", + "pytest-cov>=4.0.0", + "pytest-asyncio>=0.21.0", + "pytest-mock>=3.10.0", + "mock>=5.0.0", +] + +# Install all open-source tools +all-opensource = [ + "andes>=1.9.0", + "egret>=0.0.2", + "opendssdirect.py>=0.8.0", + "pandapower>=2.13.0", + "pypsa>=0.25.0", + "highspy>=1.5.0", + "networkx>=3.0", + "PyLTSpice>=1.0.0", + "matplotlib>=3.5.0", +] + +# Install everything including dev tools +all = [ + "powermcp[all-opensource,dev,test]", +] + +[project.urls] +Homepage = "https://github.com/Power-Agent/PowerMCP" +Documentation = "https://power-agent.github.io/" +Repository = "https://github.com/Power-Agent/PowerMCP" +"Bug Tracker" = "https://github.com/Power-Agent/PowerMCP/issues" + +[tool.setuptools] +packages = ["common", "ANDES", "Egret", "OpenDSS", "pandapower_tools", "PowerWorld", "PSLF", "PSSE", "PSSE35", "PyLTSpice", "pypsa_tools"] + +[tool.setuptools.package-data] +"*" = ["*.json", "*.dss", "*.csv", "*.pwb", "*.pwd", "*.sav", "*.dyr", "*.dyd", "*.m", "*.nc", "*.otg", "*.cntl", "*.dycr", "*.con", "*.mon", "*.sub"] + +[tool.black] +line-length = 100 +target-version = ['py310'] + +[tool.isort] +profile = "black" +line_length = 100 + +[tool.pytest.ini_options] +minversion = "7.0" +testpaths = ["tests"] +python_files = ["test_*.py"] +python_classes = ["Test*"] +python_functions = ["test_*"] +addopts = "-v --cov=. --cov-report=term-missing" +asyncio_mode = "auto" + +[tool.mypy] +python_version = "3.10" +warn_return_any = true +warn_unused_configs = true +disallow_untyped_defs = false +ignore_missing_imports = true diff --git a/PyPSA/README.md b/pypsa_tools/README.md similarity index 100% rename from PyPSA/README.md rename to pypsa_tools/README.md diff --git a/pypsa_tools/__init__.py b/pypsa_tools/__init__.py new file mode 100644 index 0000000..e61943a --- /dev/null +++ b/pypsa_tools/__init__.py @@ -0,0 +1,29 @@ +"""PowerMCP PyPSA Tools - Power system optimization using PyPSA.""" + +from .tools import ( + PYPSA_AVAILABLE, + create_network, + get_network_info, + add_bus, + add_generator, + add_load, + add_line, + run_power_flow, + run_optimal_power_flow, + load_network, + save_network, +) + +__all__ = [ + 'PYPSA_AVAILABLE', + 'create_network', + 'get_network_info', + 'add_bus', + 'add_generator', + 'add_load', + 'add_line', + 'run_power_flow', + 'run_optimal_power_flow', + 'load_network', + 'save_network', +] diff --git a/PyPSA/pypsa_mcp.py b/pypsa_tools/pypsa_mcp.py similarity index 100% rename from PyPSA/pypsa_mcp.py rename to pypsa_tools/pypsa_mcp.py diff --git a/PyPSA/requirements.txt b/pypsa_tools/requirements.txt similarity index 100% rename from PyPSA/requirements.txt rename to pypsa_tools/requirements.txt diff --git a/PyPSA/tests/test_pypsa_mcp.py b/pypsa_tools/tests/test_pypsa_mcp.py similarity index 100% rename from PyPSA/tests/test_pypsa_mcp.py rename to pypsa_tools/tests/test_pypsa_mcp.py diff --git a/pypsa_tools/tools.py b/pypsa_tools/tools.py new file mode 100644 index 0000000..1e86bd2 --- /dev/null +++ b/pypsa_tools/tools.py @@ -0,0 +1,341 @@ +""" +PowerMCP PyPSA Module +Provides power system optimization tools using PyPSA. +""" + +from typing import Dict, List, Optional, Any +import json + +try: + import pypsa + PYPSA_AVAILABLE = True +except ImportError: + PYPSA_AVAILABLE = False + pypsa = None + +# Global variable to store the current network +_current_net = None + + +def _get_network(): + """Get the current PyPSA network instance.""" + global _current_net + if _current_net is None: + raise RuntimeError("No PyPSA network is currently loaded. Please create or load a network first.") + return _current_net + + +def _set_network(net): + """Set the current PyPSA network instance.""" + global _current_net + _current_net = net + + +def create_network(name: str = "PyPSA Network") -> Dict[str, Any]: + """Create a new PyPSA network. + + Args: + name: Name of the network + + Returns: + Dict containing status and network information + """ + if not PYPSA_AVAILABLE: + return {"status": "error", "message": "pypsa is not installed"} + + global _current_net + try: + _current_net = pypsa.Network(name=name) + return { + "status": "success", + "message": f"PyPSA network '{name}' created successfully", + "network_name": name + } + except Exception as e: + return {"status": "error", "message": str(e)} + + +def get_network_info() -> Dict[str, Any]: + """Get information about the current PyPSA network. + + Returns: + Dict containing network statistics + """ + if not PYPSA_AVAILABLE: + return {"status": "error", "message": "pypsa is not installed"} + + try: + net = _get_network() + return { + "status": "success", + "network_name": net.name, + "component_counts": { + "buses": len(net.buses), + "generators": len(net.generators), + "loads": len(net.loads), + "lines": len(net.lines), + "transformers": len(net.transformers), + "storage_units": len(net.storage_units) + } + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": str(e)} + + +def add_bus(bus_id: str, v_nom: float = 380.0, x: Optional[float] = None, + y: Optional[float] = None, carrier: str = "AC") -> Dict[str, Any]: + """Add a bus to the current PyPSA network. + + Args: + bus_id: Unique identifier for the bus + v_nom: Nominal voltage in kV + x: X coordinate (optional) + y: Y coordinate (optional) + carrier: Energy carrier (default: AC) + + Returns: + Dict with status + """ + if not PYPSA_AVAILABLE: + return {"status": "error", "message": "pypsa is not installed"} + + try: + net = _get_network() + net.add("Bus", bus_id, v_nom=v_nom, x=x, y=y, carrier=carrier) + return { + "status": "success", + "message": f"Bus '{bus_id}' added to network", + "total_buses": len(net.buses) + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": str(e)} + + +def add_generator(gen_id: str, bus: str, p_nom: float, + marginal_cost: float = 0.0, carrier: str = "generator", + p_min_pu: float = 0.0, p_max_pu: float = 1.0) -> Dict[str, Any]: + """Add a generator to the current PyPSA network. + + Args: + gen_id: Unique identifier for the generator + bus: Bus to connect generator to + p_nom: Nominal power in MW + marginal_cost: Marginal cost + carrier: Energy carrier + p_min_pu: Minimum power output (per unit) + p_max_pu: Maximum power output (per unit) + + Returns: + Dict with status + """ + if not PYPSA_AVAILABLE: + return {"status": "error", "message": "pypsa is not installed"} + + try: + net = _get_network() + net.add("Generator", gen_id, bus=bus, p_nom=p_nom, + marginal_cost=marginal_cost, carrier=carrier, + p_min_pu=p_min_pu, p_max_pu=p_max_pu) + return { + "status": "success", + "message": f"Generator '{gen_id}' added to network", + "total_generators": len(net.generators) + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": str(e)} + + +def add_load(load_id: str, bus: str, p_set: float) -> Dict[str, Any]: + """Add a load to the current PyPSA network. + + Args: + load_id: Unique identifier for the load + bus: Bus to connect load to + p_set: Active power consumption in MW + + Returns: + Dict with status + """ + if not PYPSA_AVAILABLE: + return {"status": "error", "message": "pypsa is not installed"} + + try: + net = _get_network() + net.add("Load", load_id, bus=bus, p_set=p_set) + return { + "status": "success", + "message": f"Load '{load_id}' added to network", + "total_loads": len(net.loads) + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": str(e)} + + +def add_line(line_id: str, bus0: str, bus1: str, x: float, + r: float = 0.0, s_nom: float = 1000.0) -> Dict[str, Any]: + """Add a line to the current PyPSA network. + + Args: + line_id: Unique identifier for the line + bus0: From bus + bus1: To bus + x: Reactance in Ohms + r: Resistance in Ohms + s_nom: Nominal power in MVA + + Returns: + Dict with status + """ + if not PYPSA_AVAILABLE: + return {"status": "error", "message": "pypsa is not installed"} + + try: + net = _get_network() + net.add("Line", line_id, bus0=bus0, bus1=bus1, x=x, r=r, s_nom=s_nom) + return { + "status": "success", + "message": f"Line '{line_id}' added to network", + "total_lines": len(net.lines) + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": str(e)} + + +def run_power_flow() -> Dict[str, Any]: + """Run power flow analysis on the current PyPSA network. + + Returns: + Dict containing power flow results + """ + if not PYPSA_AVAILABLE: + return {"status": "error", "message": "pypsa is not installed"} + + try: + net = _get_network() + net.pf() + + return { + "status": "success", + "message": "Power flow completed", + "bus_results": { + "v_mag_pu": net.buses_t.v_mag_pu.to_dict() if hasattr(net.buses_t, 'v_mag_pu') and len(net.buses_t.v_mag_pu) > 0 else {}, + "v_ang": net.buses_t.v_ang.to_dict() if hasattr(net.buses_t, 'v_ang') and len(net.buses_t.v_ang) > 0 else {} + }, + "line_results": { + "p0": net.lines_t.p0.to_dict() if hasattr(net.lines_t, 'p0') and len(net.lines_t.p0) > 0 else {}, + "p1": net.lines_t.p1.to_dict() if hasattr(net.lines_t, 'p1') and len(net.lines_t.p1) > 0 else {} + } + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": f"Power flow failed: {str(e)}"} + + +def run_optimal_power_flow() -> Dict[str, Any]: + """Run optimal power flow on the current PyPSA network. + + Returns: + Dict containing OPF results + """ + if not PYPSA_AVAILABLE: + return {"status": "error", "message": "pypsa is not installed"} + + try: + net = _get_network() + status, termination_condition = net.optimize() + + return { + "status": "success" if status == "ok" else "warning", + "message": f"OPF completed with status: {status}", + "termination_condition": str(termination_condition), + "objective_value": float(net.objective) if hasattr(net, 'objective') else None, + "generator_dispatch": net.generators_t.p.to_dict() if hasattr(net.generators_t, 'p') and len(net.generators_t.p) > 0 else {} + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": f"OPF failed: {str(e)}"} + + +def load_network(file_path: str) -> Dict[str, Any]: + """Load a PyPSA network from a file. + + Args: + file_path: Path to the network file (.nc or .h5) + + Returns: + Dict containing status and network information + """ + if not PYPSA_AVAILABLE: + return {"status": "error", "message": "pypsa is not installed"} + + global _current_net + try: + _current_net = pypsa.Network(file_path) + return { + "status": "success", + "message": f"Network loaded from {file_path}", + "network_name": _current_net.name, + "component_counts": { + "buses": len(_current_net.buses), + "generators": len(_current_net.generators), + "loads": len(_current_net.loads), + "lines": len(_current_net.lines) + } + } + except FileNotFoundError: + return {"status": "error", "message": f"File not found: {file_path}"} + except Exception as e: + return {"status": "error", "message": f"Failed to load network: {str(e)}"} + + +def save_network(file_path: str) -> Dict[str, Any]: + """Save the current PyPSA network to a file. + + Args: + file_path: Path to save the network (.nc for NetCDF) + + Returns: + Dict with status + """ + if not PYPSA_AVAILABLE: + return {"status": "error", "message": "pypsa is not installed"} + + try: + net = _get_network() + net.export_to_netcdf(file_path) + return { + "status": "success", + "message": f"Network saved to {file_path}" + } + except RuntimeError as re: + return {"status": "error", "message": str(re)} + except Exception as e: + return {"status": "error", "message": f"Failed to save network: {str(e)}"} + + +# Export all public functions +__all__ = [ + 'PYPSA_AVAILABLE', + 'create_network', + 'get_network_info', + 'add_bus', + 'add_generator', + 'add_load', + 'add_line', + 'run_power_flow', + 'run_optimal_power_flow', + 'load_network', + 'save_network', +] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..01b0b1b --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +# Core MCP dependencies +fastmcp>=2.0.1 +mcp>=1.0.0 + +# For development and testing +# Install with: pip install -e .[dev] diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..80a09f2 --- /dev/null +++ b/setup.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python +""" +Setup script for PowerMCP +""" + +from setuptools import setup, find_packages +from pathlib import Path + +# Read the README file +this_directory = Path(__file__).parent +long_description = (this_directory / "README.md").read_text(encoding='utf-8') + +setup( + name="powermcp", + version="0.1.0", + description="Open-source collection of MCP servers for power system software", + long_description=long_description, + long_description_content_type="text/markdown", + author="PowerMCP Team", + author_email="", + url="https://github.com/Power-Agent/PowerMCP", + project_urls={ + "Documentation": "https://power-agent.github.io/", + "Source": "https://github.com/Power-Agent/PowerMCP", + "Bug Tracker": "https://github.com/Power-Agent/PowerMCP/issues", + }, + packages=find_packages(exclude=["tests", "tests.*"]), + python_requires=">=3.10", + install_requires=[ + "fastmcp>=2.0.1", + "mcp>=1.0.0", + ], + extras_require={ + # Individual power system tools + "andes": ["andes>=1.9.0"], + "egret": ["egret>=0.0.2"], + "opendss": ["opendssdirect.py>=0.8.0"], + "pandapower": ["pandapower>=2.13.0"], + "powerworld": ["pywin32>=306; sys_platform == 'win32'"], + "pypsa": [ + "pypsa>=0.25.0", + "highspy>=1.5.0", + "networkx>=3.0", + "cartopy>=0.21.0", + ], + "pyltspice": [ + "PyLTSpice>=1.0.0", + "matplotlib>=3.5.0", + ], + # Development dependencies + "dev": [ + "pytest>=7.0.0", + "pytest-cov>=4.0.0", + "pytest-asyncio>=0.21.0", + "pytest-mock>=3.10.0", + "pytest-timeout>=2.1.0", + "black>=23.0.0", + "flake8>=6.0.0", + "mypy>=1.0.0", + "pylint>=2.17.0", + "isort>=5.12.0", + ], + # Testing dependencies + "test": [ + "pytest>=7.0.0", + "pytest-cov>=4.0.0", + "pytest-asyncio>=0.21.0", + "pytest-mock>=3.10.0", + "mock>=5.0.0", + ], + # Install all open-source tools + "all-opensource": [ + "andes>=1.9.0", + "egret>=0.0.2", + "opendssdirect.py>=0.8.0", + "pandapower>=2.13.0", + "pypsa>=0.25.0", + "highspy>=1.5.0", + "networkx>=3.0", + "PyLTSpice>=1.0.0", + "matplotlib>=3.5.0", + ], + }, + package_data={ + "": ["*.json", "*.dss", "*.csv", "*.pwb", "*.pwd", "*.sav", "*.dyr", + "*.dyd", "*.m", "*.nc", "*.otg", "*.cntl", "*.dycr", "*.con", + "*.mon", "*.sub"], + }, + include_package_data=True, + classifiers=[ + "Development Status :: 3 - Alpha", + "Intended Audience :: Science/Research", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Scientific/Engineering", + "Topic :: Software Development :: Libraries :: Python Modules", + ], + keywords="mcp power-systems simulation ai llm", + license="MIT", +)