agenda/agenda/data.py

"""Agenda data."""

import asyncio
import configparser
import operator
import os
import typing
from datetime import date, datetime, time, timedelta

import dateutil.tz
import holidays
import lxml
import pytz
import yaml
from dateutil.relativedelta import FR, relativedelta

from . import (
    accommodation,
    birthday,
    calendar,
    conference,
    economist,
    fx,
    gwr,
    markets,
    stock_market,
    subscription,
    sun,
    thespacedevs,
    travel,
    uk_holiday,
    uk_time,
    waste_schedule,
)
from .types import Event

here = dateutil.tz.tzlocal()

next_us_presidential_election = date(2024, 11, 5)

# deadline to file tax return
# credit card expiry dates
# morzine ski lifts
# chalet availablity calendar

# starlink visible


def timezone_transition(
    start_dt: datetime, end_dt: datetime, key: str, tz_name: str
) -> list[Event]:
    """Clocks changes."""
    tz = pytz.timezone(tz_name)
    return [
        Event(name=key, date=pytz.utc.localize(t).astimezone(tz))
        for t in tz._utc_transition_times  # type: ignore
        if start_dt <= t <= end_dt
    ]


def uk_financial_year_end(start_date: date, end_date: date) -> list[Event]:
    """Generate a list of UK financial year end dates (April 5th) between two dates."""
    # Initialize an empty list to store the financial year ends
    financial_year_ends: list[date] = []

    # Start from the year of the start date
    year = start_date.year

    # Calculate the first possible financial year end date
    financial_year_end = date(year, 4, 5)

    # Loop over the years until we reach the year of the end date
    while financial_year_end < end_date:
        # If the financial year end date is on or after the start date,
        # add it to the list
        if financial_year_end >= start_date:
            financial_year_ends.append(financial_year_end)

        year += 1  # Move to the next year
        financial_year_end = date(year, 4, 5)

    return [Event(name="uk_financial_year_end", date=d) for d in financial_year_ends]


def get_us_holidays(start_date: date, end_date: date) -> list[Event]:
    """Date and name of next US holiday."""
    found: list[Event] = []
    for year in range(start_date.year, end_date.year + 1):
        hols = holidays.UnitedStates(years=year)
        found += [
            Event(name="us_holiday", date=hol_date, title=title.replace("'", "’"))
            for hol_date, title in hols.items()
            if start_date < hol_date < end_date
        ]

    extra = []
    for e in found:
        if e.title != "Thanksgiving":
            continue
        extra += [
            Event(
                name="us_holiday", date=e.date + timedelta(days=1), title="Black Friday"
            ),
            Event(
                name="us_holiday", date=e.date + timedelta(days=4), title="Cyber Monday"
            ),
        ]

    return found + extra


def critical_mass(start_date: date, end_date: date) -> list[Event]:
    """Future dates for Critical Mass."""
    events: list[Event] = []
    current_date = start_date
    t = time(18, 0)

    while current_date < end_date:
        # Calculate the last Friday of the current month
        last_friday = current_date + relativedelta(day=31, weekday=FR(-1))

        events.append(Event(name="critical_mass", date=uk_time(last_friday, t)))

        # Move to the next month
        current_date += relativedelta(months=1)

    return events


async def waste_collection_events(data_dir: str) -> list[Event]:
    """Waste colllection events."""
    postcode = "BS48 3HG"
    uprn = "24071046"

    html = await waste_schedule.get_html(data_dir, postcode, uprn)
    root = lxml.html.fromstring(html)
    events = waste_schedule.parse(root)
    return events


async def bristol_waste_collection_events(
    data_dir: str, start_date: date
) -> list[Event]:
    """Waste colllection events."""
    uprn = "358335"

    return await waste_schedule.get_bristol_gov_uk(start_date, data_dir, uprn)


def combine_holidays(events: list[Event]) -> list[Event]:
    """Combine UK and US holidays with the same date and title."""
    combined: dict[tuple[date, str], Event] = {}

    for e in events:
        assert isinstance(e.title, str) and isinstance(e.date, date)
        event_key = (e.date, e.title)
        combined[event_key] = (
            Event(name="bank_holiday", date=e.date, title=e.title + " (UK & US)")
            if event_key in combined
            else e
        )

    return list(combined.values())


def read_events_yaml(data_dir: str) -> list[Event]:
    """Read eventes from YAML file."""
    return [
        Event(
            name=item["name"],
            date=item["end_date" if item["name"] == "travel_insurance" else "date"],
            title=item.get("title"),
            url=item.get("url"),
        )
        for item in yaml.safe_load(open(os.path.join(data_dir, "events.yaml")))
    ]


async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
    """Get data to display on agenda dashboard."""
    config_filename = os.path.join(os.path.dirname(__file__), "..", "config")

    assert os.path.exists(config_filename)

    config = configparser.ConfigParser()
    config.read(config_filename)

    data_dir = config.get("data", "dir")

    rocket_dir = os.path.join(data_dir, "thespacedevs")
    today = now.date()
    two_weeks_ago = today - timedelta(weeks=2)
    last_week = today - timedelta(weeks=1)
    last_year = today - timedelta(days=365)
    next_year = today + timedelta(days=365)

    minus_365 = now - timedelta(days=365)
    plus_365 = now + timedelta(days=365)

    (
        gbpusd,
        gwr_advance_tickets,
        bank_holiday,
        rockets,
        backwell_bins,
        bristol_bins,
    ) = await asyncio.gather(
        fx.get_gbpusd(config),
        gwr.advance_ticket_date(data_dir),
        uk_holiday.bank_holiday_list(last_year, next_year, data_dir),
        thespacedevs.get_launches(rocket_dir, limit=40),
        waste_collection_events(data_dir),
        bristol_waste_collection_events(data_dir, today),
    )

    reply = {
        "now": now,
        "gbpusd": gbpusd,
        "economist": economist.publication_dates(last_year, next_year),
        "next_us_presidential_election": next_us_presidential_election,
        "stock_markets": stock_market.open_and_close(),
        "uk_clock_change": timezone_transition(
            minus_365, plus_365, "uk_clock_change", "Europe/London"
        ),
        "us_clock_change": timezone_transition(
            minus_365, plus_365, "us_clock_change", "America/New_York"
        ),
        "mothers_day": uk_holiday.get_mothers_day(today),
        "fathers_day": uk_holiday.get_fathers_day(today),
        "uk_financial_year_end": uk_financial_year_end(last_year, next_year),
        "gwr_advance_tickets": gwr_advance_tickets,
        "critical_mass": critical_mass(last_year, next_year),
        "market": (
            markets.windmill_hill(last_year, next_year)
            + markets.tobacco_factory(last_year, next_year)
            + markets.nailsea_farmers(last_year, next_year)
        ),
        "rockets": rockets,
    }

    skip = {"now", "gbpusd", "rockets", "stock_markets"}
    events: list[Event] = []
    for key, value in reply.items():
        if key in skip:
            continue
        if isinstance(value, list):
            events += value
        else:
            assert isinstance(value, date)
            event = Event(name=key, date=value)
            events.append(event)

    observer = sun.bristol()
    reply["sunrise"] = sun.sunrise(observer)
    reply["sunset"] = sun.sunset(observer)

    events += combine_holidays(bank_holiday + get_us_holidays(last_year, next_year))

    my_data = config["data"]["personal-data"]
    events += birthday.get_birthdays(last_year, os.path.join(my_data, "entities.yaml"))
    events += accommodation.get_events(os.path.join(my_data, "accommodation.yaml"))
    events += travel.all_events(config["data"]["personal-data"])
    events += conference.get_list(os.path.join(my_data, "conferences.yaml"))
    events += backwell_bins + bristol_bins
    events += read_events_yaml(my_data)

    events += subscription.get_events(os.path.join(my_data, "subscriptions.yaml"))

    events.sort(key=operator.attrgetter("as_datetime"))

    reply["events"] = events
    reply["last_week"] = last_week
    reply["two_weeks_ago"] = two_weeks_ago

    reply["fullcalendar_events"] = calendar.build_events(events)

    return reply