activitysim_visualizer/processor/summarize/summaries/long_term_vehicle.py at main · RSGInc/activitysim_visualizer · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
"""Vehicle and household-asset long-term summaries."""

from __future__ import annotations

import polars as pl

from processor.models import RunData
from processor.summarize.contracts import summary
from runtime.config import Config


@summary(
    id="autonomous_vehicle_ownership_totals",
    schema={"household_with_autonomous_vehicle_count": pl.Float64},
    required_columns={"hh": ("av_ownership", "finalweight")},
)
def av_ownership(rd: RunData, config: Config) -> pl.DataFrame:
    required = {"av_ownership", "finalweight"}
    if not required.issubset(set(rd.hh.columns)):
        return av_ownership.empty()

    return rd.hh.filter(pl.col("av_ownership") == True).select(
        pl.col("finalweight")
        .sum()
        .cast(pl.Float64)
        .alias("household_with_autonomous_vehicle_count")
    )


@summary(
    id="auto_ownership_distribution",
    schema={
        "household_size": pl.Utf8,
        "household_vehicle_count": pl.Int64,
        "household_count": pl.Float64,
    },
    required_columns={"hh": ("HHSIZE", "HHVEH", "finalweight")},
)
def auto_ownership(rd: RunData, config: Config) -> pl.DataFrame:
    return (
        rd.hh.with_columns(
            pl.when(pl.col("HHSIZE").cast(pl.Int64, strict=False) >= 5)
            .then(pl.lit("5+"))
            .otherwise(pl.col("HHSIZE").cast(pl.Int64, strict=False).cast(pl.Utf8))
            .alias("household_size")
        )
        .group_by(["household_size", "HHVEH"])
        .agg(household_count=pl.col("finalweight").sum())
        .rename({"HHVEH": "household_vehicle_count"})
        .with_columns(
            pl.col("household_size").cast(pl.Utf8),
            pl.col("household_vehicle_count").cast(pl.Int64),
            pl.col("household_count").cast(pl.Float64),
            pl.when(pl.col("household_size") == "5+")
            .then(999)
            .otherwise(pl.col("household_size").cast(pl.Int64, strict=False))
            .alias("_sort_household_size"),
        )
        .sort(["_sort_household_size", "household_vehicle_count"])
        .select("household_size", "household_vehicle_count", "household_count")
    )


@summary(
    id="vehicle_age_distribution",
    schema={
        "age": pl.Utf8,
        "vehicle_count": pl.Float64,
    },
    required_columns={"vehicles": ("vehicle_age", "finalweight")},
)
def vehicle_char_age(rd: RunData, config: Config) -> pl.DataFrame:
    required = {"vehicle_age", "finalweight"}
    if not hasattr(rd, "vehicles"):
        return vehicle_char_age.empty()
    if not required.issubset(set(rd.vehicles.columns)):
        return vehicle_char_age.empty()

    return (
        rd.vehicles.filter(pl.col("vehicle_age").is_not_null())
        .with_columns(
            pl.when(pl.col("vehicle_age") >= 20)
            .then(pl.lit("20+"))
            .otherwise(pl.col("vehicle_age").cast(pl.Utf8))
            .alias("age")
        )
        .group_by("age")
        .agg(vehicle_count=pl.col("finalweight").sum())
        .with_columns(
            pl.col("age").cast(pl.Utf8),
            pl.col("vehicle_count").cast(pl.Float64),
            pl.when(pl.col("age") == "20+")
            .then(999)
            .otherwise(pl.col("age").cast(pl.Int64, strict=False))
            .alias("_sort_age"),
        )
        .sort("_sort_age")
        .select("age", "vehicle_count")
    )


@summary(
    id="vehicle_fuel_type_distribution",
    schema={
        "fuel_type": pl.Utf8,
        "vehicle_count": pl.Float64,
    },
    required_columns={"vehicles": ("fuel_type", "finalweight")},
)
def vehicle_char_fuel(rd: RunData, config: Config) -> pl.DataFrame:
    required = {"fuel_type", "finalweight"}
    if not hasattr(rd, "vehicles"):
        return vehicle_char_fuel.empty()
    if not required.issubset(set(rd.vehicles.columns)):
        return vehicle_char_fuel.empty()

    return (
        rd.vehicles.filter(pl.col("fuel_type").is_not_null())
        .group_by("fuel_type")
        .agg(vehicle_count=pl.col("finalweight").sum())
        .with_columns(
            pl.col("fuel_type").cast(pl.Utf8),
            pl.col("vehicle_count").cast(pl.Float64),
        )
        .select("fuel_type", "vehicle_count")
        .sort("fuel_type")
    )


@summary(
    id="vehicle_body_type_distribution",
    schema={
        "body_type": pl.Utf8,
        "vehicle_count": pl.Float64,
    },
    required_columns={"vehicles": ("body_type", "finalweight")},
)
def vehicle_char_body(rd: RunData, config: Config) -> pl.DataFrame:
    required = {"body_type", "finalweight"}
    if not hasattr(rd, "vehicles"):
        return vehicle_char_body.empty()
    if not required.issubset(set(rd.vehicles.columns)):
        return vehicle_char_body.empty()

    return (
        rd.vehicles.filter(pl.col("body_type").is_not_null())
        .group_by("body_type")
        .agg(vehicle_count=pl.col("finalweight").sum())
        .with_columns(
            pl.col("body_type").cast(pl.Utf8),
            pl.col("vehicle_count").cast(pl.Float64),
        )
        .select("body_type", "vehicle_count")
        .sort("body_type")
    )