Merge branch 'olusanya' into refactor/emme_matrices

Scripts/assignment/abstract_assignment.py

@@ -1,20 +1,25 @@
+from __future__ import annotations
 from abc import ABCMeta, abstractmethod
+from typing import Any, Dict, List, Union
 
 
 class AssignmentModel:
     __metaclass__ = ABCMeta
 
+    @property
     @abstractmethod
-    def mapping(self):
+    def mapping(self) -> Dict[int,int]:
         """Dictionary of zone numbers and corresponding indices."""
         pass
 
+    @property
     @abstractmethod
-    def zone_numbers(self):
+    def zone_numbers(self) -> List[int]:
         pass
 
+    @property
     @abstractmethod
-    def nr_zones(self):
+    def nr_zones(self) -> int:
         pass
 
     @abstractmethod
@@ -41,5 +46,5 @@ class Period:
     __metaclass__ = ABCMeta
 
     @abstractmethod
-    def assign(self):
+    def assign(self, matrices: Dict[Any, Any], iteration: Union[int, str]) -> Dict[Any, Any]:
         pass

Scripts/assignment/assignment_period.py

@@ -1,13 +1,19 @@
-import numpy
+from __future__ import annotations
+import numpy # type: ignore
 import pandas
 
+from typing import TYPE_CHECKING, Any, Dict, Optional, Union
 import utils.log as log
 import parameters.assignment as param
 import parameters.zone as zone_param
 from assignment.datatypes.car_specification import CarSpecification
 from assignment.datatypes.transit import TransitSpecification
 from assignment.datatypes.path_analysis import PathAnalysis
 from assignment.abstract_assignment import Period
+if TYPE_CHECKING:
+    from assignment.emme_bindings.emme_project import EmmeProject
+    from assignment.datatypes.transit_fare import TransitFareZoneSpecification
+    from emme_context.modeller.emmebank import Scenario # type: ignore
 
 
 class AssignmentPeriod(Period):
@@ -31,24 +37,28 @@ class AssignmentPeriod(Period):
                 Assignment class (car_work/transit_leisure/...)
             value : dict
                 key : str
-                    Matrix type (demand/time/cost/dist/...)
+                    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, emme_scenario, emme_context,
-                 emme_matrices, separate_emme_scenarios=False):
+    def __init__(self, name: str, emme_scenario: int,
+                 emme_context: EmmeProject,
+                 emme_matrices: Dict[str, Dict[str, Any]],
+                 separate_emme_scenarios: bool = False):
         self.name = name
-        self.emme_scenario = emme_context.modeller.emmebank.scenario(
+        self.emme_scenario: Scenario = emme_context.modeller.emmebank.scenario(
             emme_scenario)
         self.emme_project = emme_context
         self._separate_emme_scenarios = separate_emme_scenarios
         self.emme_matrices = emme_matrices
         self.dist_unit_cost = param.dist_unit_cost
 
-    def extra(self, attr):
+    def extra(self, attr: str) -> str:
         """Add prefix "@" and time-period suffix.
 
         Parameters
@@ -63,7 +73,7 @@ def extra(self, attr):
         """
         return "@{}_{}".format(attr, self.name)
 
-    def prepare(self, segment_results):
+    def prepare(self, segment_results: Dict[str,Dict[str,str]]):
         """Prepare network for assignment.
 
         Calculate road toll cost, set boarding penalties,
@@ -86,7 +96,7 @@ def prepare(self, segment_results):
         self._calc_background_traffic()
         self._specify()
 
-    def assign(self, matrices, iteration):
+    def assign(self, matrices: dict, iteration: Union[int,str]) -> Dict:
         """Assign cars, bikes and transit for one time period.
 
         Get travel impedance matrices for one time period from assignment.
@@ -173,7 +183,10 @@ def assign(self, matrices, iteration):
                 mtxs["cost"][ass_cl] += self.dist_unit_cost * mtxs["dist"][ass_cl]
         return mtxs
 
-    def calc_transit_cost(self, fares, peripheral_cost, mapping):
+    def calc_transit_cost(self, 
+                          fares: TransitFareZoneSpecification, 
+                          peripheral_cost: numpy.ndarray, 
+                          mapping: dict):
         """Calculate transit zone cost matrix.
         
         Perform multiple transit assignments.
@@ -385,7 +398,9 @@ def _set_bike_vdfs(self):
                 link.modes -= {main_mode}
         self.emme_scenario.publish_network(network)
 
-    def _set_emmebank_matrices(self, matrices, is_last_iteration):
+    def _set_emmebank_matrices(self, 
+                               matrices: Dict[str,numpy.ndarray], 
+                               is_last_iteration: bool):
         """Set matrices in emmebank.
 
         Bike matrices are added together, so that only one matrix is to be
@@ -416,7 +431,10 @@ def _set_emmebank_matrices(self, matrices, is_last_iteration):
             else:
                 self._set_matrix(mtx, matrices[mtx])
 
-    def _set_matrix(self, ass_class, matrix, matrix_type="demand"):
+    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(ass_class)
                    + "would cause infinite loop in Emme assignment.")
@@ -427,7 +445,9 @@ def _set_matrix(self, ass_class, matrix, matrix_type="demand"):
                 self.emme_matrices[ass_class][matrix_type]).set_numpy_data(
                     matrix, scenario_id=self.emme_scenario.id)
 
-    def _get_matrices(self, mtx_type, is_last_iteration=False):
+    def _get_matrices(self, 
+                      mtx_type: str, 
+                      is_last_iteration: bool=False) -> Dict[str,numpy.ndarray]:
         """Get all matrices of specified type.
 
         Parameters
@@ -460,7 +480,9 @@ def _get_matrices(self, mtx_type, is_last_iteration=False):
             matrices["transit_leisure"] = matrices["transit_work"]
         return matrices
 
-    def _get_matrix(self, ass_class, matrix_type):
+    def _get_matrix(self, 
+                    ass_class: str, 
+                    matrix_type: str) -> numpy.ndarray:
         """Get matrix with type pair (e.g., demand, car_work).
 
         Parameters
@@ -479,14 +501,14 @@ def _get_matrix(self, 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):
+    def _damp_travel_time(self, demand_type: str):
         """Reduce the impact from first waiting time on total travel 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)
 
-    def _extract_timecost_from_gcost(self, ass_class):
+    def _extract_timecost_from_gcost(self, ass_class: str):
         """Remove monetary cost from generalized cost.
 
         Traffic assignment produces a generalized cost matrix.
@@ -504,7 +526,7 @@ def _extract_timecost_from_gcost(self, ass_class):
         self._set_matrix(ass_class, time, "time")
         return time
 
-    def _calc_background_traffic(self, include_trucks=False):
+    def _calc_background_traffic(self, include_trucks: bool=False):
         """Calculate background traffic (buses)."""
         network = self.emme_scenario.get_network()
         # emme api has name "data3" for ul3
@@ -541,7 +563,9 @@ def _calc_road_cost(self):
             link[self.extra("total_cost")] = toll_cost + dist_cost
         self.emme_scenario.publish_network(network)
 
-    def _calc_boarding_penalties(self, extra_penalty=0, is_last_iteration=False):
+    def _calc_boarding_penalties(self, 
+                                 extra_penalty: int = 0, 
+                                 is_last_iteration: bool = False):
         """Calculate boarding penalties for transit assignment."""
         # Definition of line specific boarding penalties
         network = self.emme_scenario.get_network()
@@ -557,8 +581,8 @@ def _calc_boarding_penalties(self, extra_penalty=0, is_last_iteration=False):
             except KeyError:
                 missing_penalties.add(line.mode.id)
         if missing_penalties:
-            missing_penalties = ", ".join(missing_penalties)
-            log.warn("No boarding penalty found for transit modes " + missing_penalties)
+            missing_penalties_str: str = ", ".join(missing_penalties)
+            log.warn("No boarding penalty found for transit modes " + missing_penalties_str)
         self.emme_scenario.publish_network(network)
 
     def _specify(self):
@@ -633,7 +657,9 @@ def _specify(self):
             },
         }
 
-    def _assign_cars(self, stopping_criteria, lightweight=False):
+    def _assign_cars(self, 
+                     stopping_criteria: Dict[str, Union[int, float]], 
+                     lightweight: bool=False):
         """Perform car_work traffic assignment for one scenario."""
         log.info("Car assignment started...")
         car_spec = self._car_spec.spec(lightweight)
@@ -655,8 +681,10 @@ def _assign_cars(self, stopping_criteria, lightweight=False):
         if assign_report["stopping_criterion"] == "MAX_ITERATIONS":
             log.warn("Car assignment not fully converged.")
 
-    def _assign_bikes(self, length_mat_id, length_for_links):
-        """Perform bike traffic assignment for one scenario."""
+    def _assign_bikes(self, 
+                      length_mat_id: Union[float, int, str], 
+                      length_for_links: str):
+        """Perform bike traffic assignment for one scenario.???TYPES"""
         scen = self.emme_scenario
         spec = self.bike_spec
         spec["classes"][0]["results"]["link_volumes"] = self.extra("bike")

Scripts/assignment/datatypes/car.py

@@ -1,5 +1,8 @@
+from __future__ import annotations
 import parameters.assignment as param
 from assignment.datatypes.path_analysis import PathAnalysis
+from collections.abc import Callable
+from typing import Any, Dict, Optional, Union
 
 
 class Car:
@@ -22,11 +25,15 @@ class Car:
     value_of_time_inv : float (optional)
         Inversed value of time [min/eur], default is param.vot_inv
     """
-    def __init__(self, ass_class, extra, emme_matrices,
-                link_costs, value_of_time_inv=None):
+    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):
         if value_of_time_inv is None:
             value_of_time_inv = param.vot_inv[param.vot_classes[ass_class]]
-        self.spec = {
+        self.spec: Dict[str, Any] = {
             "mode": param.assignment_modes[ass_class],
             "demand": emme_matrices["demand"],
             "generalized_cost": {
@@ -45,6 +52,8 @@ def __init__(self, ass_class, extra, emme_matrices,
         if ass_class not in ("trailer_truck", "truck"):
             self.add_analysis(extra("toll_cost"), emme_matrices["cost"])
 
-    def add_analysis (self, link_component, od_values):
+    def add_analysis (self, 
+                      link_component: str, 
+                      od_values: Union[int, str]):
         analysis = PathAnalysis(link_component, od_values)
         self.spec["path_analyses"].append(analysis.spec)

Scripts/assignment/datatypes/car_specification.py

@@ -1,5 +1,8 @@
+from __future__ import annotations
+from typing import Any, Dict, List, Union
 import parameters.assignment as param
 from assignment.datatypes.car import Car
+from collections.abc import Callable
 
 class CarSpecification:
     """
@@ -19,7 +22,9 @@ class CarSpecification:
             value : str
                 Emme matrix id
     """
-    def __init__(self, extra, emme_matrices):
+    def __init__(self,
+                 extra: Callable, 
+                 emme_matrices: Dict[str, Union[str, Dict[str, str]]]):
         self._modes = {}
         self._freight_modes = list(param.freight_dist_unit_cost)
         for mode in param.assignment_modes:
@@ -31,7 +36,7 @@ def __init__(self, extra, emme_matrices):
             else:
                 kwargs = {"link_costs": extra("total_cost")}
             self._modes[mode] = Car(mode, extra, emme_matrices[mode], **kwargs)
-        self._spec = {
+        self._spec: Dict[str, Any] = {
             "type": "SOLA_TRAFFIC_ASSIGNMENT",
             "background_traffic": {
                 "link_component": param.background_traffic_attr,
@@ -41,7 +46,7 @@ def __init__(self, extra, emme_matrices):
             "stopping_criteria": None, # This is defined later
         }
 
-    def spec(self, lightweight=False):
+    def spec(self, lightweight: bool = False) -> Dict[str, Any]:
         self._spec["classes"] = [self._modes[mode].spec for mode in self._modes
             if not lightweight or mode not in self._freight_modes]
         return self._spec

Scripts/assignment/datatypes/journey_level.py

@@ -1,16 +1,21 @@
+from __future__ import annotations
+from typing import Any, Dict
 import parameters.assignment as param
 
 
 class JourneyLevel:
-    def __init__(self, headway_attribute, boarded, count_zone_boardings=False):
+    def __init__(self, 
+                 headway_attribute: str, 
+                 boarded: bool, 
+                 count_zone_boardings: bool = False):
         # Definition of transition rules: all modes are allowed
         transitions = []
         for mode in param.transit_modes:
             transitions.append({
                 "mode": mode,
                 "next_journey_level": 1
             })
-        self.spec = {
+        self.spec: Dict[str, Any] = {
             "transition_rules": transitions,
             "boarding_time": None,
             "boarding_cost": dict.fromkeys([

Scripts/assignment/datatypes/path_analysis.py

@@ -1,5 +1,8 @@
+from typing import Optional, Union
+
+
 class PathAnalysis:
-    def __init__(self, link_component, od_values=None):
+    def __init__(self, link_component:str, od_values:Optional[Union[str,int]]=None):
         self.spec = {
             "link_component": link_component,
             "operator": "+",

Scripts/assignment/datatypes/transit.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+from typing import Any, Dict, Union
 import parameters.assignment as param
 from assignment.datatypes.journey_level import JourneyLevel
 
@@ -28,14 +30,17 @@ class TransitSpecification:
     count_zone_boardings : bool (optional)
         Whether assignment is performed only to count fare zone boardings
     """
-    def __init__(self, segment_results, headway_attribute,
-                 emme_matrices, count_zone_boardings=False):
+    def __init__(self, 
+                 segment_results: Dict[str,str], 
+                 headway_attribute: str,
+                 emme_matrices: Dict[str, Union[str, Dict[str, str]]], 
+                 count_zone_boardings: bool = False):
         no_penalty = dict.fromkeys(["at_nodes", "on_lines", "on_segments"])
         no_penalty["global"] = {
             "penalty": 0, 
             "perception_factor": 1,
         }
-        self.transit_spec = {
+        self.transit_spec: Dict[str, Any] = {
             "type": "EXTENDED_TRANSIT_ASSIGNMENT",
             "modes": param.transit_assignment_modes,
             "demand": emme_matrices["demand"],