Compare commits
4 commits
78c90b0164
...
3ec7f5c18a
Author | SHA1 | Date | |
---|---|---|---|
Edward Betts | 3ec7f5c18a | ||
Edward Betts | dd59c809e1 | ||
Edward Betts | 7bb6110f45 | ||
Edward Betts | 18d8fa6b7c |
159
agenda/busy.py
Normal file
159
agenda/busy.py
Normal 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
33
agenda/carnival.py
Normal 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
|
266
agenda/data.py
266
agenda/data.py
|
@ -1,7 +1,6 @@
|
||||||
"""Agenda data."""
|
"""Agenda data."""
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import itertools
|
|
||||||
import os
|
import os
|
||||||
import typing
|
import typing
|
||||||
from datetime import date, datetime, timedelta
|
from datetime import date, datetime, timedelta
|
||||||
|
@ -10,19 +9,19 @@ from time import time
|
||||||
import dateutil.rrule
|
import dateutil.rrule
|
||||||
import dateutil.tz
|
import dateutil.tz
|
||||||
import flask
|
import flask
|
||||||
import isodate # type: ignore
|
|
||||||
import lxml
|
import lxml
|
||||||
import pytz
|
import pytz
|
||||||
import yaml
|
|
||||||
from dateutil.easter import easter
|
|
||||||
|
|
||||||
from . import (
|
from . import (
|
||||||
accommodation,
|
accommodation,
|
||||||
birthday,
|
birthday,
|
||||||
|
busy,
|
||||||
calendar,
|
calendar,
|
||||||
|
carnival,
|
||||||
conference,
|
conference,
|
||||||
domains,
|
domains,
|
||||||
economist,
|
economist,
|
||||||
|
events_yaml,
|
||||||
gandi,
|
gandi,
|
||||||
gwr,
|
gwr,
|
||||||
hn,
|
hn,
|
||||||
|
@ -34,10 +33,9 @@ from . import (
|
||||||
thespacedevs,
|
thespacedevs,
|
||||||
travel,
|
travel,
|
||||||
uk_holiday,
|
uk_holiday,
|
||||||
uk_tz,
|
|
||||||
waste_schedule,
|
waste_schedule,
|
||||||
)
|
)
|
||||||
from .types import Event, StrDict, Trip
|
from .types import Event
|
||||||
|
|
||||||
here = dateutil.tz.tzlocal()
|
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(
|
async def waste_collection_events(
|
||||||
data_dir: str, postcode: str, uprn: str
|
data_dir: str, postcode: str, uprn: str
|
||||||
) -> list[Event]:
|
) -> 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)
|
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(
|
def find_events_during_stay(
|
||||||
accommodation_events: list[Event], markets: list[Event]
|
accommodation_events: list[Event], markets: list[Event]
|
||||||
) -> list[Event]:
|
) -> list[Event]:
|
||||||
|
@ -195,84 +94,6 @@ def find_events_during_stay(
|
||||||
return overlapping_markets
|
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(
|
async def time_function(
|
||||||
name: str,
|
name: str,
|
||||||
func: typing.Callable[..., typing.Coroutine[typing.Any, typing.Any, typing.Any]],
|
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
|
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(
|
async def get_data(
|
||||||
now: datetime, config: flask.config.Config
|
now: datetime, config: flask.config.Config
|
||||||
) -> typing.Mapping[str, str | object]:
|
) -> typing.Mapping[str, str | object]:
|
||||||
|
@ -452,13 +202,13 @@ async def get_data(
|
||||||
for key in "backwell_bins", "bristol_bins":
|
for key in "backwell_bins", "bristol_bins":
|
||||||
if results[key]:
|
if results[key]:
|
||||||
events += 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 += subscription.get_events(os.path.join(my_data, "subscriptions.yaml"))
|
||||||
events += gandi.get_events(data_dir)
|
events += gandi.get_events(data_dir)
|
||||||
events += economist.publication_dates(last_week, next_year)
|
events += economist.publication_dates(last_week, next_year)
|
||||||
events += meetup.get_events(my_data)
|
events += meetup.get_events(my_data)
|
||||||
events += hn.whoishiring(last_year, next_year)
|
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
|
# hide markets that happen while away
|
||||||
optional = [
|
optional = [
|
||||||
|
@ -503,10 +253,10 @@ async def get_data(
|
||||||
busy_events = [
|
busy_events = [
|
||||||
e
|
e
|
||||||
for e in sorted(events, key=lambda e: e.as_date)
|
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 += [
|
events += [
|
||||||
Event(name="gap", date=gap["start"], end_date=gap["end"]) for gap in gaps
|
Event(name="gap", date=gap["start"], end_date=gap["end"]) for gap in gaps
|
||||||
|
|
85
agenda/events_yaml.py
Normal file
85
agenda/events_yaml.py
Normal 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
|
|
@ -268,6 +268,26 @@ class Event:
|
||||||
else None
|
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:
|
def delta_days(self, today: datetime.date) -> str:
|
||||||
"""Return number of days from today as a string."""
|
"""Return number of days from today as a string."""
|
||||||
delta = (self.as_date - today).days
|
delta = (self.as_date - today).days
|
||||||
|
|
|
@ -174,9 +174,10 @@
|
||||||
|
|
||||||
<div class="col-md-2{{ cell_bg }}">
|
<div class="col-md-2{{ cell_bg }}">
|
||||||
{% if event.end_date %}
|
{% 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") }}
|
end: {{event.end_date.strftime("%H:%M") }}
|
||||||
(duration: {{event.end_date - event.date}})
|
(duration: {{duration}})
|
||||||
{% elif event.end_date != event.date %}
|
{% elif event.end_date != event.date %}
|
||||||
{{event.end_date}}
|
{{event.end_date}}
|
||||||
{% endif %}
|
{% endif %}
|
||||||
|
|
|
@ -92,8 +92,8 @@ async def gaps_page() -> str:
|
||||||
"""List of available gaps."""
|
"""List of available gaps."""
|
||||||
now = datetime.now()
|
now = datetime.now()
|
||||||
trip_list = agenda.trip.build_trip_list()
|
trip_list = agenda.trip.build_trip_list()
|
||||||
busy_events = agenda.data.get_busy_events(now.date(), app.config, trip_list)
|
busy_events = agenda.busy.get_busy_events(now.date(), app.config, trip_list)
|
||||||
gaps = agenda.data.find_gaps(busy_events)
|
gaps = agenda.busy.find_gaps(busy_events)
|
||||||
return flask.render_template("gaps.html", today=now.date(), gaps=gaps)
|
return flask.render_template("gaps.html", today=now.date(), gaps=gaps)
|
||||||
|
|
||||||
|
|
||||||
|
@ -102,8 +102,8 @@ async def weekends() -> str:
|
||||||
"""List of available gaps."""
|
"""List of available gaps."""
|
||||||
now = datetime.now()
|
now = datetime.now()
|
||||||
trip_list = agenda.trip.build_trip_list()
|
trip_list = agenda.trip.build_trip_list()
|
||||||
busy_events = agenda.data.get_busy_events(now.date(), app.config, trip_list)
|
busy_events = agenda.busy.get_busy_events(now.date(), app.config, trip_list)
|
||||||
weekends = agenda.data.weekends(busy_events)
|
weekends = agenda.busy.weekends(busy_events)
|
||||||
return flask.render_template("weekends.html", today=now.date(), items=weekends)
|
return flask.render_template("weekends.html", today=now.date(), items=weekends)
|
||||||
|
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue