Add global option to skip households on simulation failure

.github/workflows/core_tests.yml

@@ -130,6 +130,7 @@ jobs:
 
       - run: uv run pytest test/test_skim_name_conflicts.py
       - run: uv run pytest test/random_seed/test_random_seed.py
+      - run: uv run pytest test/skip_failed_choices/test_skip_failed_choices.py
 
   builtin_regional_models:
     needs: foundation

activitysim/abm/models/location_choice.py

@@ -234,6 +234,11 @@ def location_sample(
 ):
     # FIXME - MEMORY HACK - only include columns actually used in spec
     chooser_columns = model_settings.SIMULATE_CHOOSER_COLUMNS
+    # Drop this when PR #1017 is merged
+    if ("household_id" not in chooser_columns) and (
+        "household_id" in persons_merged.columns
+    ):
+        chooser_columns = chooser_columns + ["household_id"]
     choosers = persons_merged[chooser_columns]
 
     # create wrapper with keys for this lookup - in this case there is a home_zone_id in the choosers
@@ -390,6 +395,11 @@ def location_presample(
     # FIXME maybe we should add it for multi-zone (from maz_taz) if missing?
     chooser_columns = model_settings.SIMULATE_CHOOSER_COLUMNS
     chooser_columns = [HOME_TAZ if c == HOME_MAZ else c for c in chooser_columns]
+    # Drop this when PR #1017 is merged
+    if ("household_id" not in chooser_columns) and (
+        "household_id" in persons_merged.columns
+    ):
+        chooser_columns = chooser_columns + ["household_id"]
     choosers = persons_merged[chooser_columns]
 
     # create wrapper with keys for this lookup - in this case there is a HOME_TAZ in the choosers
@@ -620,6 +630,11 @@ def run_location_simulate(
 
     # FIXME - MEMORY HACK - only include columns actually used in spec
     chooser_columns = model_settings.SIMULATE_CHOOSER_COLUMNS
+    # Drop this when PR #1017 is merged
+    if ("household_id" not in chooser_columns) and (
+        "household_id" in persons_merged.columns
+    ):
+        chooser_columns = chooser_columns + ["household_id"]
     choosers = persons_merged[chooser_columns]
 
     alt_dest_col_name = model_settings.ALT_DEST_COL_NAME
@@ -1072,6 +1087,23 @@ def iterate_location_choice(
         else:
             choices_df = choices_df_
 
+        # drop choices that belong to the failed households: state.skipped_household_ids
+        # so that their choices are not considered in shadow price calculations
+        # first append household_id to choices_df
+        choices_df = choices_df.merge(
+            persons_merged_df[["household_id"]],
+            left_index=True,
+            right_index=True,
+            how="left",
+        )
+        if len(choices_df) > 0:
+            choices_df = choices_df[
+                ~choices_df["household_id"].isin(
+                    state.get("skipped_household_ids", set())
+                )
+            ]
+        choices_df = choices_df.drop(columns=["household_id"])
+
         spc.set_choices(
             choices=choices_df["choice"],
             segment_ids=persons_merged_df[chooser_segment_column].reindex(

activitysim/abm/models/school_escorting.py

@@ -472,6 +472,11 @@ def school_escorting(
         # reduce memory by limiting columns if selected columns are supplied
         chooser_columns = model_settings.SIMULATE_CHOOSER_COLUMNS
         if chooser_columns is not None:
+            # Drop this when PR #1017 is merged
+            if ("household_id" not in chooser_columns) and (
+                "household_id" in choosers.columns
+            ):
+                chooser_columns = chooser_columns + ["household_id"]
             chooser_columns = chooser_columns + participant_columns
             choosers = choosers[chooser_columns]
 

activitysim/abm/models/trip_matrices.py

@@ -307,6 +307,12 @@ def write_trip_matrices(
                 .TAZ.tolist()
             )
 
+    # print out number of households skipped due to failed choices
+    if state.settings.skip_failed_choices:
+        logger.info(
+            f"\n!!!\nATTENTION: Skipped households with failed choices during simulation. Number of households skipped: {state.get('num_skipped_households', 0)}.\n!!!"
+        )
+
 
 def annotate_trips(
     state: workflow.State,
@@ -393,6 +399,21 @@ def write_matrices(
     if not matrix_settings:
         logger.error("Missing MATRICES setting in write_trip_matrices.yaml")
 
+    hh_weight_col = model_settings.HH_EXPANSION_WEIGHT_COL
+    if hh_weight_col:
+        if state.get("num_skipped_households", 0) > 0:
+            logger.info(
+                f"Adjusting household expansion weights in {hh_weight_col} to account for {state.get('num_skipped_households', 0)} skipped households."
+            )
+            # adjust the hh expansion weights to account for skipped households
+            adjustment_factor = state.get_dataframe("households").shape[0] / (
+                state.get_dataframe("households").shape[0]
+                + state.get("num_skipped_households", 0)
+            )
+            aggregate_trips[hh_weight_col] = (
+                aggregate_trips[hh_weight_col] * adjustment_factor
+            )
+
     for matrix in matrix_settings:
         matrix_is_tap = matrix.is_tap
 

activitysim/abm/models/trip_mode_choice.py

@@ -366,7 +366,14 @@ def trip_mode_choice(
         "trip_mode_choice choices", trips_df[mode_column_name], value_counts=True
     )
 
-    assert not trips_df[mode_column_name].isnull().any()
+    # if we're skipping failed choices, the trip modes for failed simulations will be null
+    if state.settings.skip_failed_choices:
+        mask_skipped = trips_df["household_id"].isin(
+            state.get("skipped_household_ids", set())
+        )
+        assert not trips_df.loc[~mask_skipped, mode_column_name].isnull().any()
+    else:
+        assert not trips_df[mode_column_name].isnull().any()
 
     state.add_table("trips", trips_df)
 
@@ -382,6 +389,11 @@ def trip_mode_choice(
     # need to update locals_dict to access skims that are the same .shape as trips table
     locals_dict = {}
     locals_dict.update(constants)
+    if state.settings.skip_failed_choices:
+        mask_skipped = trips_merged["household_id"].isin(
+            state.get("skipped_household_ids", set())
+        )
+        trips_merged = trips_merged.loc[~mask_skipped]
     simulate.set_skim_wrapper_targets(trips_merged, skims)
     locals_dict.update(skims)
     locals_dict["timeframe"] = "trip"

activitysim/abm/models/util/school_escort_tours_trips.py

@@ -1043,6 +1043,10 @@ def force_escortee_trip_modes_to_match_chauffeur(state: workflow.State, trips):
         f"Changed {diff.sum()} trip modes of school escortees to match their chauffeur"
     )
 
+    # trip_mode can be na if the run allows skipping failed choices and the trip mode choice has failed
+    if state.settings.skip_failed_choices:
+        return trips
+
     assert (
         ~trips.trip_mode.isna()
     ).all(), f"Missing trip mode for {trips[trips.trip_mode.isna()]}"

activitysim/abm/models/util/tour_destination.py

@@ -625,6 +625,11 @@ def run_destination_sample(
     # if special person id is passed
     chooser_id_column = model_settings.CHOOSER_ID_COLUMN
 
+    # Drop this when PR #1017 is merged
+    if ("household_id" not in chooser_columns) and (
+        "household_id" in persons_merged.columns
+    ):
+        chooser_columns = chooser_columns + ["household_id"]
     persons_merged = persons_merged[
         [c for c in persons_merged.columns if c in chooser_columns]
     ]
@@ -799,6 +804,11 @@ def run_destination_simulate(
     # if special person id is passed
     chooser_id_column = model_settings.CHOOSER_ID_COLUMN
 
+    # Drop this when PR #1017 is merged
+    if ("household_id" not in chooser_columns) and (
+        "household_id" in persons_merged.columns
+    ):
+        chooser_columns = chooser_columns + ["household_id"]
     persons_merged = persons_merged[
         [c for c in persons_merged.columns if c in chooser_columns]
     ]

activitysim/abm/models/util/tour_od.py

@@ -692,6 +692,9 @@ def run_od_sample(
     choosers = tours
     # FIXME - MEMORY HACK - only include columns actually used in spec
     chooser_columns = model_settings.SIMULATE_CHOOSER_COLUMNS
+    # Drop this when PR #1017 is merged
+    if ("household_id" not in chooser_columns) and ("household_id" in choosers.columns):
+        chooser_columns = chooser_columns + ["household_id"]
     choosers = choosers[chooser_columns]
 
     # interaction_sample requires that choosers.index.is_monotonic_increasing
@@ -951,6 +954,9 @@ def run_od_simulate(
 
     # FIXME - MEMORY HACK - only include columns actually used in spec
     chooser_columns = model_settings.SIMULATE_CHOOSER_COLUMNS
+    # Drop this when PR #1017 is merged
+    if ("household_id" not in chooser_columns) and ("household_id" in choosers.columns):
+        chooser_columns = chooser_columns + ["household_id"]
     choosers = choosers[chooser_columns]
 
     # interaction_sample requires that choosers.index.is_monotonic_increasing

activitysim/abm/models/util/tour_scheduling.py

@@ -40,6 +40,12 @@ def run_tour_scheduling(
         c for c in model_columns if c not in logsum_columns
     ]
 
+    # Drop this when PR #1017 is merged
+    if ("household_id" not in chooser_columns) and (
+        "household_id" in persons_merged.columns
+    ):
+        chooser_columns = chooser_columns + ["household_id"]
+
     persons_merged = expressions.filter_chooser_columns(persons_merged, chooser_columns)
 
     timetable = state.get_injectable("timetable")

activitysim/core/configuration/top.py

@@ -781,6 +781,13 @@ def _check_store_skims_in_shm(self):
     should catch many common errors early, including missing required configurations or specified coefficient labels without defined values.  
     """
 
+    skip_failed_choices: bool = True
+    """
+    Skip households that cause errors during processing instead of failing the model run.
+
+    .. versionadded:: 1.6
+    """
+
     other_settings: dict[str, Any] = None
 
     def _get_attr(self, attr):

activitysim/core/interaction_sample_simulate.py

@@ -351,6 +351,11 @@ def _interaction_sample_simulate(
         # that is, we want the index value of the row that is offset by <position> rows into the
         # tranche of this choosers alternatives created by cross join of alternatives and choosers
 
+        # when skip failed choices is enabled, the position may be -99 for failed choices, which gets droppped eventually
+        # here we just need to clip to zero to avoid getting the wrong index in the take() below
+        if state.settings.skip_failed_choices:
+            positions = positions.clip(lower=0)
+
         # resulting pandas Int64Index has one element per chooser row and is in same order as choosers
         choices = alternatives[choice_column].take(positions + first_row_offsets)
 

activitysim/core/logit.py

@@ -30,6 +30,7 @@ def report_bad_choices(
     state: workflow.State,
     bad_row_map,
     df,
+    skip_failed_choices,
     trace_label,
     msg,
     trace_choosers=None,
@@ -87,6 +88,27 @@ def report_bad_choices(
 
         logger.warning(row_msg)
 
+    if skip_failed_choices:
+        # update counter in state
+        num_skipped_households = state.get("num_skipped_households", 0)
+        skipped_household_ids = state.get("skipped_household_ids", set())
+        for hh_id in df[trace_col].unique():
+            if hh_id is None:
+                continue
+            if hh_id not in skipped_household_ids:
+                skipped_household_ids.add(hh_id)
+                num_skipped_households += 1
+            else:
+                continue
+        state.set("num_skipped_households", num_skipped_households)
+        state.set("skipped_household_ids", skipped_household_ids)
+
+        logger.debug(
+            f"Skipping {bad_row_map.sum()} bad choices. Total skipped households so far: {num_skipped_households}. Skipped household IDs: {skipped_household_ids}"
+        )
+
+        return
+
     if raise_error:
         raise InvalidTravelError(msg_with_count)
 
@@ -136,6 +158,7 @@ def utils_to_probs(
     allow_zero_probs=False,
     trace_choosers=None,
     overflow_protection: bool = True,
+    skip_failed_choices: bool = True,
     return_logsums: bool = False,
 ):
     """
@@ -176,6 +199,16 @@ def utils_to_probs(
         overflow_protection will have no benefit but impose a modest computational
         overhead cost.
 
+    skip_failed_choices : bool, default True
+        If True, when bad choices are detected (all zero probabilities or infinite
+        probabilities), the entire household that's causing bad choices will be skipped instead of
+        being masked by overflow protection or causing an error.
+        A counter will be incremented for each skipped household. This is useful when running large
+        simulations where occasional bad choices are encountered and should not halt the process.
+        The counter can be accessed via `state.get("num_skipped_households", 0)`.
+        The number of skipped households and their IDs will be logged at the end of the simulation.
+        When `skip_failed_choices` is True, `overflow_protection` will be reverted to False to avoid conflicts.
+
     Returns
     -------
     probs : pandas.DataFrame
@@ -203,6 +236,13 @@ def utils_to_probs(
             utils_arr.dtype == np.float32 and utils_arr.max() > 85
         )
 
+    if state.settings.skip_failed_choices is not None:
+        skip_failed_choices = state.settings.skip_failed_choices
+        # when skipping failed choices, we cannot use overflow protection
+        # because it would mask the underlying issue causing bad choices
+        if skip_failed_choices:
+            overflow_protection = False
+
     if overflow_protection:
         # exponentiated utils will overflow, downshift them
         shifts = utils_arr.max(1, keepdims=True)
@@ -240,6 +280,7 @@ def utils_to_probs(
                 state,
                 zero_probs,
                 utils,
+                skip_failed_choices,
                 trace_label=tracing.extend_trace_label(trace_label, "zero_prob_utils"),
                 msg="all probabilities are zero",
                 trace_choosers=trace_choosers,
@@ -251,6 +292,7 @@ def utils_to_probs(
             state,
             inf_utils,
             utils,
+            skip_failed_choices,
             trace_label=tracing.extend_trace_label(trace_label, "inf_exp_utils"),
             msg="infinite exponentiated utilities",
             trace_choosers=trace_choosers,
@@ -281,6 +323,7 @@ def make_choices(
     trace_label: str = None,
     trace_choosers=None,
     allow_bad_probs=False,
+    skip_failed_choices=True,
 ) -> tuple[pd.Series, pd.Series]:
     """
     Make choices for each chooser from among a set of alternatives.
@@ -316,11 +359,15 @@ def make_choices(
         np.ones(len(probs.index))
     ).abs() > BAD_PROB_THRESHOLD * np.ones(len(probs.index))
 
+    if state.settings.skip_failed_choices is not None:
+        skip_failed_choices = state.settings.skip_failed_choices
+
     if bad_probs.any() and not allow_bad_probs:
         report_bad_choices(
             state,
             bad_probs,
             probs,
+            skip_failed_choices,
             trace_label=tracing.extend_trace_label(trace_label, "bad_probs"),
             msg="probabilities do not add up to 1",
             trace_choosers=trace_choosers,
@@ -329,6 +376,8 @@ def make_choices(
     rands = state.get_rn_generator().random_for_df(probs)
 
     choices = pd.Series(choice_maker(probs.values, rands), index=probs.index)
+    # mark bad choices with -99
+    choices[bad_probs] = -99
 
     rands = pd.Series(np.asanyarray(rands).flatten(), index=probs.index)
 

activitysim/core/simulate.py

@@ -1325,7 +1325,9 @@ def eval_mnl(
     if custom_chooser:
         choices, rands = custom_chooser(state, probs, choosers, spec, trace_label)
     else:
-        choices, rands = logit.make_choices(state, probs, trace_label=trace_label)
+        choices, rands = logit.make_choices(
+            state, probs, trace_label=trace_label, trace_choosers=choosers
+        )
 
     del probs
     chunk_sizer.log_df(trace_label, "probs", None)
@@ -1485,11 +1487,15 @@ def eval_nl(
     BAD_PROB_THRESHOLD = 0.001
     no_choices = (base_probabilities.sum(axis=1) - 1).abs() > BAD_PROB_THRESHOLD
 
+    if state.settings.skip_failed_choices is not None:
+        skip_failed_choices = state.settings.skip_failed_choices
+
     if no_choices.any():
         logit.report_bad_choices(
             state,
             no_choices,
             base_probabilities,
+            skip_failed_choices,
             trace_label=tracing.extend_trace_label(trace_label, "bad_probs"),
             trace_choosers=choosers,
             msg="base_probabilities do not sum to one",

activitysim/core/test/test_logit.py

@@ -78,6 +78,7 @@ def test_utils_to_probs(utilities, test_data):
 
 def test_utils_to_probs_raises():
     state = workflow.State().default_settings()
+    state.settings.skip_failed_choices = False
     idx = pd.Index(name="household_id", data=[1])
     with pytest.raises(RuntimeError) as excinfo:
         logit.utils_to_probs(