Add support for RRULE for YAML events
Remove code for events that can be represented via RRULE
This commit is contained in:
parent
e52dfb7fef
commit
1af5d7856f
|
@ -10,3 +10,8 @@ uk_tz = pytz.timezone("Europe/London")
|
||||||
def uk_time(d: date, t: time) -> datetime:
|
def uk_time(d: date, t: time) -> datetime:
|
||||||
"""Combine time and date for UK timezone."""
|
"""Combine time and date for UK timezone."""
|
||||||
return uk_tz.localize(datetime.combine(d, t))
|
return uk_tz.localize(datetime.combine(d, t))
|
||||||
|
|
||||||
|
|
||||||
|
def uk_midnight(d: date) -> datetime:
|
||||||
|
"""Midnight UK time."""
|
||||||
|
return uk_time(d, datetime.min.time())
|
||||||
|
|
|
@ -5,14 +5,14 @@ import configparser
|
||||||
import operator
|
import operator
|
||||||
import os
|
import os
|
||||||
import typing
|
import typing
|
||||||
from datetime import date, datetime, time, timedelta
|
from datetime import date, datetime, timedelta
|
||||||
|
|
||||||
|
import dateutil.rrule
|
||||||
import dateutil.tz
|
import dateutil.tz
|
||||||
import holidays
|
import holidays # type: ignore
|
||||||
import lxml
|
import lxml
|
||||||
import pytz
|
import pytz
|
||||||
import yaml
|
import yaml
|
||||||
from dateutil.relativedelta import FR, relativedelta
|
|
||||||
|
|
||||||
from . import (
|
from . import (
|
||||||
accommodation,
|
accommodation,
|
||||||
|
@ -29,7 +29,7 @@ from . import (
|
||||||
thespacedevs,
|
thespacedevs,
|
||||||
travel,
|
travel,
|
||||||
uk_holiday,
|
uk_holiday,
|
||||||
uk_time,
|
uk_midnight,
|
||||||
waste_schedule,
|
waste_schedule,
|
||||||
)
|
)
|
||||||
from .types import Event
|
from .types import Event
|
||||||
|
@ -47,41 +47,17 @@ next_us_presidential_election = date(2024, 11, 5)
|
||||||
|
|
||||||
|
|
||||||
def timezone_transition(
|
def timezone_transition(
|
||||||
start_dt: datetime, end_dt: datetime, key: str, tz_name: str
|
start: datetime, end: datetime, key: str, tz_name: str
|
||||||
) -> list[Event]:
|
) -> list[Event]:
|
||||||
"""Clocks changes."""
|
"""Clocks changes."""
|
||||||
tz = pytz.timezone(tz_name)
|
tz = pytz.timezone(tz_name)
|
||||||
return [
|
return [
|
||||||
Event(name=key, date=pytz.utc.localize(t).astimezone(tz))
|
Event(name=key, date=pytz.utc.localize(t).astimezone(tz))
|
||||||
for t in tz._utc_transition_times # type: ignore
|
for t in tz._utc_transition_times # type: ignore
|
||||||
if start_dt <= t <= end_dt
|
if start <= t <= end
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
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]:
|
def get_us_holidays(start_date: date, end_date: date) -> list[Event]:
|
||||||
"""Date and name of next US holiday."""
|
"""Date and name of next US holiday."""
|
||||||
found: list[Event] = []
|
found: list[Event] = []
|
||||||
|
@ -109,22 +85,18 @@ def get_us_holidays(start_date: date, end_date: date) -> list[Event]:
|
||||||
return found + extra
|
return found + extra
|
||||||
|
|
||||||
|
|
||||||
def critical_mass(start_date: date, end_date: date) -> list[Event]:
|
def dates_from_rrule(
|
||||||
"""Future dates for Critical Mass."""
|
rrule: str, start: date, end: date
|
||||||
events: list[Event] = []
|
) -> typing.Sequence[datetime | date]:
|
||||||
current_date = start_date
|
"""Generate events from an RRULE between start_date and end_date."""
|
||||||
t = time(18, 0)
|
all_day = not any(param in rrule for param in ["BYHOUR", "BYMINUTE", "BYSECOND"])
|
||||||
|
|
||||||
while current_date < end_date:
|
return [
|
||||||
# Calculate the last Friday of the current month
|
i.date() if all_day else i
|
||||||
last_friday = current_date + relativedelta(day=31, weekday=FR(-1))
|
for i in dateutil.rrule.rrulestr(rrule, dtstart=uk_midnight(start)).between(
|
||||||
|
uk_midnight(start), uk_midnight(end)
|
||||||
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]:
|
async def waste_collection_events(data_dir: str) -> list[Event]:
|
||||||
|
@ -163,17 +135,24 @@ def combine_holidays(events: list[Event]) -> list[Event]:
|
||||||
return list(combined.values())
|
return list(combined.values())
|
||||||
|
|
||||||
|
|
||||||
def read_events_yaml(data_dir: str) -> list[Event]:
|
def read_events_yaml(data_dir: str, start: date, end: date) -> list[Event]:
|
||||||
"""Read eventes from YAML file."""
|
"""Read eventes from YAML file."""
|
||||||
return [
|
events: list[Event] = []
|
||||||
Event(
|
for item in yaml.safe_load(open(os.path.join(data_dir, "events.yaml"))):
|
||||||
|
dates = (
|
||||||
|
dates_from_rrule(item["rrule"], start, end)
|
||||||
|
if "rrule" in item
|
||||||
|
else [item["end_date" if item["name"] == "travel_insurance" else "date"]]
|
||||||
|
)
|
||||||
|
for dt in dates:
|
||||||
|
e = Event(
|
||||||
name=item["name"],
|
name=item["name"],
|
||||||
date=item["end_date" if item["name"] == "travel_insurance" else "date"],
|
date=dt,
|
||||||
title=item.get("title"),
|
title=item.get("title"),
|
||||||
url=item.get("url"),
|
url=item.get("url"),
|
||||||
)
|
)
|
||||||
for item in yaml.safe_load(open(os.path.join(data_dir, "events.yaml")))
|
events.append(e)
|
||||||
]
|
return events
|
||||||
|
|
||||||
|
|
||||||
async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
|
async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
|
||||||
|
@ -226,10 +205,8 @@ async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
|
||||||
minus_365, plus_365, "us_clock_change", "America/New_York"
|
minus_365, plus_365, "us_clock_change", "America/New_York"
|
||||||
),
|
),
|
||||||
"mothers_day": uk_holiday.get_mothers_day(today),
|
"mothers_day": uk_holiday.get_mothers_day(today),
|
||||||
"fathers_day": uk_holiday.get_fathers_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,
|
"gwr_advance_tickets": gwr_advance_tickets,
|
||||||
"critical_mass": critical_mass(last_year, next_year),
|
|
||||||
"market": (
|
"market": (
|
||||||
markets.windmill_hill(last_year, next_year)
|
markets.windmill_hill(last_year, next_year)
|
||||||
+ markets.tobacco_factory(last_year, next_year)
|
+ markets.tobacco_factory(last_year, next_year)
|
||||||
|
@ -250,24 +227,21 @@ async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
|
||||||
event = Event(name=key, date=value)
|
event = Event(name=key, date=value)
|
||||||
events.append(event)
|
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"]
|
my_data = config["data"]["personal-data"]
|
||||||
|
events += combine_holidays(bank_holiday + get_us_holidays(last_year, next_year))
|
||||||
events += birthday.get_birthdays(last_year, os.path.join(my_data, "entities.yaml"))
|
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 += accommodation.get_events(os.path.join(my_data, "accommodation.yaml"))
|
||||||
events += travel.all_events(config["data"]["personal-data"])
|
events += travel.all_events(config["data"]["personal-data"])
|
||||||
events += conference.get_list(os.path.join(my_data, "conferences.yaml"))
|
events += conference.get_list(os.path.join(my_data, "conferences.yaml"))
|
||||||
events += backwell_bins + bristol_bins
|
events += backwell_bins + bristol_bins
|
||||||
events += read_events_yaml(my_data)
|
events += read_events_yaml(my_data, last_year, next_year)
|
||||||
|
|
||||||
events += subscription.get_events(os.path.join(my_data, "subscriptions.yaml"))
|
events += subscription.get_events(os.path.join(my_data, "subscriptions.yaml"))
|
||||||
|
|
||||||
events.sort(key=operator.attrgetter("as_datetime"))
|
events.sort(key=operator.attrgetter("as_datetime"))
|
||||||
|
|
||||||
|
observer = sun.bristol()
|
||||||
|
reply["sunrise"] = sun.sunrise(observer)
|
||||||
|
reply["sunset"] = sun.sunset(observer)
|
||||||
reply["events"] = events
|
reply["events"] = events
|
||||||
reply["last_week"] = last_week
|
reply["last_week"] = last_week
|
||||||
reply["two_weeks_ago"] = two_weeks_ago
|
reply["two_weeks_ago"] = two_weeks_ago
|
||||||
|
|
|
@ -52,30 +52,3 @@ def get_mothers_day(input_date: date) -> date:
|
||||||
mothers_day = easter_date + timedelta(weeks=3)
|
mothers_day = easter_date + timedelta(weeks=3)
|
||||||
|
|
||||||
return mothers_day
|
return mothers_day
|
||||||
|
|
||||||
|
|
||||||
def get_fathers_day(input_date: date) -> date:
|
|
||||||
"""Calculate the date of the next UK Father's Day from the current date."""
|
|
||||||
# Get the current date
|
|
||||||
# Calculate the day of the week for the current date (0 = Monday, 6 = Sunday)
|
|
||||||
current_day_of_week = input_date.weekday()
|
|
||||||
|
|
||||||
# Calculate the number of days until the next Sunday
|
|
||||||
days_until_sunday = (6 - current_day_of_week) % 7
|
|
||||||
|
|
||||||
# Calculate the date of the next Sunday
|
|
||||||
next_sunday = input_date + timedelta(days=days_until_sunday)
|
|
||||||
|
|
||||||
# Calculate the date of Father's Day, which is the third Sunday of June
|
|
||||||
fathers_day = date(next_sunday.year, 6, 1) + timedelta(
|
|
||||||
weeks=2, days=next_sunday.weekday()
|
|
||||||
)
|
|
||||||
|
|
||||||
# Check if Father's Day has already passed this year
|
|
||||||
if input_date > fathers_day:
|
|
||||||
# If it has passed, calculate for the next year
|
|
||||||
fathers_day = date(fathers_day.year + 1, 6, 1) + timedelta(
|
|
||||||
weeks=2, days=next_sunday.weekday()
|
|
||||||
)
|
|
||||||
|
|
||||||
return fathers_day
|
|
||||||
|
|
Loading…
Reference in a new issue