115 lines
4.7 KiB
Python
115 lines
4.7 KiB
Python
"""Trip statistic functions."""
|
|
|
|
from collections import defaultdict
|
|
from typing import Counter, Mapping
|
|
|
|
import agenda
|
|
from agenda.types import StrDict, Trip, airport_label
|
|
|
|
|
|
def travel_legs(trip: Trip, stats: StrDict) -> None:
|
|
"""Calculate stats for travel legs."""
|
|
for leg in trip.travel:
|
|
stats.setdefault("co2_kg", 0)
|
|
stats.setdefault("co2_by_transport_type", {})
|
|
if "co2_kg" in leg:
|
|
stats["co2_kg"] += leg["co2_kg"]
|
|
transport_type = leg["type"]
|
|
stats["co2_by_transport_type"].setdefault(transport_type, 0)
|
|
stats["co2_by_transport_type"][transport_type] += leg["co2_kg"]
|
|
if leg["type"] == "flight":
|
|
stats.setdefault("flight_count", 0)
|
|
stats.setdefault("airlines", Counter())
|
|
stats.setdefault("airports", Counter())
|
|
stats["flight_count"] += 1
|
|
stats["airlines"][leg["airline_detail"]["name"]] += 1
|
|
for field in ("from_airport", "to_airport"):
|
|
airport = leg.get(field)
|
|
if airport:
|
|
country = agenda.get_country(airport.get("country"))
|
|
label = airport_label(airport)
|
|
display = f"{country.flag} {label}" if country else label
|
|
stats["airports"][display] += 1
|
|
if leg["type"] == "train":
|
|
stats.setdefault("train_count", 0)
|
|
stats["train_count"] += 1
|
|
stats.setdefault("stations", Counter())
|
|
train_legs = leg.get("legs", [])
|
|
if train_legs:
|
|
for train_leg in train_legs:
|
|
for field in ("from_station", "to_station"):
|
|
station = train_leg.get(field)
|
|
if station:
|
|
country = agenda.get_country(station.get("country"))
|
|
label = station["name"]
|
|
display = f"{country.flag} {label}" if country else label
|
|
stats["stations"][display] += 1
|
|
else:
|
|
for field in ("from_station", "to_station"):
|
|
station = leg.get(field)
|
|
if station:
|
|
country = agenda.get_country(station.get("country"))
|
|
label = station["name"]
|
|
display = f"{country.flag} {label}" if country else label
|
|
stats["stations"][display] += 1
|
|
|
|
|
|
def conferences(trip: Trip, yearly_stats: Mapping[int, StrDict]) -> None:
|
|
"""Calculate conference stats."""
|
|
for c in trip.conferences:
|
|
yearly_stats[c["start"].year].setdefault("conferences", 0)
|
|
yearly_stats[c["start"].year]["conferences"] += 1
|
|
|
|
|
|
def calculate_yearly_stats(
|
|
trips: list[Trip], previously_visited: set[str] | None = None
|
|
) -> dict[int, StrDict]:
|
|
"""Calculate total distance and distance by transport type grouped by year."""
|
|
yearly_stats: defaultdict[int, StrDict] = defaultdict(dict)
|
|
first_visit_year: dict[str, int] = {}
|
|
excluded_new: set[str] = previously_visited or set()
|
|
|
|
for trip in trips:
|
|
year = trip.start.year
|
|
for country in trip.countries:
|
|
if country.alpha_2 == "GB":
|
|
continue
|
|
alpha_2 = country.alpha_2
|
|
if alpha_2 not in first_visit_year or year < first_visit_year[alpha_2]:
|
|
first_visit_year[alpha_2] = year
|
|
|
|
for trip in trips:
|
|
year = trip.start.year
|
|
dist = trip.total_distance()
|
|
yearly_stats[year].setdefault("count", 0)
|
|
yearly_stats[year]["count"] += 1
|
|
|
|
conferences(trip, yearly_stats)
|
|
|
|
if dist:
|
|
yearly_stats[year]["total_distance"] = (
|
|
yearly_stats[year].get("total_distance", 0) + trip.total_distance()
|
|
)
|
|
|
|
for transport_type, distance in trip.distances_by_transport_type():
|
|
yearly_stats[year].setdefault("distances_by_transport_type", {})
|
|
yearly_stats[year]["distances_by_transport_type"][transport_type] = (
|
|
yearly_stats[year]["distances_by_transport_type"].get(transport_type, 0)
|
|
+ distance
|
|
)
|
|
|
|
for country in trip.countries:
|
|
if country.alpha_2 == "GB":
|
|
continue
|
|
yearly_stats[year].setdefault("countries", set())
|
|
yearly_stats[year]["countries"].add(country)
|
|
if (
|
|
first_visit_year.get(country.alpha_2) == year
|
|
and country.alpha_2 not in excluded_new
|
|
):
|
|
yearly_stats[year].setdefault("new_countries", set())
|
|
yearly_stats[year]["new_countries"].add(country)
|
|
|
|
travel_legs(trip, yearly_stats[year])
|
|
|
|
return dict(yearly_stats)
|