diff --git a/parsers/CA_SK.py b/parsers/CA_SK.py index fd51e9f711..dc461a8e34 100644 --- a/parsers/CA_SK.py +++ b/parsers/CA_SK.py @@ -1,142 +1,116 @@ from datetime import datetime, timedelta from logging import Logger, getLogger +from typing import Any +from zoneinfo import ZoneInfo -from pytz import timezone -from requests import Response, Session +from requests import Session +from electricitymap.contrib.config import ZoneKey +from electricitymap.contrib.lib.models.event_lists import ( + ProductionBreakdownList, + TotalConsumptionList, +) +from electricitymap.contrib.lib.models.events import ProductionMix from parsers.lib.exceptions import ParserException -TIMEZONE = timezone("America/Regina") - -# URLs for the different endpoints. +TIMEZONE = ZoneInfo("America/Regina") PRODUCTION_URL = ( - "https://www.saskpower.com/ignitionapi/PowerUseDashboard/GetPowerUseDashboardData" + "https://www.saskpower.com/ignitionapi/PowerUseDashboard" + "/GetPowerUseDashboardData" ) CONSUMPTION_URL = "https://www.saskpower.com/ignitionapi/Content/GetNetLoad" - PRODUCTION_MAPPING = { "Hydro": "hydro", "Wind": "wind", "Solar": "solar", "Natural Gas": "gas", "Coal": "coal", - "Other": "unknown", # This is internal consumption, losses, heat recovery facilities and small independent power producers. + # "Other" represents internal consumption, losses, heat recovery facilities + # and small independent power producers. + "Other": "unknown", } - -USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.142 Safari/537.36" - - -def validate_zone_key(zone_key: str) -> None: - if zone_key != "CA-SK": - raise ParserException( - "CA_SK.py", - f"CA_SK.py is not designed to parse zone_key: {zone_key}.", - zone_key, - ) +USER_AGENT = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/75.0.3770.142 Safari/537.36" +) -def validate_no_datetime(target_datetime: datetime | None, zone_key) -> None: +def _request( + session: Session | None, + target_datetime: datetime | None, + url: str, + zone_key: ZoneKey, +) -> dict | str: + # The source does not offer historical data, so bail out if it's requested. if target_datetime: + raise ParserException("CA_SK.py", "Unable to fetch historical data", zone_key) + # The zone key must be "CA-SK"; bail out otherwise. + if zone_key != "CA-SK": + raise ParserException("CA_SK.py", f"Cannot parse zone '{zone_key}'", zone_key) + session = session or Session() + # Mimic a user browser in the headers or the API will respond with a 403. + response = session.get(url, headers={"user-agent": USER_AGENT}) + if not response.ok: raise ParserException( "CA_SK.py", - "This parser is unable to fetch historical data.", + f"Request to {url} failed. Response Code: {response.status_code}\n" + f"Error:\n{response.text}", zone_key, ) + return response.json() def fetch_production( - zone_key: str = "CA-SK", + zone_key: ZoneKey = ZoneKey("CA-SK"), session: Session | None = None, target_datetime: datetime | None = None, logger: Logger = getLogger(__name__), -): - """This parser function will currently return the daily average of the day in question as hourly data. - This is because the API only returns daily data but the backend expects hourly values. - This is in order to facilitate the estimation of the hourly values from the daily average. - """ - # Validate that the zone key is equal to CA-SK. - validate_zone_key(zone_key) - # Validate that the target_datetime is None as this parser is unable to fetch historical data. - validate_no_datetime(target_datetime, zone_key) - - session = session or Session() - - # Set the headers to mimic a user browser as the API will return a 403 if not. - headers = {"user-agent": USER_AGENT} - response: Response = session.get(PRODUCTION_URL, headers=headers) - - if not response.ok: - raise ParserException( - "CA_SK.py", - f"Failed to fetch production data. Response Code: {response.status_code}\nError:\n{response.text}", - zone_key, - ) - - raw_data = response.json() +) -> list[dict[str, Any]]: + payload = _request(session, target_datetime, PRODUCTION_URL, zone_key) + if not isinstance(payload, dict): + raise ParserException("CA_SK.py", "Unexpected payload type", zone_key) # Date is in the format "Jan 01, 2020" - raw_date = raw_data["SupplyDataText"] - date = datetime.strptime(raw_date, "%b %d, %Y") - production_data = {} - - for value in raw_data["PowerCacheData"]["generationByType"]: - production_data[PRODUCTION_MAPPING[value["type"]]] = value[ - "totalGenerationForType" - ] - - data_list: list[dict] = [] - # Hack to return hourly data from daily data for the backend as it expects hourly data. - for hour in range(0, 24): - data_list.append( - { - "zoneKey": zone_key, - "datetime": date.replace(hour=hour, tzinfo=TIMEZONE), - "production": production_data, - "source": "saskpower.com", - } + date = datetime.strptime(payload["SupplyDataText"], "%b %d, %Y") + production_mix = ProductionMix() + for generation_by_type in payload["PowerCacheData"]["generationByType"]: + production_mix.add_value( + PRODUCTION_MAPPING[generation_by_type["type"]], + generation_by_type["totalGenerationForType"], ) - - return data_list + production_breakdown_list = ProductionBreakdownList(logger) + # Copy the daily average returned by the API into hourly values. This is a + # bit of a hack, but it's required because the back-end requires hourly + # datapoints while the API only provides daily averages. + for hour in range(24): + production_breakdown_list.append( + datetime=date.replace(hour=hour, tzinfo=TIMEZONE), + production=production_mix, + source="saskpower.com", + zoneKey=ZoneKey(zone_key), + ) + return production_breakdown_list.to_list() def fetch_consumption( - zone_key: str = "CA-SK", + zone_key: ZoneKey = ZoneKey("CA-SK"), session: Session | None = None, target_datetime: datetime | None = None, logger: Logger = getLogger(__name__), -): - # Validate that the zone key is equal to CA-SK. - validate_zone_key(zone_key) - # Validate that the target_datetime is None as this parser is unable to fetch historical data. - validate_no_datetime(target_datetime, zone_key) - - session = session or Session() - - # Set the headers to mimic a user browser as the API will return a 403 if not. - headers = {"user-agent": USER_AGENT} - - response: Response = session.get(CONSUMPTION_URL, headers=headers) - - if not response.ok: - raise ParserException( - "CA_SK.py", - f"Failed to fetch consumption data. Response Code: {response.status_code}\nError:\n{response.text}", - zone_key, - ) - - raw_data = response.json() - +) -> list[dict[str, Any]]: + payload = _request(session, target_datetime, CONSUMPTION_URL, zone_key) + if not isinstance(payload, str): + raise ParserException("CA_SK.py", "Unexpected payload type", zone_key) + # The source refreshes every 5 minutes, so we assume the current data is + # from 5 minutes before the most recent multiple of 5 minutes. now = datetime.now(TIMEZONE) - - # Data is updated every 5 minutes so we assume the data is from a multiple of 5 minutes and has a 5 minute delay from that multiple. assumed_datetime = now.replace(second=0, microsecond=0) - timedelta( - minutes=(now.minute % 5) + 5 + minutes=(5 + now.minute % 5) ) - - return [ - { - "zoneKey": zone_key, - "datetime": assumed_datetime, - "consumption": int(raw_data), - "source": "saskpower.com", - } - ] + total_consumption = TotalConsumptionList(logger) + total_consumption.append( + consumption=float(payload), + datetime=assumed_datetime, + source="saskpower.com", + zoneKey=zone_key, + ) + return total_consumption.to_list()