Added resource for historic-generation-mix

Author: atsangarides

README.md

@@ -25,9 +25,13 @@ r: bytes = client.query(date_col=date_col, start_date=start_date, end_date=end_d
 
 ## Tested reports
 
+### Queryable via NG's api
 * `historic-day-ahead-demand-forecast`
 * `day-ahead-demand-forecast`
 * `historic-2day-ahead-demand-forecast`
 * `2day-ahead-demand-forecast`
 * `historic-day-ahead-wind-forecast`
 * `day-ahead-wind-forecast`
+
+### Download of files
+* `historic-generation-mix`

pyngeso/__init__.py

@@ -1,3 +1,3 @@
 from .pyngeso import NgEso
 
-__version__ = '0.2.1'
+__version__ = '0.2.2'

pyngeso/pyngeso.py

@@ -1,27 +1,48 @@
 import logging
-from typing import Optional, List
+from typing import Optional, List, Literal
 from datetime import datetime, timedelta
 import json
 import re
 
 import requests
 
 from .configure_logging import setup_logger
-from .resources import resource_ids
+from .resources import api_resource_ids, file_resource_ids
 from .exceptions import UnsuccessfulRequest
 
 logger = setup_logger(logging.getLogger("PyNgEso"))
 
 
 class NgEso:
+    """
+    A class for fetching data from the National Grid ESO data portal.
+
+    Args:
+        resource_id (str): id for the resource when using the ESO API functionality
+        resource (str): name of the resource when using the ESO API functionality
+    Returns:
+
+    """
     def __init__(
             self,
             resource: str,
-            dataset: Optional[str] = None
+            backend: Literal['api', 'file'] = "api"
     ):
         self.resource = resource
-        self.dataset = dataset
-        self.resource_id = resource_ids.get(resource).get("id")
+        self.backend = backend
+
+        self.resource_id, self.dataset_id, self.filename = self.set_resource_info()
+
+    def set_resource_info(self) -> (str, str, str):
+        dataset_id = None
+        filename = None
+        if self.backend == "api":
+            resource_id = api_resource_ids.get(self.resource).get("id")
+        else:
+            dataset_id = file_resource_ids.get(self.resource).get("dataset_id")
+            resource_id = file_resource_ids.get(self.resource).get("resource_id")
+            filename = file_resource_ids.get(self.resource).get("filename")
+        return resource_id, dataset_id, filename
 
     def query(
         self,
@@ -120,19 +141,22 @@ def validate_date_range(start_date: str, end_date: str) -> None:
             end_date >= start_date
         ), "end_date should be the same of greater than start_date"
 
-    @staticmethod
-    def _check_for_errors(r: requests.Response) -> None:
+    def _check_for_errors(self, r: requests.Response) -> None:
         """Inspect the request response and the metadata in xml"""
         # http response errors
-        status_code = r.status_code
-        if status_code != 200 or r.content is None:
-            raise UnsuccessfulRequest(f"status_code={status_code}:{r.content}")
+        self._check_request_errors(r)
 
         # inspect response body
         rb: dict = json.loads(r.content)
         if not rb.get("success"):
             logger.error(f"Request failed: {rb.get('error')}")
 
+    @staticmethod
+    def _check_request_errors(r: requests.Response) -> None:
+        status_code = r.status_code
+        if status_code != 200 or r.content is None:
+            raise UnsuccessfulRequest(f"status_code={status_code}:{r.content}")
+
     @staticmethod
     def _missing_data(r: requests.Response) -> None:
         """
@@ -144,3 +168,13 @@ def _missing_data(r: requests.Response) -> None:
         query = rb.get("query")
         if not records:
             logger.warning(f"{query}: No data found")
+
+    def download_file(self) -> bytes:
+        url = (
+            f"https://data.nationalgrideso.com/backend/dataset/{self.dataset_id}/"
+            f"resource/{self.resource_id}/download/{self.filename}"
+        )
+        r = requests.get(url)
+        self._check_request_errors(r)
+
+        return r.content

pyngeso/resources.py

@@ -1,6 +1,6 @@
 from typing import Dict
 
-resource_ids: Dict[str, Dict[str, str]] = {
+api_resource_ids: Dict[str, Dict[str, str]] = {
     "historic-day-ahead-demand-forecast": {
         "id": "9847e7bb-986e-49be-8138-717b25933fbb",
         "url": "https://data.nationalgrideso.com/demand/1-day-ahead-demand-forecast/r/historic_day_ahead_demand_forecasts"
@@ -26,3 +26,12 @@
         "url": "https://data.nationalgrideso.com/demand/day-ahead-wind-forecast/r/day_ahead_wind_forecast"
     }
 }
+
+file_resource_ids: Dict[str, Dict[str, str]] = {
+    "historic-generation-mix": {
+        "dataset_id": "88313ae5-94e4-4ddc-a790-593554d8c6b9",
+        "resource_id": "f93d1835-75bc-43e5-84ad-12472b180a98",
+        "filename": "df_fuel_ckan.csv",
+        "url": "https://data.nationalgrideso.com/carbon-intensity1/historic-generation-mix"
+    }
+}

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pyngeso"
-version = "0.2.1"
+version = "0.2.2"
 description = "Simple python wrapper for the National Grid ESO Portal"
 authors = ["atsangarides <[email protected]>"]
 license = "MIT"

tests/test_pyngeso.py

@@ -1,5 +1,6 @@
 import logging
 import json
+import csv
 
 import pytest
 
@@ -70,3 +71,22 @@ def test_historic_day_ahead_wind_forecast():
     assert len(records) > 0
     unique_target_dates = set([record.get(date_col) for record in records])
     assert len(unique_target_dates) == 1
+
+
+@pytest.mark.vcr
+def test_historic_generation_mix():
+    client = NgEso("historic-generation-mix", "file")
+    r = client.download_file()
+    # test response type
+    assert isinstance(r, bytes)
+
+    # test bytes -> csv
+    decoded_content = r.decode('utf-8')
+    c = csv.reader(decoded_content.splitlines(), delimiter=',')
+    headers_row = next(c)
+    first_row = next(c)
+
+    assert "DATETIME" in headers_row
+    assert "2009-01-01 00:00:00" in first_row
+    assert len(headers_row) == len(first_row)
+