From 4ef47374c23f7dff3c162ead3e1ffe6497a46a88 Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Sun, 7 Jul 2024 13:27:00 +0100 Subject: [PATCH] Refactor bristol waste schedule code --- agenda/waste_schedule.py | 39 ++++++++++++++++++++++++++------------- 1 file changed, 26 insertions(+), 13 deletions(-) diff --git a/agenda/waste_schedule.py b/agenda/waste_schedule.py index 6f0349b..61be501 100644 --- a/agenda/waste_schedule.py +++ b/agenda/waste_schedule.py @@ -188,25 +188,38 @@ async def get_bristol_gov_uk_data(uprn: str) -> httpx.Response: return response +def bristol_service(item: dict[str, typing.Any]) -> str: + """Bristol waste service name.""" + service: str = item["containerName"] + return "Recycling" if "Recycling" in service else service.partition(" ")[2] + + +def bristol_collections(item: dict[str, typing.Any]) -> typing.Iterable[date]: + """Bristol dates from collections.""" + for collection in item["collection"]: + for collection_date_key in ["nextCollectionDate", "lastCollectionDate"]: + yield date.fromisoformat(collection[collection_date_key][:10]) + + +def bristol_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 = bristol_service(item) + for d in bristol_collections(item): + if d < start_date and service not in by_date[d]: + by_date[d].append(service) + return by_date + + async def get_bristol_gov_uk( start_date: date, data_dir: str, uprn: str, refresh: bool = False ) -> list[Event]: """Get waste collection schedule from Bristol City Council.""" data = await get_bristol_data(data_dir, uprn, refresh) - by_date: defaultdict[date, list[str]] = defaultdict(list) - - for item in data: - service = item["containerName"] - service = "Recycling" if "Recycling" in service else service.partition(" ")[2] - for collection in item["collection"]: - for collection_date_key in ["nextCollectionDate", "lastCollectionDate"]: - d = date.fromisoformat(collection[collection_date_key][:10]) - if d < start_date: - continue - if service not in by_date[d]: - by_date[d].append(service) - + by_date = bristol_by_date(data, start_date) return [ Event(name="waste_schedule", date=d, title="Bristol: " + ", ".join(services)) for d, services in by_date.items()