Add a national end point, that provides elexon BMRS solar foreacsts from elexonpy package

requirements.txt

@@ -19,3 +19,4 @@ slowapi
 pathy==0.10.3
 fsspec
 s3fs
+elexonpy

src/main.py

@@ -208,7 +208,6 @@ async def add_process_time_header(request: Request, call_next):
 # Dependency
 v0_route_solar = "/v0/solar/GB"
 v0_route_system = "/v0/system/GB"
-
 app.include_router(national_router, prefix=f"{v0_route_solar}/national")
 app.include_router(gsp_router, prefix=f"{v0_route_solar}/gsp")
 app.include_router(status_router, prefix=f"{v0_route_solar}")

src/national.py

@@ -1,10 +1,14 @@
 """National API routes"""
 
 import os
+from datetime import datetime, timedelta
 from typing import List, Optional, Union
 
+import pandas as pd
 import structlog
-from fastapi import APIRouter, Depends, HTTPException, Request, Security
+from elexonpy.api.generation_forecast_api import GenerationForecastApi
+from elexonpy.api_client import ApiClient
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, Security
 from fastapi_auth0 import Auth0User
 from nowcasting_datamodel.read.read import get_latest_forecast_for_gsps
 from sqlalchemy.orm.session import Session
@@ -21,14 +25,64 @@
 
 logger = structlog.stdlib.get_logger()
 
-
 adjust_limit = float(os.getenv("ADJUST_MW_LIMIT", 0.0))
 get_plevels = bool(os.getenv("GET_PLEVELS", True))
 
 router = APIRouter(
     tags=["National"],
 )
 
+# Initialize Elexon API client
+api_client = ApiClient()
+forecast_api = GenerationForecastApi(api_client)
+
+
+@router.get("/elexon", summary="Get elexon Solar Forecast")
+@limiter.limit(f"{N_CALLS_PER_HOUR}/hour")
+def get_elexon_forecast(
+    request: Request,
+    start_datetime_utc: datetime = Query(
+        default=datetime.utcnow() - timedelta(days=3), description="Start date and time in UTC"
+    ),
+    end_datetime_utc: datetime = Query(
+        default=datetime.utcnow() + timedelta(days=3), description="End date and time in UTC"
+    ),
+    process_type: str = Query("Day Ahead", description="Process type"),
+):
+    """
+    Fetch elexon Solar and wind(?) forecasts from the Elexon API.
+
+    Args:
+        request (Request): The request object containing metadata about the HTTP request.
+        start_datetime_utc (datetime): The start date and time in UTC.
+        end_datetime_utc (datetime): The end date and time in UTC.
+        process_type (str): The type of process (e.g., 'Day Ahead').
+
+    Returns:
+        dict: Dictionary containing the fetched elexon solar forecast data.
+
+    External API:
+        [Elexon API Documentation]
+        (https://elexon.elexon.co.uk/api-documentation/endpoint/forecast/generation/wind-and-solar/day-ahead)
+    """
+    response = forecast_api.forecast_generation_wind_and_solar_day_ahead_get(
+        _from=start_datetime_utc.isoformat(),
+        to=end_datetime_utc.isoformat(),
+        process_type=process_type,
+        format="json",
+    )
+
+    if not response.data:
+        return {"data": []}
+
+    df = pd.DataFrame([item.to_dict() for item in response.data])
+
+    # Filter to include only solar forecasts
+    solar_df = df[df["business_type"] == "Solar generation"]
+    result = {"data": solar_df.to_dict(orient="records")}
+
+    return result
+
 
 @router.get(
     "/forecast",
@@ -47,7 +101,9 @@ def get_national_forecast(
     end_datetime_utc: Optional[str] = None,
     creation_limit_utc: Optional[str] = None,
 ) -> Union[NationalForecast, List[NationalForecastValue]]:
-    """Get the National Forecast
+    """
+
+    Fetch national forecasts.
 
     This route returns the most recent forecast for each _target_time_.
 
@@ -67,6 +123,9 @@ def get_national_forecast(
     - **creation_utc_limit**: optional, only return forecasts made before this datetime.
     Note you can only go 7 days back at the moment
 
+    Returns:
+        dict: The national forecast data.
+
     """
     logger.debug("Get national forecasts")
 
@@ -182,35 +241,10 @@ def get_national_pvlive(
 
     #### Parameters
     - **regime**: can choose __in-day__ or __day-after__
-    """
 
+    """
     logger.info(f"Get national PV Live estimates values " f"for regime {regime} for  {user}")
 
     return get_truth_values_for_a_specific_gsp_from_database(
         session=session, gsp_id=0, regime=regime
     )
-
-
-@router.get(
-    "/bmrs",
-    response_model=dict,
-    # dependencies=[Depends(get_auth_implicit_scheme())],
-    summary="Get BMRS Forecast",
-)
-# @cache_response
-@limiter.limit(f"{N_CALLS_PER_HOUR}/hour")
-def get_bmrs_forecast(
-    request: Request,
-    # session: Session = Depends(get_session),
-    # user: Auth0User = Security(get_user()),
-) -> dict:
-    """
-
-    This route returns the most recent BMRS forecast for each _target_time_.
-
-    #### Parameters
-
-    """
-    logger.debug("Get bmrs forecast")
-
-    return {"message": "This route is not yet implemented. Please check back later."}

src/tests/test_elexon_forecast.py

@@ -0,0 +1,109 @@
+from unittest.mock import MagicMock, patch
+
+from fastapi.testclient import TestClient
+
+from main import app
+
+client = TestClient(app)
+
+
+class MockForecastItem:
+    def __init__(self, **entries):
+        self.__dict__.update(entries)
+
+    def to_dict(self):
+        return self.__dict__
+
+
+mock_data = [
+    MockForecastItem(
+        publish_time="2024-07-24T16:45:09+00:00",
+        process_type="Day ahead",
+        business_type="Solar generation",
+        psr_type="Solar",
+        start_time="2024-07-25T23:30:00+00:00",
+        settlement_date="2024-07-26",
+        settlement_period=2,
+        quantity=0,
+    ),
+    MockForecastItem(
+        publish_time="2024-07-24T16:45:09+00:00",
+        process_type="Day ahead",
+        business_type="Solar generation",
+        psr_type="Solar",
+        start_time="2024-07-25T23:00:00+00:00",
+        settlement_date="2024-07-26",
+        settlement_period=1,
+        quantity=0,
+    ),
+]
+
+# Define the full path for the patch target
+PATCH_TARGET = (
+    "elexonpy.api.generation_forecast_api."
+    "GenerationForecastApi.forecast_generation_wind_and_solar_day_ahead_get"
+)
+
+
+def setup_mock_forecast(mock_forecast_api, data):
+    mock_response = MagicMock()
+    mock_response.data = data
+    mock_forecast_api.return_value = mock_response
+
+
+@patch(PATCH_TARGET)
+def test_get_elexon_forecast_with_data(mock_forecast_api):
+    setup_mock_forecast(mock_forecast_api, mock_data)
+    endpoint = "/v0/solar/GB/national/elexon"
+    params = {
+        "start_datetime_utc": "2024-07-22T10:56:59.194610",
+        "end_datetime_utc": "2024-07-28T10:56:59.194680",
+        "process_type": "Day Ahead",
+    }
+
+    response = client.get(endpoint, params=params)
+
+    expected_response = {
+        "data": [
+            {
+                "publish_time": "2024-07-24T16:45:09+00:00",
+                "process_type": "Day ahead",
+                "business_type": "Solar generation",
+                "psr_type": "Solar",
+                "start_time": "2024-07-25T23:30:00+00:00",
+                "settlement_date": "2024-07-26",
+                "settlement_period": 2,
+                "quantity": 0,
+            },
+            {
+                "publish_time": "2024-07-24T16:45:09+00:00",
+                "process_type": "Day ahead",
+                "business_type": "Solar generation",
+                "psr_type": "Solar",
+                "start_time": "2024-07-25T23:00:00+00:00",
+                "settlement_date": "2024-07-26",
+                "settlement_period": 1,
+                "quantity": 0,
+            },
+        ]
+    }
+
+    assert response.status_code == 200
+    assert response.headers["Content-Type"] == "application/json"
+    assert response.json() == expected_response
+
+
+@patch(PATCH_TARGET)
+def test_get_elexon_forecast_no_data(mock_forecast_api):
+    setup_mock_forecast(mock_forecast_api, [])
+
+    endpoint = "/v0/solar/GB/national/elexon"
+    params = {
+        "start_datetime_utc": "2024-07-22T10:56:59.194610",
+        "end_datetime_utc": "2024-07-28T10:56:59.194680",
+        "process_type": "Day Ahead",
+    }
+    response = client.get(endpoint, params=params)
+    assert response.status_code == 200
+    assert response.headers["Content-Type"] == "application/json"
+    assert response.json() == {"data": []}