From 7bb6110f450db784358c6c84ef75699fe81e73d6 Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Thu, 16 May 2024 15:20:58 +0100 Subject: [PATCH] Split out code for reading events from YAML --- agenda/data.py | 78 +-------------------------------------- agenda/events_yaml.py | 85 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 76 deletions(-) create mode 100644 agenda/events_yaml.py diff --git a/agenda/data.py b/agenda/data.py index 9699dac..3f6de3e 100644 --- a/agenda/data.py +++ b/agenda/data.py @@ -10,10 +10,8 @@ from time import time import dateutil.rrule import dateutil.tz import flask -import isodate # type: ignore import lxml import pytz -import yaml from . import ( accommodation, @@ -23,6 +21,7 @@ from . import ( conference, domains, economist, + events_yaml, gandi, gwr, hn, @@ -61,25 +60,6 @@ def timezone_transition( ] -def midnight(d: date) -> datetime: - """Convert from date to midnight on that day.""" - return datetime.combine(d, datetime.min.time()) - - -def dates_from_rrule( - rrule: str, start: date, end: date -) -> typing.Sequence[datetime | date]: - """Generate events from an RRULE between start_date and end_date.""" - all_day = not any(param in rrule for param in ["BYHOUR", "BYMINUTE", "BYSECOND"]) - - return [ - i.date() if all_day else uk_tz.localize(i) - for i in dateutil.rrule.rrulestr(rrule, dtstart=midnight(start)).between( - midnight(start), midnight(end) - ) - ] - - async def waste_collection_events( data_dir: str, postcode: str, uprn: str ) -> list[Event]: @@ -97,60 +77,6 @@ async def bristol_waste_collection_events( return await waste_schedule.get_bristol_gov_uk(start_date, data_dir, uprn) -def get_yaml_event_date_field(item: dict[str, str]) -> str: - """Event date field name.""" - return ( - "end_date" - if item["name"] == "travel_insurance" - else ("start_date" if "start_date" in item else "date") - ) - - -def get_yaml_event_end_date_field(item: dict[str, str]) -> str: - """Event date field name.""" - return ( - "end_date" - if item["name"] == "travel_insurance" - else ("start_date" if "start_date" in item else "date") - ) - - -def read_events_yaml( - data_dir: str, start: date, end: date, skip_trips: bool = False -) -> list[Event]: - """Read eventes from YAML file.""" - events: list[Event] = [] - for item in yaml.safe_load(open(os.path.join(data_dir, "events.yaml"))): - if "trip" in item and skip_trips: - continue - duration = ( - isodate.parse_duration(item["duration"]) if "duration" in item else None - ) - dates = ( - dates_from_rrule(item["rrule"], start, end) - if "rrule" in item - else [item[get_yaml_event_date_field(item)]] - ) - for dt in dates: - e = Event( - name=item["name"], - date=dt, - end_date=( - dt + duration - if duration - else ( - item.get("end_date") - if item["name"] != "travel_insurance" - else None - ) - ), - title=item.get("title"), - url=item.get("url"), - ) - events.append(e) - return events - - def find_events_during_stay( accommodation_events: list[Event], markets: list[Event] ) -> list[Event]: @@ -426,7 +352,7 @@ async def get_data( for key in "backwell_bins", "bristol_bins": if results[key]: events += results[key] - events += read_events_yaml(my_data, last_year, next_year) + events += events_yaml.read(my_data, last_year, next_year) events += subscription.get_events(os.path.join(my_data, "subscriptions.yaml")) events += gandi.get_events(data_dir) events += economist.publication_dates(last_week, next_year) diff --git a/agenda/events_yaml.py b/agenda/events_yaml.py new file mode 100644 index 0000000..ca55eea --- /dev/null +++ b/agenda/events_yaml.py @@ -0,0 +1,85 @@ +"""Read events from YAML.""" + +import os +import typing +from datetime import date, datetime + +import dateutil.rrule +import isodate # type: ignore +import yaml + +from . import uk_tz +from .types import Event + + +def midnight(d: date) -> datetime: + """Convert from date to midnight on that day.""" + return datetime.combine(d, datetime.min.time()) + + +def dates_from_rrule( + rrule: str, start: date, end: date +) -> typing.Sequence[datetime | date]: + """Generate events from an RRULE between start_date and end_date.""" + all_day = not any(param in rrule for param in ["BYHOUR", "BYMINUTE", "BYSECOND"]) + + return [ + i.date() if all_day else uk_tz.localize(i) + for i in dateutil.rrule.rrulestr(rrule, dtstart=midnight(start)).between( + midnight(start), midnight(end) + ) + ] + + +def get_yaml_event_date_field(item: dict[str, str]) -> str: + """Event date field name.""" + return ( + "end_date" + if item["name"] == "travel_insurance" + else ("start_date" if "start_date" in item else "date") + ) + + +def get_yaml_event_end_date_field(item: dict[str, str]) -> str: + """Event date field name.""" + return ( + "end_date" + if item["name"] == "travel_insurance" + else ("start_date" if "start_date" in item else "date") + ) + + +def read( + data_dir: str, start: date, end: date, skip_trips: bool = False +) -> list[Event]: + """Read eventes from YAML file.""" + events: list[Event] = [] + for item in yaml.safe_load(open(os.path.join(data_dir, "events.yaml"))): + if "trip" in item and skip_trips: + continue + duration = ( + isodate.parse_duration(item["duration"]) if "duration" in item else None + ) + dates = ( + dates_from_rrule(item["rrule"], start, end) + if "rrule" in item + else [item[get_yaml_event_date_field(item)]] + ) + for dt in dates: + e = Event( + name=item["name"], + date=dt, + end_date=( + dt + duration + if duration + else ( + item.get("end_date") + if item["name"] != "travel_insurance" + else None + ) + ), + title=item.get("title"), + url=item.get("url"), + ) + events.append(e) + return events