import io
from zipfile import ZipFile
import pandas as pd
import requests
import tqdm
import gridstatus
from gridstatus.base import ISOBase, Markets, NotSupported
from gridstatus.caiso import CAISO
from gridstatus.ercot import Ercot
from gridstatus.isone import ISONE
from gridstatus.miso import MISO
from gridstatus.nyiso import NYISO
from gridstatus.pjm import PJM
from gridstatus.spp import SPP
[docs]all_isos = [MISO, CAISO, PJM, Ercot, SPP, NYISO, ISONE]
[docs]def list_isos():
"""List available ISOs"""
isos = [[i.name, i.iso_id, i.__name__] for i in all_isos]
return pd.DataFrame(isos, columns=["Name", "Id", "Class"])
[docs]def get_iso(iso_id):
"""Get an ISO by its id"""
for i in all_isos:
if i.iso_id == iso_id:
return i
raise KeyError
[docs]def make_availability_df():
methods = [
"get_status",
"get_fuel_mix",
"get_load",
"get_supply",
"get_load_forecast",
"get_storage",
]
availability = {}
for i in tqdm.tqdm(gridstatus.all_isos):
availability[i.__name__] = {}
for method in methods:
availability[i.__name__][method] = {}
for date in ["latest", "today", "historical"]:
test = date
if date == "historical":
test = pd.Timestamp.now(
tz=i.default_timezone,
).date() - pd.Timedelta(days=3)
if method == "get_load_forecast" and date == "latest":
is_defined = "❌" # red x
else:
try:
getattr(i(), method)(test)
is_defined = "✅" # green checkmark
except NotSupported:
is_defined = "❌" # red x
except NotImplementedError:
is_defined = "❌" # red x
availability[i.__name__][method][date] = is_defined
availability_dfs = {}
for i in all_isos:
availability_dfs[i.__name__] = pd.DataFrame(availability[i.__name__])
return availability_dfs
[docs]def make_availability_table():
dfs = make_availability_df()
markdown = ""
for method, df in sorted(dfs.items()):
markdown += "## " + method + "\n"
# df.index = ["`" + v + "`" for v in df.index.values]
markdown += df.to_markdown() + "\n"
return markdown
def _handle_date(date, tz=None):
if not isinstance(date, pd.Timestamp):
date = pd.to_datetime(date)
if tz and date.tzinfo is None:
date = date.tz_localize(tz)
return date
[docs]def make_lmp_availability():
lmp_availability = {}
for i in all_isos:
lmp_availability[i.name] = i.markets
return lmp_availability
[docs]def make_lmp_availability_table():
a = make_lmp_availability()
for iso in a:
a[iso] = ["`" + v.value + "`" for v in a[iso]]
a[iso] = ", ".join(a[iso])
s = pd.Series(a, name="Markets")
return s.to_markdown()
[docs]def filter_lmp_locations(data, locations: list):
if locations == "ALL" or locations is None:
return data
return data[data["Location"].isin(locations)]
[docs]def get_zip_file(url):
# todo add retry logic
# todo does this need to be a with statement?
r = requests.get(url)
z = ZipFile(io.BytesIO(r.content))
return z.open(z.namelist()[0])
[docs]def is_today(date, tz=None):
return _handle_date(date, tz=tz).date() == pd.Timestamp.now(tz=tz).date()
