Make handling of EMME matrices less error-prone and more flexible

Scripts/assignment/assignment_period.py

@@ -1,8 +1,6 @@
 from __future__ import annotations
-import os
 import numpy # type: ignore
 import pandas
-import copy
 
 from typing import TYPE_CHECKING, Any, Dict, Optional, Union
 import utils.log as log
@@ -19,24 +17,45 @@
 
 
 class AssignmentPeriod(Period):
-    def __init__(self, name: str, emme_scenario: int, 
+    """
+    EMME assignment period definition.
+
+    This typically represents an hour of the day, which may or may not
+    have a dedicated EMME scenario. In case it does not have its own
+    EMME scenario, assignment results are stored only in extra attributes.
+
+    Parameters
+    ----------
+    name : str
+        Time period name (aht/pt/iht)
+    emme_scenario : int
+        EMME scenario linked to the time period
+    emme_context : assignment.emme_bindings.emme_project.EmmeProject
+        Emme project to connect to this assignment
+    emme_matrices : dict
+        key : str
+                Assignment class (car_work/transit_leisure/...)
+            value : dict
+                key : str
+                    Matrix type (demand/time/c
+
+                    t/dist/...)
+                value : str
+                    EMME matrix id
+    separate_emme_scenarios : bool (optional)
+        Whether separate scenarios have been created in EMME
+        for storing time-period specific network results.
+    """
+    def __init__(self, name: str, emme_scenario: int,
                  emme_context: EmmeProject,
-                 demand_mtx: Dict[str, Dict[str, Any]] = param.emme_demand_mtx,
-                 result_mtx: Dict[str, Dict[str, Dict[str, Any]]] = param.emme_result_mtx, 
-                 save_matrices: bool = False,
+                 emme_matrices: Dict[str, Dict[str, Any]],
                  separate_emme_scenarios: bool = False):
         self.name = name
         self.emme_scenario: Scenario = emme_context.modeller.emmebank.scenario(
             emme_scenario)
         self.emme_project = emme_context
         self._separate_emme_scenarios = separate_emme_scenarios
-        if save_matrices:
-            self.demand_mtx = copy.deepcopy(demand_mtx)
-            self.result_mtx = copy.deepcopy(result_mtx)
-        else:
-            # Refer to the same matrices for all time periods
-            self.demand_mtx = demand_mtx
-            self.result_mtx = result_mtx
+        self.emme_matrices = emme_matrices
         self.dist_unit_cost = param.dist_unit_cost
 
     def extra(self, attr: str) -> str:
@@ -99,7 +118,7 @@ def assign(self, matrices: dict, iteration: Union[int,str]) -> Dict:
         if iteration=="init":
             self._assign_pedestrians()
             self._set_bike_vdfs()
-            self._assign_bikes(self.result_mtx["dist"]["bike"]["id"], "all")
+            self._assign_bikes(self.emme_matrices["bike"]["dist"], "all")
             self._set_car_and_transit_vdfs()
             if not self._separate_emme_scenarios:
                 self._calc_background_traffic()
@@ -131,7 +150,7 @@ def assign(self, matrices: dict, iteration: Union[int,str]) -> Dict:
             self._assign_transit()
         elif iteration=="last":
             self._set_bike_vdfs()
-            self._assign_bikes(self.result_mtx["dist"]["bike"]["id"], "all")
+            self._assign_bikes(self.emme_matrices["bike"]["dist"], "all")
             self._set_car_and_transit_vdfs()
             self._calc_background_traffic()
             self._assign_cars(param.stopping_criteria_fine)
@@ -197,11 +216,7 @@ def calc_transit_cost(self,
         tc = "transit_work"
         spec = TransitSpecification(
             self._segment_results[tc], self.extra("hw"),
-            self.demand_mtx[tc]["id"],
-            self.result_mtx["time"][tc]["id"],
-            self.result_mtx["dist"][tc]["id"],
-            self.result_mtx["trip_part_"+tc],
-            count_zone_boardings=True)
+            self.emme_matrices[tc], count_zone_boardings=True)
         is_in_transit_zone_attr = param.is_in_transit_zone_attr.replace(
             "ui", "data")
         for transit_zone in transit_zones:
@@ -216,7 +231,7 @@ def calc_transit_cost(self,
             self.emme_project.matrix_results(
                 spec.transit_result_spec, self.emme_scenario)
             nr_visits = self._get_matrix(
-                "trip_part_transit_work", "board_cost")
+                tc, "actual_total_boarding_costs")
             # If the number of visits is less than 1, there seems to
             # be an easy way to avoid visiting this transit zone
             has_visited[transit_zone] = (nr_visits > 0.99)
@@ -254,7 +269,7 @@ def calc_transit_cost(self,
         cost[l:u, :u] = peripheral_cost
         cost[:u, l:u] = peripheral_cost.T
         # Calculate distance-based cost from inv-distance
-        dist = self._get_matrix("dist", "transit_work")
+        dist = self._get_matrix(tc, "dist")
         dist_cost = fares.start_fare + fares.dist_fare*dist
         cost[cost>=maxfare] = dist_cost[cost>=maxfare]
         # Reset boarding penalties
@@ -416,22 +431,18 @@ def _set_emmebank_matrices(self,
             else:
                 self._set_matrix(mtx, matrices[mtx])
 
-    def _set_matrix(self, 
-                    mtx_label: str, 
-                    matrix: numpy.ndarray, 
-                    result_type: Optional[str] = None):
+    def _set_matrix(self,
+                    ass_class: str,
+                    matrix: numpy.ndarray,
+                    matrix_type: Optional[str] = "demand"):
         if numpy.isnan(matrix).any():
-            msg = ("NAs in demand matrix {}. ".format(mtx_label)
-                   + "Would cause infinite loop in Emme assignment.")
+            msg = ("NAs in demand matrix {} ".format(ass_class)
+                   + "would cause infinite loop in Emme assignment.")
             log.error(msg)
             raise ValueError(msg)
-        elif result_type is None:
-            self.emme_project.modeller.emmebank.matrix(
-                self.demand_mtx[mtx_label]["id"]).set_numpy_data(
-                    matrix, scenario_id=self.emme_scenario.id)
         else:
             self.emme_project.modeller.emmebank.matrix(
-                self.result_mtx[result_type][mtx_label]["id"]).set_numpy_data(
+                self.emme_matrices[ass_class][matrix_type]).set_numpy_data(
                     matrix, scenario_id=self.emme_scenario.id)
 
     def _get_matrices(self, 
@@ -453,46 +464,47 @@ def _get_matrices(self,
             Subtype (car_work/truck/inv_time/...) : numpy 2-d matrix
                 Matrix of the specified type
         """
-        last_iteration_classes = param.freight_classes + ("transit_leisure",)
+        last_iter_classes = param.freight_classes + ("transit_leisure",)
         matrices = {}
-        for subtype in self.result_mtx[mtx_type]:
-            if is_last_iteration or subtype not in last_iteration_classes:
-                if mtx_type == "time" and subtype in param.assignment_modes:
-                    mtx = self._extract_timecost_from_gcost(subtype)
-                elif mtx_type == "time" and subtype in param.transit_classes:
-                    mtx = self._damp_travel_time(subtype)
+        for ass_class, mtx_types in self.emme_matrices.items():
+            if (mtx_type in mtx_types and
+                    (is_last_iteration or ass_class not in last_iter_classes)):
+                if mtx_type == "time" and ass_class in param.assignment_modes:
+                    mtx = self._extract_timecost_from_gcost(ass_class)
+                elif mtx_type == "time" and ass_class in param.transit_classes:
+                    mtx = self._damp_travel_time(ass_class)
                 else:
-                    mtx = self._get_matrix(mtx_type, subtype)
-                matrices[subtype] = mtx
+                    mtx = self._get_matrix(ass_class, mtx_type)
+                matrices[ass_class] = mtx
         if not is_last_iteration:
             matrices["transit_leisure"] = matrices["transit_work"]
         return matrices
 
     def _get_matrix(self, 
-                    assignment_result_type: str, 
-                    subtype: str) -> numpy.ndarray:
+                    ass_class: str, 
+                    matrix_type: str) -> numpy.ndarray:
         """Get matrix with type pair (e.g., demand, car_work).
 
         Parameters
         ----------
-        assignment_result_type : str
-            Type (demand/time/transit/...)
-        subtype : str
-            Subtype (car_work/truck/inv_time/...)
+        ass_class : str
+            Assignment class (car_work/transit_leisure/truck/...)
+        matrix_type : str
+            Type (demand/time/cost/...)
 
         Return
         ------
         numpy 2-d matrix
             Matrix of the specified type
         """
-        emme_id = self.result_mtx[assignment_result_type][subtype]["id"]
+        emme_id = self.emme_matrices[ass_class][matrix_type]
         return (self.emme_project.modeller.emmebank.matrix(emme_id)
                 .get_numpy_data(scenario_id=self.emme_scenario.id))
 
     def _damp_travel_time(self, demand_type: str):
         """Reduce the impact from first waiting time on total travel time."""
-        travel_time = self._get_matrix("time", demand_type)
-        fw_time = self._get_matrix("trip_part_"+demand_type, "fw_time")
+        travel_time = self._get_matrix(demand_type, "time")
+        fw_time = self._get_matrix(demand_type, "actual_first_waiting_times")
         wt_weight = param.waiting_time_perception_factor
         return travel_time + wt_weight*((5./3.*fw_time)**0.8 - fw_time)
 
@@ -503,9 +515,9 @@ def _extract_timecost_from_gcost(self, ass_class: str):
         To get travel time, monetary cost is removed from generalized cost.
         """
         vot_inv = param.vot_inv[param.vot_classes[ass_class]]
-        gcost = self._get_matrix("gen_cost", ass_class)
-        cost = self._get_matrix("cost", ass_class)
-        dist = self._get_matrix("dist", ass_class)
+        gcost = self._get_matrix(ass_class, "gen_cost")
+        cost = self._get_matrix(ass_class, "cost")
+        dist = self._get_matrix(ass_class, "dist")
         if ass_class in ("trailer_truck", "truck"):
             # toll costs are not applied to freight
             time = gcost - vot_inv*param.freight_dist_unit_cost[ass_class]*dist
@@ -574,24 +586,20 @@ def _calc_boarding_penalties(self,
         self.emme_scenario.publish_network(network)
 
     def _specify(self):
-        self._car_spec = CarSpecification(
-            self.extra, self.demand_mtx, self.result_mtx)
+        self._car_spec = CarSpecification(self.extra, self.emme_matrices)
         self._transit_specs = {tc: TransitSpecification(
                 self._segment_results[tc], self.extra("hw"),
-                self.demand_mtx[tc]["id"],
-                self.result_mtx["time"][tc]["id"],
-                self.result_mtx["dist"][tc]["id"],
-                self.result_mtx["trip_part_"+tc])
+                self.emme_matrices[tc])
             for tc in param.transit_classes}
         self.bike_spec = {
             "type": "STANDARD_TRAFFIC_ASSIGNMENT",
             "classes": [
                 {
                     "mode": param.main_mode,
-                    "demand": self.demand_mtx["bike"]["id"],
+                    "demand": self.emme_matrices["bike"]["demand"],
                     "results": {
                         "od_travel_times": {
-                            "shortest_paths": self.result_mtx["time"]["bike"]["id"],
+                            "shortest_paths": self.emme_matrices["bike"]["time"],
                         },
                         "link_volumes": None, # This is defined later
                     },
@@ -614,7 +622,7 @@ def _specify(self):
         self.walk_spec = {
             "type": "STANDARD_TRANSIT_ASSIGNMENT",
             "modes": param.aux_modes,
-            "demand": self.demand_mtx["bike"]["id"],
+            "demand": self.emme_matrices["bike"]["demand"],
             "waiting_time": {
                 "headway_fraction": 0.01,
                 "effective_headways": "hdw",
@@ -628,7 +636,7 @@ def _specify(self):
                 "perception_factor": 1,
             },
             "od_results": {
-                "transit_times": self.result_mtx["time"]["walk"]["id"],
+                "transit_times": self.emme_matrices["walk"]["time"],
             },
             "strategy_analysis": {
                 "sub_path_combination_operator": "+",
@@ -644,7 +652,7 @@ def _specify(self):
                     },
                 },
                 "results": {
-                    "od_values": self.result_mtx["dist"]["walk"]["id"],
+                    "od_values": self.emme_matrices["walk"]["dist"],
                 },
             },
         }

Scripts/assignment/datatypes/car.py

@@ -6,34 +6,51 @@
 
 
 class Car:
-    def __init__(self, 
-                 ass_class: str, 
-                 extra: Callable, 
-                 demand_mtx: Dict[str, Dict[str, Dict[str, Union[int, str]]]], 
-                 result_mtx: Dict[str, Dict[str, Dict[str, Union[int, str]]]],
-                 link_costs: str, 
+    """Car assignment class definition.
+
+    Parameters
+    ----------
+    ass_class : str
+        Assignment class (car_work/car_leisure/van/truck/trailer_truck)
+    extra : assignment_period.AssignmentPeriod.extra()
+        Function for generating extra attribute name
+        for specific assignment period
+    emme_matrices : dict
+        key : str
+            Impedance type (time/cost/dist/...)
+        value : str
+            Emme matrix id
+    link_costs : str
+        Link attribute where link cost is found
+    value_of_time_inv : float (optional)
+        Inversed value of time [min/eur], default is param.vot_inv
+    """
+    def __init__(self,
+                 ass_class: str,
+                 extra: Callable,
+                 emme_matrices: Dict[str, Union[str, Dict[str, str]]],
+                 link_costs: str,
                  value_of_time_inv: Optional[float]=None):
-        od_travel_times = result_mtx["gen_cost"][ass_class]["id"]
         if value_of_time_inv is None:
             value_of_time_inv = param.vot_inv[param.vot_classes[ass_class]]
         self.spec: Dict[str, Any] = {
             "mode": param.assignment_modes[ass_class],
-            "demand": demand_mtx[ass_class]["id"],
+            "demand": emme_matrices["demand"],
             "generalized_cost": {
                 "link_costs": link_costs,
                 "perception_factor": value_of_time_inv,
             },
             "results": {
                 "link_volumes": extra(ass_class),
                 "od_travel_times": {
-                    "shortest_paths": od_travel_times
+                    "shortest_paths": emme_matrices["gen_cost"]
                 }
             },
             "path_analyses": []
         }
-        self.add_analysis("length", result_mtx["dist"][ass_class]["id"])
+        self.add_analysis("length", emme_matrices["dist"])
         if ass_class not in ("trailer_truck", "truck"):
-            self.add_analysis(extra("toll_cost"), result_mtx["cost"][ass_class]["id"])
+            self.add_analysis(extra("toll_cost"), emme_matrices["cost"])
 
     def add_analysis (self, 
                       link_component: str,