Add event hooks

Author: samakinen

Scripts/assignment/emme_assignment.py

@@ -1,5 +1,5 @@
 from __future__ import annotations
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, cast
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
 import pandas
 from math import log10
 
@@ -10,12 +10,14 @@
 import parameters.zone as zone_param
 from assignment.abstract_assignment import AssignmentModel
 from assignment.assignment_period import AssignmentPeriod
+from events.model_system_event_listener import EventHandler
 if TYPE_CHECKING:
     from assignment.emme_bindings.emme_project import EmmeProject
     from assignment.datatypes.transit_fare import TransitFareZoneSpecification
     from datahandling.resultdata import ResultsData
     from inro.emme.database.scenario import Scenario # type: ignore
     from inro.emme.network.Network import Network # type: ignore
+    import numpy
 
 
 class EmmeAssignmentModel(AssignmentModel):
@@ -46,6 +48,7 @@ class EmmeAssignmentModel(AssignmentModel):
     def __init__(self, 
                  emme_context: EmmeProject, 
                  first_scenario_id: int,
+                 event_handler: EventHandler=None,
                  separate_emme_scenarios: bool=False, 
                  save_matrices: bool=False,
                  time_periods: List[str]=param.time_periods, 
@@ -57,6 +60,7 @@ def __init__(self,
         self.emme_project = emme_context
         self.mod_scenario = self.emme_project.modeller.emmebank.scenario(
             first_scenario_id)
+        self._event_handler = event_handler
 
     def prepare_network(self, 
                         car_dist_unit_cost: Optional[float]=None):
@@ -112,7 +116,7 @@ def prepare_network(self,
         self.emme_project.create_extra_function_parameters(el1="@kaltevuus")
 
     def init_assign(self, 
-                    demand: Dict[str,List[numpy.ndarray]]):
+                    demand: Dict[str,List['numpy.ndarray']]):
         """??? types"""
         ap0 = self.assignment_periods[0]
         ap0.assign(demand, iteration="init")
@@ -253,8 +257,8 @@ def aggregate_results(self, resultdata: ResultsData):
 
     def calc_transit_cost(self, 
                           fares: TransitFareZoneSpecification, 
-                          peripheral_cost: numpy.ndarray, 
-                          default_cost: numpy.ndarray = None):
+                          peripheral_cost: 'numpy.ndarray', 
+                          default_cost: 'numpy.ndarray' = None):
         """Calculate transit zone cost matrix.
         
         Perform multiple transit assignments.
@@ -391,7 +395,6 @@ def _create_attributes(self,
         # Create link attributes
         ass_classes = list(param.emme_matrices) + ["bus"]
         ass_classes.remove("walk")
-        if TYPE_CHECKING: scenario = cast(Scenario, scenario)
         for ass_class in ass_classes:
             self.emme_project.create_extra_attribute(
                 "LINK", extra(ass_class), ass_class + " volume",

Scripts/helmet.py

@@ -8,6 +8,7 @@
 import utils.log as log
 from assignment.emme_assignment import EmmeAssignmentModel
 from assignment.mock_assignment import MockAssignmentModel
+from events.model_system_event_listener import EventHandler
 from modelsystem import ModelSystem, AgentModelSystem
 from datahandling.matrixdata import MatrixData
 
@@ -56,6 +57,11 @@ def main(args):
         estimation_data_path = Path(results_path) / args.scenario_name / 'estimation'
         estimation_data_path.mkdir(parents=True, exist_ok=True)
 
+    # Initialize event handler and load event listeners
+    event_handler = EventHandler()
+    # Load event listeners from 'events/examples' folder
+    event_handler.load_listeners(Path(__file__).parent / 'events' / 'examples')
+    
     # Choose and initialize the Traffic Assignment (supply)model
     if args.do_not_use_emme:
         log.info("Initializing MockAssignmentModel...")
@@ -76,6 +82,7 @@ def main(args):
         ass_model = EmmeAssignmentModel(
             EmmeProject(emme_project_path),
             first_scenario_id=args.first_scenario_id,
+            event_handler=event_handler,
             separate_emme_scenarios=args.separate_emme_scenarios,
             save_matrices=args.save_matrices,
             first_matrix_id=args.first_matrix_id)
@@ -90,7 +97,8 @@ def main(args):
     else:
         model = ModelSystem(
             forecast_zonedata_path, base_zonedata_path, base_matrices_path,
-            results_path, ass_model, args.scenario_name, estimation_data_path)
+            results_path, ass_model, args.scenario_name, event_handler,
+            estimation_data_path)
     log_extra["status"]["results"] = model.mode_share
 
     # Run traffic assignment simulation for N iterations,
@@ -128,7 +136,8 @@ def main(args):
             log_extra["status"]["converged"] = 1
         i += 1
 
-    if not log_extra["status"]["converged"]: log.warn("Model has not converged")
+    if not log_extra["status"]["converged"]:
+        log.warn("Model has not converged")
 
     # delete emme strategy files for scenarios
     if args.del_strat_files:

Scripts/modelsystem.py

@@ -26,6 +26,7 @@
 from datatypes.tour import Tour
 from transform.impedance_transformer import ImpedanceTransformer
 from models.linear import CarDensityModel
+from events.model_system_event_listener import EventHandler
 import parameters.assignment as param
 import parameters.zone as zone_param
 import parameters.tour_generation as gen_param
@@ -58,7 +59,9 @@ def __init__(self,
                  results_path: str, 
                  assignment_model: AssignmentModel, 
                  name: str,
+                 event_handler: EventHandler,
                  estimation_data_path: Path = None):
+        self.event_handler = event_handler
         self.ass_model = cast(Union[MockAssignmentModel,EmmeAssignmentModel], assignment_model) #type checker hint
         self.zone_numbers: numpy.array = self.ass_model.zone_numbers
         self.travel_modes: Dict[str, bool] = {}  # Dict instead of set, to preserve order
@@ -73,7 +76,8 @@ def __init__(self,
         if estimation_data_path:
             self.zdata_base.export_data(estimation_data_path / 'zonedata_base.csv')
             self.zdata_forecast.export_data(estimation_data_path / 'zonedata_forecast.csv')
-         
+        self.event_handler.on_zone_data_loaded(self.zdata_base, self.zdata_forecast)
+        
         # Output data
         self.resultmatrices = MatrixData(
             os.path.join(results_path, name, "Matrices"))
@@ -98,6 +102,7 @@ def __init__(self,
         self.convergence = []
         self.trucks = self.fm.calc_freight_traffic("truck")
         self.trailer_trucks = self.fm.calc_freight_traffic("trailer_truck")
+        self.event_handler.on_model_system_initialized(self)
 
     def _init_demand_model(self):
         return DemandModel(self.zdata_forecast, self.resultdata, is_agent_model=False)
@@ -127,6 +132,7 @@ def _add_internal_demand(self, previous_iter_impedance, is_last_iteration, estim
         # as logsums from probability calculation are used in tour generation.
         self.dm.create_population_segments()
         saved_pnr_impedance = {}
+        self.event_handler.on_population_segments_created(self.dm)
         for purpose in self.dm.tour_purposes:
             if isinstance(purpose, SecDestPurpose):
                 purpose.gen_model.init_tours()
@@ -143,6 +149,7 @@ def _add_internal_demand(self, previous_iter_impedance, is_last_iteration, estim
 
         # Tour generation
         self.dm.generate_tours()
+        self.event_handler.on_demand_model_tours_generated(self.dm)
 
         # Assigning of tours to mode, destination and time period
         for purpose in self.dm.tour_purposes:
@@ -172,6 +179,7 @@ def _add_internal_demand(self, previous_iter_impedance, is_last_iteration, estim
                                 break
                         log.debug(f"Park and ride demand calculation completed.")
 
+                self.event_handler.on_purpose_demand_calculated(purpose, demand)
                 if purpose.dest != "source":
                     for mode in demand:
                         self.dtm.add_demand(demand[mode])
@@ -249,9 +257,13 @@ def assign_base_demand(self,
             self._calculate_noise_areas()
             self.resultdata.flush()
         self.dtm.init_demand()
+        self.event_handler.on_base_demand_assigned(impedance)
         return impedance
 
-    def run_iteration(self, previous_iter_impedance, iteration=None, estimation_mode=False):
+    def run_iteration(self,
+                      previous_iter_impedance: Dict[str, Dict[str, numpy.ndarray]],
+                      iteration: Union[int, str] = None,
+                      estimation_mode=False):
         """Calculate demand and assign to network.
 
         Parameters
@@ -281,6 +293,7 @@ def run_iteration(self, previous_iter_impedance, iteration=None, estimation_mode
                 value : numpy.ndarray
                     Impedance (float 2-d matrix)
         """
+        self.event_handler.on_iteration_started(iteration, previous_iter_impedance)
         impedance = {}
 
         # Add truck and trailer truck demand, to time-period specific
@@ -292,9 +305,11 @@ def run_iteration(self, previous_iter_impedance, iteration=None, estimation_mode
         prediction = self.cdm.predict()
         self.zdata_forecast["car_density"] = prediction
         self.zdata_forecast["cars_per_1000"] = 1000 * prediction
+        self.event_handler.on_car_density_updated(iteration, prediction)
 
         # Calculate internal demand
         self._add_internal_demand(previous_iter_impedance, iteration=="last", estimation_mode)
+        self.event_handler.on_internal_demand_added(self.dtm)
 
         # Calculate external demand
         for mode in param.external_modes:
@@ -310,8 +325,9 @@ def run_iteration(self, previous_iter_impedance, iteration=None, estimation_mode
             else:
                 int_demand = self._sum_trips_per_zone(mode)
             ext_demand = self.em.calc_external(mode, int_demand)
+            self.event_handler.on_external_demand_calculated(ext_demand)
             self.dtm.add_demand(ext_demand)
-
+        
         # Calculate tour sums and mode shares
         tour_sum = {mode: self._sum_trips_per_zone(mode, include_dests=False)
             for mode in self.travel_modes}
@@ -343,11 +359,14 @@ def run_iteration(self, previous_iter_impedance, iteration=None, estimation_mode
                 if iteration=="last":
                     self._save_demand_to_omx(ap.name)
 
+        self.event_handler.on_demand_calculated(iteration, self.dtm)
+
         # Calculate and return traffic impedance
         for ap in self.ass_model.assignment_periods:
             tp = ap.name
             log.info("Assigning period " + tp)
             impedance[tp] = ap.assign(self.dtm.demand[tp], iteration)
+            self.event_handler.on_time_period_assigned(iteration, ap, impedance[tp])
             if tp == "aht":
                 self._update_ratios(impedance[tp], tp)
             if iteration=="last": # TODO: Get uncongested time from assignment results
@@ -373,6 +392,7 @@ def run_iteration(self, previous_iter_impedance, iteration=None, estimation_mode
         self.convergence.append(gap)
         self.resultdata._df_buffer["demand_convergence.txt"] = pandas.DataFrame(self.convergence)
         self.resultdata.flush()
+        self.event_handler.on_iteration_complete(iteration, impedance, gap)
         return impedance
 
     def _save_demand_to_omx(self, tp):

Scripts/tests/integration/test_models.py

@@ -2,6 +2,7 @@
 import numpy
 
 from datahandling.zonedata import ZoneData
+from events.model_system_event_listener import EventHandler
 import utils.log as log
 from modelsystem import ModelSystem, AgentModelSystem
 from assignment.mock_assignment import MockAssignmentModel
@@ -35,7 +36,8 @@ def test_models(self):
             TEST_DATA_PATH, "Base_input_data", "base_matrices")
         model = ModelSystem(
             zone_data_path, base_zone_data_path, base_matrices_path,
-            results_path, ass_model, "test")
+            results_path, ass_model, "test",
+            EventHandler())
         impedance = model.assign_base_demand()
         for ap in ass_model.assignment_periods:
             tp = ap.name
@@ -71,7 +73,7 @@ def test_agent_model(self):
             TEST_DATA_PATH, "Base_input_data", "base_matrices")
         model = AgentModelSystem(
             zone_data_path, base_zone_data_path, base_matrices_path,
-            results_path, ass_model, "test")
+            results_path, ass_model, "test", EventHandler())
         impedance = model.assign_base_demand()
         impedance = model.run_iteration(impedance)
         impedance = model.run_iteration(impedance, "last")