run_daily_weather.py

import sys
import psycopg2.extras
import psycopg2
from dotenv import load_dotenv
import logging
import os
import requests
import datetime
from weather_utils import extract

# This script fetches hourly weather data from the BrightSky API, aggregates it to daily weather data and stores it in the database

# Set up logging
logging.basicConfig()
logging.root.setLevel(logging.INFO)

# Load the environmental variables
load_dotenv()

# Check if all required environmental variables are accessible
for env_var in [
    "PG_SERVER",
    "PG_PORT",
    "PG_USER",
    "PG_PASS",
    "PG_DB",
    "WEATHER_HARVEST_LAT",
    "WEATHER_HARVEST_LNG",
]:
    if env_var not in os.environ:
        logging.error("❌Environmental Variable {} does not exist".format(env_var))
        sys.exit(1)

PG_SERVER = os.getenv("PG_SERVER")
PG_PORT = os.getenv("PG_PORT")
PG_USER = os.getenv("PG_USER")
PG_PASS = os.getenv("PG_PASS")
PG_DB = os.getenv("PG_DB")
WEATHER_HARVEST_LAT = os.getenv("WEATHER_HARVEST_LAT")
WEATHER_HARVEST_LNG = os.getenv("WEATHER_HARVEST_LNG")

# Establish database connection
try:
    database_connection = psycopg2.connect(
        dbname=PG_DB, user=PG_USER, password=PG_PASS, host=PG_SERVER, port=PG_PORT
    )
    logging.info("🗄 Database connection established")
except:
    logging.error("❌Could not establish database connection")
    database_connection = None
    sys.exit(1)

today = datetime.date.today()

# Calculate the date some years ago
x_years_ago = today - datetime.timedelta(days=3 * 365 + 31)

# Generate a list of all dates between x years ago and today
date_list = [
    x_years_ago + datetime.timedelta(days=x)
    for x in range((today - x_years_ago).days + 1)
]

print(f"📅 Fetching weather data for {len(date_list)} days...")
weather_days_in_db = []
with database_connection.cursor() as cur:
    cur.execute("SELECT measure_day, day_finished FROM daily_weather_data;")
    weather_days_in_db = cur.fetchall()

outdated_weather_data = [
    data_point_in_db
    for data_point_in_db in weather_days_in_db
    if data_point_in_db[0].date() < x_years_ago
]
logging.info(
    f"🌦 Deleting {len(outdated_weather_data)} outdated weather data entries..."
)
for data_point in outdated_weather_data:
    with database_connection.cursor() as cur:
        cur.execute(
            "DELETE FROM daily_weather_data WHERE measure_day = %s", [data_point[0]]
        )
database_connection.commit()


for date in date_list:
    today = datetime.date.today()

    existing_weather_in_db_for_this_day = [
        data_point_in_db
        for data_point_in_db in weather_days_in_db
        if data_point_in_db[0].date() == date
    ]

    if existing_weather_in_db_for_this_day != []:
        logging.info(f"🌦 Weather data for {date} already exists in the database...")

        unfinished_weather_data = [
            data_point_in_db
            for data_point_in_db in existing_weather_in_db_for_this_day
            if data_point_in_db[1] == False
        ]

        if unfinished_weather_data != []:
            logging.info(
                f"🌦 Weather data for {date} was not finished in last run, updating now..."
            )
            with database_connection.cursor() as cur:
                logging.info(f"🌦 Deleting old weather data for {date}...")
                cur.execute(
                    "DELETE FROM daily_weather_data WHERE measure_day = %s", [date]
                )
            database_connection.commit()
        else:
            continue

    # Using BrightSky API to fetch weather data https://brightsky.dev/docs/#/
    # Hint: No API key is required
    url = "https://api.brightsky.dev/weather"
    params = {
        "date": date,
        "lat": WEATHER_HARVEST_LAT,
        "lon": WEATHER_HARVEST_LNG,
    }
    headers = {"Accept": "application/json"}
    response = requests.get(url, params=params, headers=headers)
    weather_raw = response.json()
    weather = weather_raw["weather"]

    # Aggregate hourly weather data to daily weather data
    sum_precipitation_mm_per_sqm = sum(extract(weather, "precipitation"))
    avg_temperature_celsius = sum(extract(weather, "temperature")) / len(weather)
    avg_pressure_msl = sum(extract(weather, "pressure_msl")) / len(weather)
    sum_sunshine_minutes = sum(extract(weather, "sunshine"))
    avg_wind_direction_deg = sum(extract(weather, "wind_direction")) / len(weather)
    avg_wind_speed_kmh = sum(extract(weather, "wind_speed")) / len(weather)
    avg_cloud_cover_percentage = sum(extract(weather, "cloud_cover")) / len(weather)
    avg_dew_point_celcius = sum(extract(weather, "dew_point")) / len(weather)
    avg_relative_humidity_percentage = sum(extract(weather, "relative_humidity")) / len(
        weather
    )
    avg_visibility_m = sum(extract(weather, "visibility")) / len(weather)
    avg_wind_gust_direction_deg = sum(extract(weather, "wind_gust_direction")) / len(
        weather
    )
    avg_wind_gust_speed_kmh = sum(extract(weather, "wind_gust_speed")) / len(weather)

    source_dwd_station_ids = extract(weather_raw["sources"], "dwd_station_id")

    day_finished = date < today

    logging.info(f"🌦 Weather data for {date} fetched via BrightySky API...")

    with database_connection.cursor() as cur:
        cur.execute(
            "INSERT INTO daily_weather_data (measure_day, day_finished, sum_precipitation_mm_per_sqm, avg_temperature_celsius, avg_pressure_msl, sum_sunshine_minutes, avg_wind_direction_deg, avg_wind_speed_kmh, avg_cloud_cover_percentage, avg_dew_point_celcius, avg_relative_humidity_percentage, avg_visibility_m, avg_wind_gust_direction_deg, avg_wind_gust_speed_kmh, source_dwd_station_ids) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
            [
                date,
                day_finished,
                sum_precipitation_mm_per_sqm,
                avg_temperature_celsius,
                avg_pressure_msl,
                sum_sunshine_minutes,
                avg_wind_direction_deg,
                avg_wind_speed_kmh,
                avg_cloud_cover_percentage,
                avg_dew_point_celcius,
                avg_relative_humidity_percentage,
                avg_visibility_m,
                avg_wind_gust_direction_deg,
                avg_wind_gust_speed_kmh,
                source_dwd_station_ids,
            ],
        )

database_connection.commit()