Compare commits

..

4 commits

7 changed files with 312 additions and 264 deletions

159
agenda/busy.py Normal file
View file

@ -0,0 +1,159 @@
"""Identify busy events and gaps when nothing is scheduled."""
import itertools
import typing
from datetime import date, datetime, timedelta
import flask
from . import events_yaml
from .types import Event, StrDict, Trip
def busy_event(e: Event) -> bool:
"""Busy."""
if e.name not in {
"event",
"accommodation",
"conference",
"transport",
"meetup",
"party",
"trip",
}:
return False
if e.title in ("IA UK board meeting", "Mill Road Winter Fair"):
return False
if e.name == "conference" and not e.going:
return False
if not e.title:
return True
if e.title == "LHG Run Club" or "Third Thursday Social" in e.title:
return False
lc_title = e.title.lower()
return (
"rebels" not in lc_title
and "south west data social" not in lc_title
and "dorkbot" not in lc_title
)
def get_busy_events(
today: date, config: flask.config.Config, trips: list[Trip]
) -> list[Event]:
"""Find busy events from a year ago to two years in the future."""
last_year = today - timedelta(days=365)
next_year = today + timedelta(days=2 * 365)
my_data = config["PERSONAL_DATA"]
events = events_yaml.read(my_data, last_year, next_year, skip_trips=True)
for trip in trips:
event_type = "trip"
if trip.events and not trip.conferences:
event_type = trip.events[0]["name"]
elif len(trip.conferences) == 1 and trip.conferences[0].get("hackathon"):
event_type = "hackathon"
events.append(
Event(
name=event_type,
title=trip.title + " " + trip.country_flags,
date=trip.start,
end_date=trip.end,
url=flask.url_for("trip_page", start=trip.start.isoformat()),
)
)
busy_events = [
e
for e in sorted(events, key=lambda e: e.as_date)
if (e.as_date >= today or (e.end_date and e.end_as_date >= today))
and e.as_date < next_year
and busy_event(e)
]
return busy_events
def weekends(busy_events: list[Event]) -> typing.Sequence[StrDict]:
"""Next ten weekends."""
today = datetime.today()
weekday = today.weekday()
# Calculate the difference to the next or previous Saturday
if weekday == 6: # Sunday
start_date = (today - timedelta(days=1)).date()
else:
start_date = (today + timedelta(days=(5 - weekday))).date()
weekends_info = []
for i in range(52):
saturday = start_date + timedelta(weeks=i)
sunday = saturday + timedelta(days=1)
saturday_events = [
event
for event in busy_events
if event.end_date and event.as_date <= saturday <= event.end_as_date
]
sunday_events = [
event
for event in busy_events
if event.end_date and event.as_date <= sunday <= event.end_as_date
]
weekends_info.append(
{"date": saturday, "saturday": saturday_events, "sunday": sunday_events}
)
return weekends_info
def find_gaps(events: list[Event], min_gap_days: int = 3) -> list[StrDict]:
"""Gaps of at least `min_gap_days` between events in a list of events."""
# Sort events by start date
gaps: list[tuple[date, date]] = []
previous_event_end = None
by_start_date = {
d: list(on_day)
for d, on_day in itertools.groupby(events, key=lambda e: e.as_date)
}
by_end_date = {
d: list(on_day)
for d, on_day in itertools.groupby(events, key=lambda e: e.end_as_date)
}
for event in events:
# Use start date for current event
start_date = event.as_date
# If previous event exists, calculate the gap
if previous_event_end:
gap_days = (start_date - previous_event_end).days
if gap_days >= (min_gap_days + 2):
start_end = (
previous_event_end + timedelta(days=1),
start_date - timedelta(days=1),
)
gaps.append(start_end)
# Update previous event end date
end = event.end_as_date
if not previous_event_end or end > previous_event_end:
previous_event_end = end
return [
{
"start": gap_start,
"end": gap_end,
"after": by_start_date[gap_end + timedelta(days=1)],
"before": by_end_date[gap_start - timedelta(days=1)],
}
for gap_start, gap_end in gaps
]

33
agenda/carnival.py Normal file
View file

@ -0,0 +1,33 @@
"""Calcuate the date for carnival."""
from datetime import date, timedelta
from dateutil.easter import easter
from .types import Event
def rio_carnival_events(start_date: date, end_date: date) -> list[Event]:
"""List of events for Rio Carnival for each year between start_date and end_date."""
events = []
for year in range(start_date.year, end_date.year + 1):
easter_date = easter(year)
carnival_start = easter_date - timedelta(days=51)
carnival_end = easter_date - timedelta(days=46)
# Only include the carnival if it falls within the specified date range
if (
start_date <= carnival_start <= end_date
or start_date <= carnival_end <= end_date
):
events.append(
Event(
name="carnival",
title="Rio Carnival",
date=carnival_start,
end_date=carnival_end,
url="https://en.wikipedia.org/wiki/Rio_Carnival",
)
)
return events

View file

@ -1,7 +1,6 @@
"""Agenda data."""
import asyncio
import itertools
import os
import typing
from datetime import date, datetime, timedelta
@ -10,19 +9,19 @@ from time import time
import dateutil.rrule
import dateutil.tz
import flask
import isodate # type: ignore
import lxml
import pytz
import yaml
from dateutil.easter import easter
from . import (
accommodation,
birthday,
busy,
calendar,
carnival,
conference,
domains,
economist,
events_yaml,
gandi,
gwr,
hn,
@ -34,10 +33,9 @@ from . import (
thespacedevs,
travel,
uk_holiday,
uk_tz,
waste_schedule,
)
from .types import Event, StrDict, Trip
from .types import Event
here = dateutil.tz.tzlocal()
@ -61,51 +59,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 rio_carnival_events(start_date: date, end_date: date) -> list[Event]:
"""List of events for Rio Carnival for each year between start_date and end_date."""
events = []
for year in range(start_date.year, end_date.year + 1):
easter_date = easter(year)
carnival_start = easter_date - timedelta(days=51)
carnival_end = easter_date - timedelta(days=46)
# Only include the carnival if it falls within the specified date range
if (
start_date <= carnival_start <= end_date
or start_date <= carnival_end <= end_date
):
events.append(
Event(
name="carnival",
title="Rio Carnival",
date=carnival_start,
end_date=carnival_end,
url="https://en.wikipedia.org/wiki/Rio_Carnival",
)
)
return events
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]:
@ -123,60 +76,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]:
@ -195,84 +94,6 @@ def find_events_during_stay(
return overlapping_markets
def find_gaps(events: list[Event], min_gap_days: int = 3) -> list[StrDict]:
"""Gaps of at least `min_gap_days` between events in a list of events."""
# Sort events by start date
gaps: list[tuple[date, date]] = []
previous_event_end = None
by_start_date = {
d: list(on_day)
for d, on_day in itertools.groupby(events, key=lambda e: e.as_date)
}
by_end_date = {
d: list(on_day)
for d, on_day in itertools.groupby(events, key=lambda e: e.end_as_date)
}
for event in events:
# Use start date for current event
start_date = event.as_date
# If previous event exists, calculate the gap
if previous_event_end:
gap_days = (start_date - previous_event_end).days
if gap_days >= (min_gap_days + 2):
start_end = (
previous_event_end + timedelta(days=1),
start_date - timedelta(days=1),
)
gaps.append(start_end)
# Update previous event end date
end = event.end_as_date
if not previous_event_end or end > previous_event_end:
previous_event_end = end
return [
{
"start": gap_start,
"end": gap_end,
"after": by_start_date[gap_end + timedelta(days=1)],
"before": by_end_date[gap_start - timedelta(days=1)],
}
for gap_start, gap_end in gaps
]
def busy_event(e: Event) -> bool:
"""Busy."""
if e.name not in {
"event",
"accommodation",
"conference",
"transport",
"meetup",
"party",
"trip",
}:
return False
if e.title in ("IA UK board meeting", "Mill Road Winter Fair"):
return False
if e.name == "conference" and not e.going:
return False
if not e.title:
return True
if e.title == "LHG Run Club" or "Third Thursday Social" in e.title:
return False
lc_title = e.title.lower()
return (
"rebels" not in lc_title
and "south west data social" not in lc_title
and "dorkbot" not in lc_title
)
async def time_function(
name: str,
func: typing.Callable[..., typing.Coroutine[typing.Any, typing.Any, typing.Any]],
@ -289,77 +110,6 @@ async def time_function(
return name, result, end_time - start_time, exception
def get_busy_events(
today: date, config: flask.config.Config, trips: list[Trip]
) -> list[Event]:
"""Find busy events from a year ago to two years in the future."""
last_year = today - timedelta(days=365)
next_year = today + timedelta(days=2 * 365)
my_data = config["PERSONAL_DATA"]
events = read_events_yaml(my_data, last_year, next_year, skip_trips=True)
for trip in trips:
event_type = "trip"
if trip.events and not trip.conferences:
event_type = trip.events[0]["name"]
elif len(trip.conferences) == 1 and trip.conferences[0].get("hackathon"):
event_type = "hackathon"
events.append(
Event(
name=event_type,
title=trip.title + " " + trip.country_flags,
date=trip.start,
end_date=trip.end,
url=flask.url_for("trip_page", start=trip.start.isoformat()),
)
)
busy_events = [
e
for e in sorted(events, key=lambda e: e.as_date)
if (e.as_date >= today or (e.end_date and e.end_as_date >= today))
and e.as_date < next_year
and busy_event(e)
]
return busy_events
def weekends(busy_events: list[Event]) -> typing.Sequence[StrDict]:
"""Next ten weekends."""
today = datetime.today()
weekday = today.weekday()
# Calculate the difference to the next or previous Saturday
if weekday == 6: # Sunday
start_date = (today - timedelta(days=1)).date()
else:
start_date = (today + timedelta(days=(5 - weekday))).date()
weekends_info = []
for i in range(52):
saturday = start_date + timedelta(weeks=i)
sunday = saturday + timedelta(days=1)
saturday_events = [
event
for event in busy_events
if event.end_date and event.as_date <= saturday <= event.end_as_date
]
sunday_events = [
event
for event in busy_events
if event.end_date and event.as_date <= sunday <= event.end_as_date
]
weekends_info.append(
{"date": saturday, "saturday": saturday_events, "sunday": sunday_events}
)
return weekends_info
async def get_data(
now: datetime, config: flask.config.Config
) -> typing.Mapping[str, str | object]:
@ -452,13 +202,13 @@ 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)
events += meetup.get_events(my_data)
events += hn.whoishiring(last_year, next_year)
events += rio_carnival_events(last_year, next_year)
events += carnival.rio_carnival_events(last_year, next_year)
# hide markets that happen while away
optional = [
@ -503,10 +253,10 @@ async def get_data(
busy_events = [
e
for e in sorted(events, key=lambda e: e.as_date)
if e.as_date > today and e.as_date < next_year and busy_event(e)
if e.as_date > today and e.as_date < next_year and busy.busy_event(e)
]
gaps = find_gaps(busy_events)
gaps = busy.find_gaps(busy_events)
events += [
Event(name="gap", date=gap["start"], end_date=gap["end"]) for gap in gaps

85
agenda/events_yaml.py Normal file
View file

@ -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

View file

@ -268,6 +268,26 @@ class Event:
else None
)
def display_duration(self) -> str | None:
"""Duration for display."""
if self.end_as_date != self.as_date or not self.has_time:
return None
assert isinstance(self.date, datetime.datetime)
assert isinstance(self.end_date, datetime.datetime)
secs: int = int((self.end_date - self.date).total_seconds())
hours: int = secs // 3600
mins: int = (secs % 3600) // 60
if mins == 0:
return f"{hours:d}h"
if hours == 0:
return f"{mins:d} mins"
return f"{hours:d}h {mins:02d} mins"
def delta_days(self, today: datetime.date) -> str:
"""Return number of days from today as a string."""
delta = (self.as_date - today).days

View file

@ -174,9 +174,10 @@
<div class="col-md-2{{ cell_bg }}">
{% if event.end_date %}
{% if event.end_as_date == event.as_date and event.has_time %}
{% set duration = event.display_duration() %}
{% if duration %}
end: {{event.end_date.strftime("%H:%M") }}
(duration: {{event.end_date - event.date}})
(duration: {{duration}})
{% elif event.end_date != event.date %}
{{event.end_date}}
{% endif %}

View file

@ -92,8 +92,8 @@ async def gaps_page() -> str:
"""List of available gaps."""
now = datetime.now()
trip_list = agenda.trip.build_trip_list()
busy_events = agenda.data.get_busy_events(now.date(), app.config, trip_list)
gaps = agenda.data.find_gaps(busy_events)
busy_events = agenda.busy.get_busy_events(now.date(), app.config, trip_list)
gaps = agenda.busy.find_gaps(busy_events)
return flask.render_template("gaps.html", today=now.date(), gaps=gaps)
@ -102,8 +102,8 @@ async def weekends() -> str:
"""List of available gaps."""
now = datetime.now()
trip_list = agenda.trip.build_trip_list()
busy_events = agenda.data.get_busy_events(now.date(), app.config, trip_list)
weekends = agenda.data.weekends(busy_events)
busy_events = agenda.busy.get_busy_events(now.date(), app.config, trip_list)
weekends = agenda.busy.weekends(busy_events)
return flask.render_template("weekends.html", today=now.date(), items=weekends)