BayDAG Estimation Enhancements

activitysim/abm/models/auto_ownership.py

@@ -2,9 +2,9 @@
 # See full license in LICENSE.txt.
 import logging
 
-from activitysim.core import config, inject, pipeline, simulate, tracing
+from activitysim.core import config, expressions, inject, pipeline, simulate, tracing
 
-from .util import estimation
+from .util import estimation, annotate
 
 logger = logging.getLogger(__name__)
 
@@ -32,6 +32,21 @@ def auto_ownership_simulate(households, households_merged, chunk_size, trace_hh_
 
     logger.info("Running %s with %d households", trace_label, len(choosers))
 
+    # - preprocessor
+    preprocessor_settings = model_settings.get("preprocessor", None)
+    if preprocessor_settings:
+
+        locals_d = {}
+        if constants is not None:
+            locals_d.update(constants)
+
+        expressions.assign_columns(
+            df=choosers,
+            model_settings=preprocessor_settings,
+            locals_dict=locals_d,
+            trace_label=trace_label,
+        )
+
     if estimator:
         estimator.write_model_settings(model_settings, model_settings_file_name)
         estimator.write_spec(model_settings)
@@ -69,5 +84,8 @@ def auto_ownership_simulate(households, households_merged, chunk_size, trace_hh_
         "auto_ownership", households.auto_ownership, value_counts=True
     )
 
+    if model_settings.get("annotate_households"):
+        annotate.annotate_households(model_settings, trace_label)
+
     if trace_hh_id:
         tracing.trace_df(households, label="auto_ownership", warn_if_empty=True)

activitysim/abm/models/cdap.py

@@ -151,6 +151,11 @@ def cdap_simulate(persons_merged, persons, households, chunk_size, trace_hh_id):
         for hhsize in range(2, cdap.MAX_HHSIZE + 1):
             spec = cdap.get_cached_spec(hhsize)
             estimator.write_table(spec, "spec_%s" % hhsize, append=False)
+            if add_joint_tour_utility:
+                joint_spec = cdap.get_cached_joint_spec(hhsize)
+                estimator.write_table(
+                    joint_spec, "joint_spec_%s" % hhsize, append=False
+                )
 
     logger.info("Running cdap_simulate with %d persons", len(persons_merged.index))
 
@@ -184,6 +189,11 @@ def cdap_simulate(persons_merged, persons, households, chunk_size, trace_hh_id):
     if estimator:
         estimator.write_choices(choices)
         choices = estimator.get_survey_values(choices, "persons", "cdap_activity")
+        if add_joint_tour_utility:
+            hh_joint.index.name = "household_id"
+            hh_joint = estimator.get_survey_values(
+                hh_joint, "households", "has_joint_tour"
+            )
         estimator.write_override_choices(choices)
         estimator.end_estimation()
 

activitysim/abm/models/disaggregate_accessibility.py

@@ -275,6 +275,10 @@ def zone_sampler(self):
                     maz_candidates = maz_candidates[
                         ~maz_candidates.MAZ.isin(maz_sample_idx)
                     ]
+
+                    # Need to make sure we sample from TAZs that still exist in the maz_candidates
+                    taz_candidates = taz_candidates[taz_candidates.index.isin(maz_candidates.TAZ)]
+
                     # Calculate the remaining samples to collect
                     n_samples_remaining = n_samples - len(maz_sample_idx)
                     n_samples_remaining = (
@@ -569,14 +573,13 @@ def merge_persons(self):
         inject.add_table("proto_persons_merged", persons_merged)
 
 
-def get_disaggregate_logsums(network_los, chunk_size, trace_hh_id):
+def get_disaggregate_logsums(
+    network_los, chunk_size, trace_hh_id, disagg_model_settings
+):
     logsums = {}
     persons_merged = pipeline.get_table("proto_persons_merged").sort_index(
         inplace=False
     )
-    disagg_model_settings = read_disaggregate_accessibility_yaml(
-        "disaggregate_accessibility.yaml"
-    )
 
     for model_name in [
         "workplace_location",
@@ -696,8 +699,14 @@ def compute_disaggregate_accessibility(network_los, chunk_size, trace_hh_id):
             tracing.register_traceable_table(tablename, df)
         del df
 
+    disagg_model_settings = read_disaggregate_accessibility_yaml(
+        "disaggregate_accessibility.yaml"
+    )
+
     # Run location choice
-    logsums = get_disaggregate_logsums(network_los, chunk_size, trace_hh_id)
+    logsums = get_disaggregate_logsums(
+        network_los, chunk_size, trace_hh_id, disagg_model_settings
+    )
     logsums = {k + "_accessibility": v for k, v in logsums.items()}
 
     # Combined accessibility table
@@ -736,20 +745,20 @@ def compute_disaggregate_accessibility(network_los, chunk_size, trace_hh_id):
     logsums["proto_disaggregate_accessibility"] = access_df
 
     # Drop any tables prematurely created
-    for tablename in [
-        "school_destination_size",
-        "workplace_destination_size",
-    ]:
-        pipeline.drop_table(tablename)
+    # FIXME: dropping size tables breaks restart functionality for location choice models.
+    #        hopefully this pipeline mess just goes away with move away from orca....
+    # for tablename in [
+    #     "school_destination_size",
+    #     "workplace_destination_size",
+    # ]:
+    #     pipeline.drop_table(tablename)
 
     for ch in list(pipeline.get_rn_generator().channels.keys()):
         pipeline.get_rn_generator().drop_channel(ch)
 
-    # Drop any prematurely added traceables
-    for trace in [
-        x for x in inject.get_injectable("traceable_tables") if "proto_" not in x
-    ]:
-        tracing.deregister_traceable_table(trace)
+    # Dropping all traceable tables
+    for table in inject.get_injectable("traceable_tables"):
+        tracing.deregister_traceable_table(table)
 
     # need to clear any premature tables that were added during the previous run
     orca._TABLES.clear()
@@ -760,4 +769,22 @@ def compute_disaggregate_accessibility(network_los, chunk_size, trace_hh_id):
     # Inject accessibility results into pipeline
     [inject.add_table(k, df) for k, df in logsums.items()]
 
+    # available post-processing
+    for annotations in disagg_model_settings.get("postprocess_proto_tables", []):
+        tablename = annotations["tablename"]
+        df = pipeline.get_table(tablename)
+        assert df is not None
+        assert annotations is not None
+        assign_columns(
+            df=df,
+            model_settings={
+                **annotations["annotate"],
+                **disagg_model_settings["suffixes"],
+            },
+            trace_label=tracing.extend_trace_label(
+                "disaggregate_accessibility.postprocess", tablename
+            ),
+        )
+        pipeline.replace_table(tablename, df)
+
     return

activitysim/abm/models/joint_tour_destination.py

@@ -62,7 +62,7 @@ def joint_tour_destination(
         estimator.write_model_settings(model_settings, model_settings_file_name)
 
     choices_df, save_sample_df = tour_destination.run_tour_destination(
-        tours,
+        joint_tours,
         persons_merged,
         want_logsums,
         want_sample_table,

activitysim/abm/models/joint_tour_frequency_composition.py

@@ -35,9 +35,15 @@ def joint_tour_frequency_composition(
 
     model_settings = config.read_model_settings(model_settings_file_name)
 
+    # FIXME setting index as "alt" causes crash in estimation mode...
+    # happens in school escorting too!
+    # alt_tdd = simulate.read_model_alts(
+    #     "joint_tour_frequency_composition_alternatives.csv", set_index="alt"
+    # )
     alt_tdd = simulate.read_model_alts(
-        "joint_tour_frequency_composition_alternatives.csv", set_index="alt"
+        "joint_tour_frequency_composition_alternatives.csv", set_index=None
     )
+    alt_tdd.index = alt_tdd["alt"].values
 
     # - only interested in households with more than one cdap travel_active person and
     # - at least one non-preschooler
@@ -94,14 +100,16 @@ def joint_tour_frequency_composition(
         estimator.write_model_settings(model_settings, model_settings_file_name)
         estimator.write_coefficients(coefficients_df, model_settings)
         estimator.write_choosers(choosers)
-        estimator.write_alternatives(alts)
 
         assert choosers.index.name == "household_id"
         assert "household_id" not in choosers.columns
         choosers["household_id"] = choosers.index
 
         estimator.set_chooser_id(choosers.index.name)
 
+        # FIXME set_alt_id - do we need this for interaction_simulate estimation bundle tables?
+        estimator.set_alt_id("alt_id")
+
     # The choice value 'joint_tour_frequency_composition' assigned by interaction_simulate
     # is the index value of the chosen alternative in the alternatives table.
     choices = interaction_simulate(
@@ -134,6 +142,7 @@ def joint_tour_frequency_composition(
     # - but we don't know the tour participants yet
     # - so we arbitrarily choose the first person in the household
     # - to be point person for the purpose of generating an index and setting origin
+    # FIXME: not all models are guaranteed to have PNUM
     temp_point_persons = persons.loc[persons.PNUM == 1]
     temp_point_persons["person_id"] = temp_point_persons.index
     temp_point_persons = temp_point_persons.set_index("household_id")

activitysim/abm/models/joint_tour_participation.py

@@ -209,6 +209,10 @@ def participants_chooser(probs, choosers, spec, trace_label):
                 probs[choice_col] = np.where(probs[choice_col] > 0, 1, 0)
                 non_choice_col = [col for col in probs.columns if col != choice_col][0]
                 probs[non_choice_col] = 1 - probs[choice_col]
+                if iter > MAX_ITERATIONS + 1:
+                    raise RuntimeError(
+                        f"{num_tours_remaining} tours could not be satisfied even with forcing participation"
+                    )
             else:
                 raise RuntimeError(
                     f"{num_tours_remaining} tours could not be satisfied after {iter} iterations"