edward/agenda
1
0
Fork 0
Code Issues 27 Pull requests Packages Projects Releases Wiki Activity

Reorder functions

Browse source
This commit is contained in:
Edward Betts 2024-07-15 03:17:29 +08:00
parent ef695af7af
commit ace517c482
1 changed files with 17 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

42
agenda/bristol_waste.py
View file

@ -17,6 +17,23 @@ ttl_hours = 12
BristolSchedule = list[dict[str, typing.Any]] BristolSchedule = list[dict[str, typing.Any]]
async def get(
start_date: date, data_dir: str, uprn: str, refresh: bool = False
) -> list[Event]:
"""Get waste collection schedule from Bristol City Council."""
by_date: defaultdict[date, list[str]] = defaultdict(list)
for item in await get_data(data_dir, uprn, refresh):
service = get_service(item)
for d in collections(item):
if d < start_date and service not in by_date[d]:
by_date[d].append(service)
return [
Event(name="waste_schedule", date=d, title="Bristol: " + ", ".join(services))
for d, services in by_date.items()
]
async def get_data(data_dir: str, uprn: str, refresh: bool = False) -> BristolSchedule: async def get_data(data_dir: str, uprn: str, refresh: bool = False) -> BristolSchedule:
"""Get Bristol Waste schedule, with cache.""" """Get Bristol Waste schedule, with cache."""
now = datetime.now() now = datetime.now()
@ -110,28 +127,3 @@ def collections(item: dict[str, typing.Any]) -> typing.Iterable[date]:
for collection in item["collection"]: for collection in item["collection"]:
for collection_date_key in ["nextCollectionDate", "lastCollectionDate"]: for collection_date_key in ["nextCollectionDate", "lastCollectionDate"]:
yield date.fromisoformat(collection[collection_date_key][:10]) yield date.fromisoformat(collection[collection_date_key][:10])
def get_by_date(data: BristolSchedule, start_date: date) -> dict[date, list[str]]:
"""Build list of Bristol services by date."""
by_date: defaultdict[date, list[str]] = defaultdict(list)
for item in data:
service = get_service(item)
for d in collections(item):
if d < start_date and service not in by_date[d]:
by_date[d].append(service)
return by_date
async def get(
start_date: date, data_dir: str, uprn: str, refresh: bool = False
) -> list[Event]:
"""Get waste collection schedule from Bristol City Council."""
data = await get_data(data_dir, uprn, refresh)
by_date = get_by_date(data, start_date)
return [
Event(name="waste_schedule", date=d, title="Bristol: " + ", ".join(services))
for d, services in by_date.items()
]