Eskom API

eskom.py

#!/usr/bin/env python3

import urllib.parse
from enum import Enum
from typing import List

import requests

BASE_URL = "https://loadshedding.eskom.co.za/LoadShedding/"
USER_AGENT = "ZA_LOADSHEDDING/1.0 (github.com/rany2)"

session = requests.session()
session.headers.update({"User-Agent": USER_AGENT})


class LoadsheddingStatus(Enum):
    DATA_UNAVAILABLE = 99
    NOT_LOADSHEDDING = 1
    STAGE_1 = 2
    STAGE_2 = 3
    STAGE_3 = 4
    STAGE_4 = 5
    STAGE_5 = 6
    STAGE_6 = 7
    STAGE_7 = 8
    STAGE_8 = 9

    def jsval(self):
        if self.value == 99:
            return self.value
        return self.value - 1


def get_status() -> LoadsheddingStatus:
    """Get the current loadshedding status"""
    response = session.get(BASE_URL + "GetStatus")
    response.raise_for_status()
    return LoadsheddingStatus(int(response.text))


class Province(Enum):
    EASTERN_CAPE = 1
    FREE_STATE = 2
    GAUTENG = 3
    KWAZULU_NATAL = 4
    LIMPOPO = 5
    MPUMALANGA = 6
    NORTH_WEST = 7
    NORTHERN_CAPE = 8
    WESTERN_CAPE = 9

    def __str__(self):
        if self.value == 4:
            return "KwaZulu-Natal"
        return self.name.replace("_", " ").title()

    def __repr__(self):
        return self.__str__()

    def __eq__(self, other):
        if isinstance(other, Province):
            return self.value == other.value
        if isinstance(other, int):
            return self.value == other
        return False

    @classmethod
    def from_name(cls, name: str):
        if not isinstance(name, str):
            raise TypeError("name must be an str")
        for i in cls:
            if str(i).lower() == name.lower():
                return i
        raise ValueError("Invalid province name")

    @classmethod
    def from_value(cls, value: int):
        if not isinstance(value, int):
            raise TypeError("value must be an int")
        for i in cls:
            if i.value == value:
                return i
        raise ValueError("Invalid province value")


class BasePlace:
    def __init__(
        self,
        name: str,
        value: int,
        province: Province,
    ):
        if not isinstance(value, int):
            raise TypeError("value must be an int")
        if not isinstance(province, Province):
            raise TypeError("province must be a Province")
        if not isinstance(name, str):
            raise TypeError("name must be a str")
        self.name = name
        self.value = value
        self.province = province

    def __str__(self):
        if isinstance(self.name, str):
            return self.name + " (" + str(self.value) + ")"
        return str(self.value)

    def __repr__(self):
        return self.__str__()


class Municipality(BasePlace):
    pass


class Suburb(BasePlace):
    def __init__(
        self,
        name: str,
        value: int,
        province: Province = None,
        *,
        municipality_name: str,
        total: int,
    ):
        super().__init__(name, value, province=province)
        if not isinstance(municipality_name, str):
            raise TypeError("municipality_name must be an str")
        if not isinstance(total, int):
            raise TypeError("total must be an int")
        self.municipality_name = municipality_name
        self.total = total


def get_municipalities(province_id) -> List[Municipality]:
    """Get a list of municipalities for a given province"""
    if isinstance(province_id, Province):
        province_id = province_id.value
    if not isinstance(province_id, int):
        raise TypeError("province_id must be an int or a Municipalities enum")
    if not 1 <= province_id <= 9:
        raise ValueError("province_id must be between 1 and 9")
    response = session.get(BASE_URL + "GetMunicipalities/", params={"Id": province_id})
    response.raise_for_status()
    l = []
    for i in response.json():
        l.append(
            Municipality(
                i["Text"],
                int(i["Value"]),
                province=Province.from_value(province_id),
            )
        )
    return l


def get_suburbs(
    municipality: Municipality,
    *,
    page_size: int = 100,
    page_num: int = 1,
    searchTerm: str = "",
) -> List[Suburb]:
    if not isinstance(municipality, Municipality):
        raise TypeError("municipality_id must be a Municipality")
    if not isinstance(page_size, int):
        raise TypeError("page_size must be an int")
    if not isinstance(page_num, int):
        raise TypeError("page_num must be an int")
    if not isinstance(searchTerm, str):
        raise TypeError("searchTerm must be a str")

    response = session.get(
        BASE_URL + "GetSuburbs",
        params={
            "pageSize": page_size,
            "pageNum": page_num,
            "searchTerm": searchTerm,
            "id": municipality.value,
        },
    )
    response.raise_for_status()
    l = []
    for i in response.json():
        l.append(
            Suburb(
                i["Text"],
                int(i["Value"]),
                municipality_name=str(municipality),
                province=municipality.province,
                total=i["Total"],
            )
        )
    return l


def find_suburbs(suburb_name: str, *, max_results: int = 300) -> List[Suburb]:
    """Find suburbs by name"""
    if not isinstance(suburb_name, str):
        raise TypeError("suburb_name must be a str")
    if len(suburb_name) < 3:
        raise ValueError("suburb_name must be at least 3 characters long")
    if not isinstance(max_results, int):
        raise TypeError("max_results must be int")
    if max_results <= 0:
        raise ValueError("max_results must be greater than 0")
    response = session.get(
        BASE_URL + "FindSuburbs",
        params={"searchText": suburb_name, "maxResults": max_results},
    )
    response.raise_for_status()
    l = []
    for i in response.json():
        l.append(
            Suburb(
                i["Name"],
                int(i["Id"]),
                municipality_name=i["MunicipalityName"],
                province=Province.from_name(i["ProvinceName"]),
                total=int(i["Total"]),
            )
        )
    return l


def get_schedule_m_url(
    suburb: Suburb, cur_status: LoadsheddingStatus = get_status()
) -> str:
    if not isinstance(suburb, Suburb):
        raise TypeError("suburb must be a Suburb")
    if not isinstance(cur_status, LoadsheddingStatus):
        raise TypeError("cur_status must be a LoadsheddingStatus")
    return (
        f"{BASE_URL}GetScheduleM/{suburb.value}/{cur_status.jsval()}/"
        + f"{urllib.parse.quote(str(suburb.province).encode('utf-8'))}/{suburb.total}"
    )


def get_schedule_pdf_url(suburb: Suburb) -> str:
    if not isinstance(suburb, Suburb):
        raise TypeError("suburb must be a Suburb")
    return f"{BASE_URL}DownloadFile?ID={suburb.value}&Name={urllib.parse.quote(str(suburb.name).encode('utf-8'))}"


def get_schedule_areainfo_url(suburb: Suburb) -> str:
    if not isinstance(suburb, Suburb):
        raise TypeError("suburb must be a Suburb")
    return f"{BASE_URL}GetScheduleAreaInfo?Id={suburb.value}"


if __name__ == "__main__":
    # mun = get_municipalities(Province.NORTHERN_CAPE)
    # sub = get_suburbs(mun[0])
    # print(get_schedule_m_url(sub[0]))

    sub = find_suburbs("Gaula")
    print(get_schedule_areainfo_url(sub[0]))
    print(get_schedule_m_url(sub[0]))
    print(get_schedule_pdf_url(sub[0]))