Skip to content

pam.write.matrices

write_od_matrices(population, path, leg_filter=None, person_filter=None, time_minutes_filter=None) #

Write a core population object to tabular O-D weighted matrices.

Optionally segment matrices by leg attributes(mode/ purpose), person attributes or specific time periods. A single filter can be applied each time.

TODO include freq (assume hh).

PARAMETER DESCRIPTION
population

TYPE: Population

path

directory to write OD matrix files

TYPE: str

leg_filter

select between 'Mode', 'Purpose'. Defaults to None.

TYPE: Optional[str] DEFAULT: None

person_filter

select between given attribute categories (column names) from person attribute data. Defaults to None.

TYPE: Optional[str] DEFAULT: None

time_minutes_filter

a list of tuples to slice times, e.g. [(start_of_slicer_1, end_of_slicer_1), (start_of_slicer_2, end_of_slicer_2), ... ]. Defaults to None.

TYPE: Optional[List[Tuple[int]]] DEFAULT: None

Source code in src/pam/write/matrices.py 
 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
def write_od_matrices(
    population: Population,
    path: str,
    leg_filter: Optional[str] = None,
    person_filter: Optional[str] = None,
    time_minutes_filter: Optional[List[Tuple[int]]] = None,
) -> None:
    """Write a core population object to tabular O-D weighted matrices.

    Optionally segment matrices by leg attributes(mode/ purpose), person attributes or specific time periods.
    A single filter can be applied each time.

    TODO include freq (assume hh).

    Args:
        population (Population):
        path (str): directory to write OD matrix files
        leg_filter (Optional[str], optional): select between 'Mode', 'Purpose'. Defaults to None.
        person_filter (Optional[str], optional): select between given attribute categories (column names) from person attribute data. Defaults to None.
        time_minutes_filter (Optional[List[Tuple[int]]], optional): a list of tuples to slice times, e.g. [(start_of_slicer_1, end_of_slicer_1), (start_of_slicer_2, end_of_slicer_2), ... ]. Defaults to None.

    """
    create_local_dir(path)

    legs = []

    for hid, household in population.households.items():
        for pid, person in household.people.items():
            for leg in person.legs:
                data = {
                    "Household ID": hid,
                    "Person ID": pid,
                    "Origin": leg.start_location.area,
                    "Destination": leg.end_location.area,
                    "Purpose": leg.purp,
                    "Mode": leg.mode,
                    "Sequence": leg.seq,
                    "Start time": leg.start_time,
                    "End time": leg.end_time,
                    "Freq": household.freq,
                }
                if person_filter:
                    legs.append({**data, **person.attributes})
                else:
                    legs.append(data)

    df_total = pd.DataFrame(data=legs, columns=["Origin", "Destination"]).set_index("Origin")
    matrix = df_total.pivot_table(
        values="Destination", index="Origin", columns="Destination", fill_value=0, aggfunc=len
    )
    matrix.to_csv(os.path.join(path, "total_od.csv"))

    data_legs = pd.DataFrame(data=legs)

    if leg_filter:
        data_legs_grouped = data_legs.groupby(leg_filter)
        for filter, leg in data_legs_grouped:
            df = pd.DataFrame(data=leg, columns=["Origin", "Destination"]).set_index("Origin")
            matrix = df.pivot_table(
                values="Destination",
                index="Origin",
                columns="Destination",
                fill_value=0,
                aggfunc=len,
            )
            matrix.to_csv(os.path.join(path, filter + "_od.csv"))
        return None

    elif person_filter:
        data_legs_grouped = data_legs.groupby(person_filter)
        for filter, leg in data_legs_grouped:
            df = pd.DataFrame(data=leg, columns=["Origin", "Destination"]).set_index("Origin")
            matrix = df.pivot_table(
                values="Destination",
                index="Origin",
                columns="Destination",
                fill_value=0,
                aggfunc=len,
            )
            matrix.to_csv(os.path.join(path, filter + "_od.csv"))
        return None

    elif time_minutes_filter:
        periods = []
        for time in time_minutes_filter:
            periods.append(time)
        for start_time, end_time in periods:
            file_name = str(start_time) + "_to_" + str(end_time)
            start_time = mtdt(start_time)
            end_time = mtdt(end_time)
            data_time = data_legs[
                (data_legs["Start time"] >= start_time) & (data_legs["Start time"] < end_time)
            ]
            df = pd.DataFrame(data=data_time, columns=["Origin", "Destination"]).set_index("Origin")
            matrix = df.pivot_table(
                values="Destination",
                index="Origin",
                columns="Destination",
                fill_value=0,
                aggfunc=len,
            )
            matrix.to_csv(os.path.join(path, "time_" + file_name + "_od.csv"))
        return None