get_grid_stability
Analyze grid stability by comparing synchronous generation inertia with variable renewables to assess stability risks for specific dates and times.
Instructions
Get grid stability metrics at a specific time.
Analyzes synchronous generation (provides inertia) vs variable renewables (no inertia) to assess grid stability risk.
Args: date: Date in YYYY-MM-DD format hour: Hour in HH format (00-23, default: 12)
Returns: JSON string with grid stability analysis.
Examples: Get grid stability at noon: >>> await get_grid_stability("2025-10-08", "12")
Check overnight stability:
>>> await get_grid_stability("2025-10-08", "02")
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| date | Yes | ||
| hour | No | 12 |
Implementation Reference
- MCP tool handler for get_grid_stability, decorated with @mcp.tool(). Parses input parameters, creates services, delegates to GridStabilityService, and formats the response as JSON string.@mcp.tool() async def get_grid_stability(date: str, hour: str = "12") -> str: """Get grid stability metrics at a specific time. Analyzes synchronous generation (provides inertia) vs variable renewables (no inertia) to assess grid stability risk. Args: date: Date in YYYY-MM-DD format hour: Hour in HH format (00-23, default: 12) Returns: JSON string with grid stability analysis. Examples: Get grid stability at noon: >>> await get_grid_stability("2025-10-08", "12") Check overnight stability: >>> await get_grid_stability("2025-10-08", "02") """ try: start_datetime, end_datetime = DateTimeHelper.build_datetime_range(date, hour) async with ToolExecutor() as executor: use_case = executor.create_get_indicator_data_use_case() data_fetcher = DataFetcher(use_case) service = GridStabilityService(data_fetcher) result = await service.get_grid_stability(start_datetime, end_datetime) return ResponseFormatter.success(result) except Exception as e: return ResponseFormatter.unexpected_error(e, context="Error getting grid stability")
- Core implementation in GridStabilityService.get_grid_stability: fetches synchronous generation and variable renewables data, computes totals, percentages relative to demand, inertia ratio, and assigns stability level based on synchronous generation percentage.async def get_grid_stability(self, start_date: str, end_date: str) -> dict[str, Any]: """Get grid stability metrics. Args: start_date: Start datetime in ISO format end_date: End datetime in ISO format Returns: Grid stability analysis with synchronous/variable breakdown """ synchronous = IndicatorIDs.get_synchronous_sources() variable_renewables = IndicatorIDs.get_variable_renewable_sources() # Fetch all data sync_data = await self.data_fetcher.fetch_multiple_indicators( synchronous, start_date, end_date, "hour" ) var_data = await self.data_fetcher.fetch_multiple_indicators( variable_renewables, start_date, end_date, "hour" ) result: dict[str, Any] = { "datetime": start_date, "synchronous_generation": {}, "variable_renewables": {}, "analysis": {}, } # Process synchronous generation total_synchronous_mw = 0.0 for source_name, response_data in sync_data.items(): if "error" not in response_data: values = response_data.get("values", []) if values: value_mw = values[0]["value"] result["synchronous_generation"][source_name] = {"value_mw": value_mw} total_synchronous_mw += value_mw else: result["synchronous_generation"][source_name] = {"error": "No data"} else: result["synchronous_generation"][source_name] = response_data # Process variable renewables total_variable_mw = 0.0 for source_name, response_data in var_data.items(): if "error" not in response_data: values = response_data.get("values", []) if values: value_mw = values[0]["value"] result["variable_renewables"][source_name] = {"value_mw": value_mw} total_variable_mw += value_mw else: result["variable_renewables"][source_name] = {"error": "No data"} else: result["variable_renewables"][source_name] = response_data # Calculate analysis metrics demand_mw = await self.data_fetcher.fetch_value_at_time( IndicatorIDs.REAL_DEMAND_NATIONAL, start_date, end_date, "hour" ) if demand_mw and demand_mw > 0: synchronous_pct = (total_synchronous_mw / demand_mw) * 100 variable_pct = (total_variable_mw / demand_mw) * 100 inertia_ratio = ( (total_synchronous_mw / total_variable_mw) if total_variable_mw > 0 else float("inf") ) # Stability assessment if synchronous_pct >= 70: stability_level = "excellent" elif synchronous_pct >= 50: stability_level = "good" elif synchronous_pct >= 30: stability_level = "moderate" else: stability_level = "concerning" result["analysis"] = { "total_synchronous_mw": round(total_synchronous_mw, 2), "total_variable_renewable_mw": round(total_variable_mw, 2), "total_demand_mw": round(demand_mw, 2), "synchronous_percentage": round(synchronous_pct, 2), "variable_renewable_percentage": round(variable_pct, 2), "inertia_ratio": ( round(inertia_ratio, 2) if inertia_ratio != float("inf") else "infinite" ), "stability_level": stability_level, "interpretation": { "excellent": ">=70% synchronous (high inertia)", "good": "50-70% synchronous (adequate inertia)", "moderate": "30-50% synchronous (requires monitoring)", "concerning": "<30% synchronous (stability risk)", }, } else: result["analysis"] = {"error": "Could not calculate analysis: No demand data"} return result
- src/ree_mcp/interface/mcp_server.py:517-517 (registration)The @mcp.tool() decorator registers the get_grid_stability function as an MCP tool in the FastMCP server.@mcp.tool()