Store outages in Outages.yaml

outages.yaml

@@ -0,0 +1,5 @@
+- name: MGHPCC Shutdown 2024
+  information: https://nerc.mghpcc.org/event/mghpcc-annual-power-shutdown-2024/
+  timeframes:
+    - start: "2024-05-22T08:00:00-0400"
+      end: "2024-05-28T23:00:00-0400"

src/nerc_rates/models_outages.py

@@ -0,0 +1,33 @@
+from typing import Annotated
+from datetime import datetime, timezone
+
+import pydantic
+
+from .models import Base
+
+
+class OutageTimeFrames(Base):
+    start: datetime
+    end: datetime
+
+
+class OutageItemDict(Base):
+    name: str
+    information: str
+    timeframes: list[OutageTimeFrames]
+
+
+class Outages(pydantic.RootModel):
+    root: list[OutageItemDict]
+
+    def get_outages_during(self, start, end):
+        start = datetime.strptime(start, "%Y-%m-%d").astimezone(timezone.utc)
+        end = datetime.strptime(end, "%Y-%m-%d").astimezone(timezone.utc)
+
+        outages = []
+        for outage in self.root:
+            for o in outage.timeframes:
+                if start < o.start < end or start < o.end < end:
+                    outages.append((max(start, o.start), min(end, o.end)))
+
+        return outages

src/nerc_rates/outages.py

@@ -0,0 +1,28 @@
+import requests
+import yaml
+
+from .models_outages import Outages
+
+DEFAULT_OUTAGES_FILE = "outages.yaml"
+DEFAULT_OUTAGES_URL = (
+    "https://raw.githubusercontent.com/CCI-MOC/nerc-outages/main/outages.yaml"
+)
+
+
+def load_from_url(url: str | None = None) -> Outages:
+    if url is None:
+        url = DEFAULT_OUTAGES_URL
+
+    r = requests.get(url, allow_redirects=True)
+    r.raise_for_status()
+    config = yaml.safe_load(r.content.decode("utf-8"))
+    return Outages.model_validate(config)
+
+
+def load_from_file(path: str | None = None) -> Outages:
+    if path is None:
+        path = DEFAULT_OUTAGES_FILE
+
+    with open(path, "r") as f:
+        config = yaml.safe_load(f)
+    return Outages.model_validate(config)