diff --git a/tm2py/components/network/create_tod_scenarios.py b/tm2py/components/network/create_tod_scenarios.py index 1df6284e..7732d013 100644 --- a/tm2py/components/network/create_tod_scenarios.py +++ b/tm2py/components/network/create_tod_scenarios.py @@ -109,13 +109,15 @@ def _create_highway_scenarios(self): emmebank.extra_function_parameters.el2 = "@capacity" emmebank.extra_function_parameters.el3 = "@ja" emmebank.extra_function_parameters.el4 = "@static_rel" + # get() and put() did not work for los reliability + # remove them from the reliability tmplt reliability_tmplt = ( "* (1 + el4 + " - "( {factor[LOS_C]} * ( put(get(1).min.1.5) - {threshold[LOS_C]} + 0.01 ) ) * (get(1) .gt. {threshold[LOS_C]})" - "+ ( {factor[LOS_D]} * ( get(2) - {threshold[LOS_D]} + 0.01 ) ) * (get(1) .gt. {threshold[LOS_D]})" - "+ ( {factor[LOS_E]} * ( get(2) - {threshold[LOS_E]} + 0.01 ) ) * (get(1) .gt. {threshold[LOS_E]})" - "+ ( {factor[LOS_FL]} * ( get(2) - {threshold[LOS_FL]} + 0.01 ) ) * (get(1) .gt. {threshold[LOS_FL]})" - "+ ( {factor[LOS_FH]} * ( get(2) - {threshold[LOS_FH]} + 0.01 ) ) * (get(1) .gt. {threshold[LOS_FH]})" + "( {factor[LOS_C]} * ( ((volau + volad)/el2).min.1.5 - {threshold[LOS_C]} + 0.01 ) ) * (((volau + volad)/el2) .gt. {threshold[LOS_C]})" + "+ ( {factor[LOS_D]} * ( ((volau + volad)/el2).min.1.5 - {threshold[LOS_D]} + 0.01 ) ) * (((volau + volad)/el2) .gt. {threshold[LOS_D]})" + "+ ( {factor[LOS_E]} * ( ((volau + volad)/el2).min.1.5 - {threshold[LOS_E]} + 0.01 ) ) * (((volau + volad)/el2) .gt. {threshold[LOS_E]})" + "+ ( {factor[LOS_FL]} * ( ((volau + volad)/el2).min.1.5 - {threshold[LOS_FL]} + 0.01 ) ) * (((volau + volad)/el2) .gt. {threshold[LOS_FL]})" + "+ ( {factor[LOS_FH]} * ( ((volau + volad)/el2).min.1.5 - {threshold[LOS_FH]} + 0.01 ) ) * (((volau + volad)/el2) .gt. {threshold[LOS_FH]})" ")" ) parameters = { @@ -152,17 +154,24 @@ def _create_highway_scenarios(self): }, }, } - # TODO: should have just 3 functions, and map the FT to the vdf - # TODO: could optimize expression (to review) - bpr_tmplt = "el1 * (1 + 0.20 * ((volau + volad)/el2/0.75)^6)" - # "el1 * (1 + 0.20 * put(put((volau + volad)/el2/0.75))*get(1))*get(2)*get(2)" + # rewrite bpr_tmplt to use put() and get() for nested functions + # keeping the original for reference + # bpr_tmplt = "el1 * (1 + 0.20 * ((volau + volad)/el2/0.75)^6)" + bpr_tmplt = "el1 * (1 + 0.20 * (put((volau + volad)/el2)/0.75) ** 6)" + fixed_tmplt = "el1" + + # rewrite akcelik_tmplt to use put() and get() for nested functions + # keeping the original for reference + # akcelik_tmplt = ( + # "(el1 + 60 * (0.25 *((volau + volad)/el2 - 1 + " + # "(((volau + volad)/el2 - 1)^2 + el3 * (volau + volad)/el2)^0.5)))" + # ) akcelik_tmplt = ( - "(el1 + 60 * (0.25 *((volau + volad)/el2 - 1 + " - "(((volau + volad)/el2 - 1)^2 + el3 * (volau + volad)/el2)^0.5)))" - # "(el1 + 60 * (0.25 *(put(put((volau + volad)/el2) - 1) + " - # "(((get(2)*get(2) + (16 * el3 * get(1)^0.5))))" + "(el1 + 60 * (0.25 * (put((volau + volad)/el2) - 1 + " + "((get(1) - 1) ** 2 + el3 * get(1)) ** 0.5)))" ) + for f_id in ["fd1", "fd2"]: if emmebank.function(f_id): emmebank.delete_function(f_id) diff --git a/tm2py/components/network/highway/highway_assign.py b/tm2py/components/network/highway/highway_assign.py index fa2091c2..1dbd9f9a 100644 --- a/tm2py/components/network/highway/highway_assign.py +++ b/tm2py/components/network/highway/highway_assign.py @@ -140,13 +140,17 @@ def run(self): else: demand.run() + calculate_reliability = self.config.reliability + for time in self.time_period_names: scenario = self.highway_emmebank.scenario(time) with self._setup(scenario, time): iteration = self.controller.iteration warmstart = self.controller.config.warmstart.warmstart assign_classes = [ - AssignmentClass(c, time, iteration, warmstart) + AssignmentClass( + c, time, iteration, calculate_reliability, warmstart + ) for c in self.config.classes ] if iteration > 0: @@ -154,51 +158,59 @@ def run(self): else: self._reset_background_traffic(scenario) self._create_skim_matrices(scenario, assign_classes) - assign_spec = self._get_assignment_spec( - assign_classes, path_analysis=False - ) - # self.logger.log_dict(assign_spec, level="DEBUG") - with self.logger.log_start_end( - "Run SOLA assignment with path analyses", level="INFO" - ): - assign = self.controller.emme_manager.tool( - "inro.emme.traffic_assignment.sola_traffic_assignment" + # calculate highway reliability in global iteration 0 and 1 only + # this requires the assignment to be run twice + if (iteration <= 1) & (calculate_reliability): + # set path analysis to False to avoid skimming + assign_spec = self._get_assignment_spec( + assign_classes, path_analysis=False ) - assign(assign_spec, scenario, chart_log_interval=1) - # calucaltes link level LOS based reliability - net_calc = NetworkCalculator(self.controller, scenario) - - exf_pars = scenario.emmebank.extra_function_parameters - vdfs = [ - f for f in scenario.emmebank.functions() if f.type == "VOLUME_DELAY" - ] - for function in vdfs: - expression = function.expression - for el in ["el1", "el2", "el3", "el4"]: - expression = expression.replace(el, getattr(exf_pars, el)) - if "@static_rel" in expression: - # split function into time component and reliability component - time_expr, reliability_expr = expression.split( - "*(1+@static_rel+" + with self.logger.log_start_end( + "Run SOLA assignment without path analyses", level="INFO" + ): + assign = self.controller.emme_manager.tool( + "inro.emme.traffic_assignment.sola_traffic_assignment" ) - net_calc( - "@auto_time", - time_expr, - {"link": "vdf=%s" % function.id[2:]}, - ) - net_calc( - "@reliability", - "(@static_rel+" + reliability_expr, - {"link": "vdf=%s" % function.id[2:]}, - ) - net_calc("@reliability_sq", "@reliability**2", {"link": "all"}) + assign(assign_spec, scenario, chart_log_interval=1) + + # calucaltes link level LOS based reliability + net_calc = NetworkCalculator(self.controller, scenario) + + exf_pars = scenario.emmebank.extra_function_parameters + vdfs = [ + f + for f in scenario.emmebank.functions() + if f.type == "VOLUME_DELAY" + ] + for function in vdfs: + expression = function.expression + for el in ["el1", "el2", "el3", "el4"]: + expression = expression.replace(el, getattr(exf_pars, el)) + if "@static_rel" in expression: + # split function into time component and reliability component + time_expr, reliability_expr = expression.split( + "*(1+@static_rel+" + ) + net_calc( + "@auto_time", + time_expr, + {"link": "vdf=%s" % function.id[2:]}, + ) + net_calc( + "@reliability", + "(@static_rel+" + reliability_expr, + {"link": "vdf=%s" % function.id[2:]}, + ) + net_calc( + "@reliability_sq", "@reliability**2", {"link": "all"} + ) assign_spec = self._get_assignment_spec( assign_classes, path_analysis=True ) with self.logger.log_start_end( - "Run SOLA assignment with path analyses and highway reliability", + "Run SOLA assignment with path analyses", level="INFO", ): assign = self.controller.emme_manager.tool( @@ -289,6 +301,14 @@ def _create_skim_matrices( self.logger.debug( f"Create matrix name: {matrix_name}, id: {matrix.id}" ) + # if not skimming reliability, set reliability matrices to 0 + if not self.config.reliability: + if ("rlbty" in matrix_name) | ("autotime" in matrix_name): + data = self._matrix_cache.get_data(matrix_name) + # NOTE: sets values for external zones as well + data = 0 * data + self._matrix_cache.set_data(matrix_name, data) + self._skim_matrices.append(matrix) def _get_assignment_spec( @@ -430,18 +450,22 @@ def _log_debug_report(self, scenario: EmmeScenario, time_period: str): class AssignmentClass: """Highway assignment class, represents data from config and conversion to Emme specs.""" - def __init__(self, class_config, time_period, iteration, warmstart): + def __init__(self, class_config, time_period, iteration, reliability, warmstart): """Constructor of Highway Assignment class. Args: class_config (_type_): _description_ time_period (_type_): _description_ iteration (_type_): _description_ + reliability (bool): include reliability in path analysis or not. + If true, reliability is included in path analysis using link field. + If false, reliability is not included in path analysis, reliability skim is overwritten as 0. warmstart (bool): True if assigning warmstart demand """ self.class_config = class_config self.time_period = time_period self.iteration = iteration + self.skim_reliability = reliability self.warmstart = warmstart self.name = class_config["name"].lower() self.skims = class_config.get("skims", []) @@ -538,6 +562,17 @@ def emme_class_analysis(self) -> List[EmmeHighwayAnalysisSpec]: for skim_type in self.skims: if skim_type == "time": continue + # if not skimming reliability in all global iterations + if not self.skim_reliability: + if skim_type in ["rlbty", "autotime"]: + continue + # if skimming reliability + # reliability is only skimmed in global iteration 0 and 1 + if self.iteration > 1: + if skim_type == "rlbty": + continue + if skim_type == "autotime": + continue if "_" in skim_type: skim_type, group = skim_type.split("_") matrix_name = f"mf{self.time_period}_{self.name}_{skim_type}_{group}" diff --git a/tm2py/config.py b/tm2py/config.py index fed43e3b..adbf06bb 100644 --- a/tm2py/config.py +++ b/tm2py/config.py @@ -87,7 +87,7 @@ class WarmStartConfig(ConfigItem): Note that the components will be executed in the order listed. Properties: - warmstart: Boolean indicating whether warmstart demand matrices are used. + warmstart: Boolean indicating whether warmstart demand matrices are used. If set to True, the global iteration 0 will either assign warmstart demand for highway and transit, or skip the assignment and just use warmstart skims. If set to False, the global iteration 0 will assign zero demand for highway and transit. warmstart_skim: Boolean indicating whether to use warmstart skims. If set to True, then skips warmstart assignment in iteraton 0. @@ -952,6 +952,12 @@ class HighwayConfig(ConfigItem): to the free-flow speed, capacity, and critical speed values interchange_nodes_file: relative path to the interchange nodes file, this is used for calculating highway reliability + reliability: bool to skim highway reliability. Default to true. If true, assignment + will be run twice in global iterations 0 (warmstart) and 1, to calculate reliability, + assignment will be run only once in global iterations 2 and 3, + reliability skim will stay the same as global iteration 1. + If false, reliability will not be calculated nor skimmed in all global + iterations, and the resulting reliability skims will be 0. """ generic_highway_mode_code: str = Field(min_length=1, max_length=1) @@ -967,6 +973,7 @@ class HighwayConfig(ConfigItem): classes: Tuple[HighwayClassConfig, ...] = Field() capclass_lookup: Tuple[HighwayCapClassConfig, ...] = Field() interchange_nodes_file: str = Field() + reliability: bool = Field(default=True) @validator("output_skim_filename_tmpl") def valid_skim_template(value):