Pirate Weather → iCal & JSON Script

weather2calendar.py

#!/usr/bin/env python3
import datetime
import json
import os
import sys
import gzip

import ics
import requests
import pytz

API_TOKEN = os.getenv("API_TOKEN")
LAT, LON = os.getenv("LAT"), os.getenv("LON")
TZ_NAME = os.getenv("TZ")


def desc(key, val, tz):
    if not key.lower().endswith("time"):
        return val
    dt = datetime.datetime.fromtimestamp(val, tz)
    return str(dt.time())


def main(dirname):
    tz = pytz.timezone(TZ_NAME)

    url = f"https://api.pirateweather.net/forecast/{API_TOKEN}/{LAT},{LON}?exclude=hourly,minutely"
    print(url, file=sys.stderr)

    got = requests.get(url)
    print(got, file=sys.stderr)

    now = datetime.datetime.now(pytz.timezone(TZ_NAME))
    today = str(now.date())
    with open(os.path.join(dirname, f"{today}.jsonl.gz"), "ab") as file:
        line = json.dumps(got.json()["currently"]) + "\n"
        file.write(gzip.compress(line.encode("utf8")))
        print("wrote", file.name, file=sys.stderr)

    cal = ics.Calendar()
    days_since_rain = 0

    for day in got.json()["daily"]["data"]:
        dt = datetime.datetime.fromtimestamp(day["time"], tz)
        print(
            dt,
            day["precipIntensityMax"],
            day["temperatureHigh"],
            day["temperatureLow"],
            day["apparentTemperatureHigh"],
            day["apparentTemperatureLow"],
            file=sys.stderr,
        )
        sst = datetime.datetime.fromtimestamp(day["sunsetTime"], tz)
        sunset_event = ics.Event(name="Sunset", begin=str(sst), end=str(sst))
        cal.events.add(sunset_event)
        is_rain_expected = day["precipProbability"] > 0.35
        if is_rain_expected:
            days_since_rain = 0
        else:
            days_since_rain += 1
        weather_labels = []
        if is_rain_expected:
            weather_labels.append("Rain")
        if days_since_rain in (2, 3):
            weather_labels.append(f"{days_since_rain}d dry")
        if day["temperatureLow"] < 50:
            weather_labels.append(
                f"{day['temperatureLow']:.0f}°/{day['temperatureHigh']:.0f}°"
            )
        if not weather_labels:
            continue
        weather_event = ics.Event(
            name=", ".join(weather_labels),
            begin=str(dt),
            description="\n".join(f"{k}: {desc(k, v, tz)}" for k, v in day.items()),
        )
        weather_event.make_all_day()
        cal.events.add(weather_event)

    with open(os.path.join(dirname, "daily.ics"), "w") as file:
        file.writelines(cal.serialize_iter())
        print("wrote", file.name, file=sys.stderr)


if __name__ == "__main__":
    dirname = sys.argv[1]
    exit(main(dirname))