get_generation_mix_timeline
Track electricity generation mix over time to analyze energy source patterns and visualize transition trends by hour or day intervals.
Instructions
Get generation mix over time for a full day or period.
Returns generation breakdown by source across multiple time points, useful for visualizing energy transition patterns.
Args: date: Date in YYYY-MM-DD format time_granularity: Time aggregation (hour or day, default: hour)
Returns: JSON string with generation mix timeline.
Examples: Get hourly generation mix for a day: >>> await get_generation_mix_timeline("2025-10-08", "hour")
Get daily generation mix for a month:
>>> await get_generation_mix_timeline("2025-10-01", "day")
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| date | Yes | ||
| time_granularity | No | hour |
Implementation Reference
- MCP tool handler decorated with @mcp.tool(), which registers the tool and executes the main logic: parses date to date range, creates GenerationMixService instance, calls its get_generation_mix_timeline method, and returns formatted JSON response or error.@mcp.tool() async def get_generation_mix_timeline(date: str, time_granularity: str = "hour") -> str: """Get generation mix over time for a full day or period. Returns generation breakdown by source across multiple time points, useful for visualizing energy transition patterns. Args: date: Date in YYYY-MM-DD format time_granularity: Time aggregation (hour or day, default: hour) Returns: JSON string with generation mix timeline. Examples: Get hourly generation mix for a day: >>> await get_generation_mix_timeline("2025-10-08", "hour") Get daily generation mix for a month: >>> await get_generation_mix_timeline("2025-10-01", "day") """ try: start_date, end_date = DateTimeHelper.build_day_range(date) async with ToolExecutor() as executor: use_case = executor.create_get_indicator_data_use_case() data_fetcher = DataFetcher(use_case) service = GenerationMixService(data_fetcher) result = await service.get_generation_mix_timeline( start_date, end_date, time_granularity ) return ResponseFormatter.success(result) except Exception as e: return ResponseFormatter.unexpected_error(e, context="Error getting generation timeline")
- src/ree_mcp/interface/mcp_server.py:554-554 (registration)The @mcp.tool() decorator registers get_generation_mix_timeline as an MCP tool.@mcp.tool()
- Helper method in GenerationMixService class that implements the core logic: fetches data for multiple generation sources using DataFetcher, aligns time series data, computes per-timepoint source breakdowns and totals, structures as timeline dictionary.async def get_generation_mix_timeline( self, start_date: str, end_date: str, time_granularity: str = "hour" ) -> dict[str, Any]: """Get generation mix over a period. Args: start_date: Start datetime in ISO format end_date: End datetime in ISO format time_granularity: Time aggregation level Returns: Timeline data with generation mix at each point """ sources = IndicatorIDs.get_generation_mix_sources() raw_data = await self.data_fetcher.fetch_multiple_indicators( sources, start_date, end_date, time_granularity ) result: dict[str, Any] = { "period": { "start": start_date, "end": end_date, "granularity": time_granularity, }, "timeline": [], } # Build timeline by combining data points source_data: dict[str, list[dict[str, Any]]] = {} for source_name, response_data in raw_data.items(): if "error" not in response_data: source_data[source_name] = response_data.get("values", []) else: source_data[source_name] = [] if source_data: # Use first source to get timestamps first_source_values = next(iter(source_data.values())) for i, value_point in enumerate(first_source_values): timestamp = value_point["datetime"] generation_point: dict[str, Any] = { "datetime": timestamp, "sources": {}, "total_mw": 0.0, } for source_name, values in source_data.items(): if i < len(values): mw_value = values[i]["value"] generation_point["sources"][source_name] = mw_value generation_point["total_mw"] += mw_value else: generation_point["sources"][source_name] = 0.0 generation_point["total_mw"] = round(generation_point["total_mw"], 2) result["timeline"].append(generation_point) return result