get_grid_stability
Analyzes grid stability by comparing synchronous generation (provides inertia) with variable renewables (no inertia) at specified times to assess stability risks in Spain's electrical grid.
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 |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/ree_mcp/interface/mcp_server.py:517-552 (registration)Registration of the get_grid_stability tool using @mcp.tool() decorator and the wrapper handler function that handles input parameters, creates services, and formats response.
@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 handler logic in GridStabilityService.get_grid_stability that fetches generation data for synchronous and variable renewables, computes stability metrics including percentages, inertia ratio, and stability level assessment.
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 - GridStabilityService class providing the data fetching and analysis capabilities used by the tool handler.
class GridStabilityService: