Refactor bristol waste schedule code

This commit is contained in:
Edward Betts 2024-07-07 13:27:00 +01:00
parent e677082560
commit 4ef47374c2

View file

@ -188,25 +188,38 @@ async def get_bristol_gov_uk_data(uprn: str) -> httpx.Response:
return 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( async def get_bristol_gov_uk(
start_date: date, data_dir: str, uprn: str, refresh: bool = False start_date: date, data_dir: str, uprn: str, refresh: bool = False
) -> list[Event]: ) -> list[Event]:
"""Get waste collection schedule from Bristol City Council.""" """Get waste collection schedule from Bristol City Council."""
data = await get_bristol_data(data_dir, uprn, refresh) data = await get_bristol_data(data_dir, uprn, refresh)
by_date: defaultdict[date, list[str]] = defaultdict(list) by_date = bristol_by_date(data, start_date)
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)
return [ return [
Event(name="waste_schedule", date=d, title="Bristol: " + ", ".join(services)) Event(name="waste_schedule", date=d, title="Bristol: " + ", ".join(services))
for d, services in by_date.items() for d, services in by_date.items()