diff --git a/activitysim/abm/models/atwork_subtour_destination.py b/activitysim/abm/models/atwork_subtour_destination.py index 1c313a089..6b8a0aec3 100644 --- a/activitysim/abm/models/atwork_subtour_destination.py +++ b/activitysim/abm/models/atwork_subtour_destination.py @@ -7,7 +7,7 @@ import pandas as pd from activitysim.abm.models.util import tour_destination -from activitysim.core import config, estimation, los, tracing, workflow, expressions +from activitysim.core import config, estimation, expressions, los, tracing, workflow from activitysim.core.configuration.logit import TourLocationComponentSettings from activitysim.core.util import assign_in_place @@ -111,7 +111,8 @@ def atwork_subtour_destination( destination_column_name, subtours[destination_column_name], describe=True ) - if want_sample_table: + # save_sample_df is None if every chooser failed sampling under skip_failed_choices + if want_sample_table and save_sample_df is not None: assert len(save_sample_df.index.get_level_values(0).unique()) == len(choices_df) # save_sample_df.set_index(model_settings['ALT_DEST_COL_NAME'], append=True, inplace=True) state.extend_table(sample_table_name, save_sample_df) diff --git a/activitysim/abm/models/joint_tour_destination.py b/activitysim/abm/models/joint_tour_destination.py index 29b117821..341194e2d 100644 --- a/activitysim/abm/models/joint_tour_destination.py +++ b/activitysim/abm/models/joint_tour_destination.py @@ -7,11 +7,10 @@ import pandas as pd from activitysim.abm.models.util import tour_destination -from activitysim.core import estimation, los, tracing, workflow, expressions +from activitysim.core import estimation, expressions, los, tracing, workflow from activitysim.core.configuration.logit import TourLocationComponentSettings from activitysim.core.util import assign_in_place - logger = logging.getLogger(__name__) @@ -94,7 +93,6 @@ def joint_tour_destination( state.settings.downcast_int, state.settings.downcast_float, ) - state.add_table("tours", tours) if want_logsums: joint_tours[logsum_column_name] = choices_df["logsum"] @@ -105,9 +103,14 @@ def joint_tour_destination( state.settings.downcast_float, ) + # add_table drops the tours of households skipped under skip_failed_choices, + # so assign every column first + state.add_table("tours", tours) + tracing.print_summary("destination", joint_tours.destination, describe=True) - if want_sample_table: + # save_sample_df is None if every chooser failed sampling under skip_failed_choices + if want_sample_table and save_sample_df is not None: assert len(save_sample_df.index.get_level_values(0).unique()) == len(choices_df) # save_sample_df.set_index(model_settings['ALT_DEST_COL_NAME'], append=True, inplace=True) state.extend_table(sample_table_name, save_sample_df) diff --git a/activitysim/abm/models/location_choice.py b/activitysim/abm/models/location_choice.py index c955f4a60..f4320a0b1 100644 --- a/activitysim/abm/models/location_choice.py +++ b/activitysim/abm/models/location_choice.py @@ -463,6 +463,9 @@ def location_presample( n_total_alts=n_total_alts, ) + if state.settings.skip_failed_choices and taz_sample.empty: + return taz_sample.rename(columns={DEST_TAZ: alt_dest_col_name}) + # print(f"taz_sample\n{taz_sample}") # dest_TAZ prob pick_count # person_id @@ -869,6 +872,12 @@ def run_location_choice( ), ) + if state.settings.skip_failed_choices: + # Failed sampling choices have no alternatives for logsums or simulation. + choosers = choosers[choosers.index.isin(location_sample_df.index)] + if choosers.empty: + continue + # - location_logsums # skip logsum calculations if LOGSUM_SETTINGS is None if model_settings.LOGSUM_SETTINGS: diff --git a/activitysim/abm/models/non_mandatory_destination.py b/activitysim/abm/models/non_mandatory_destination.py index b0c6ea3fb..56062f744 100644 --- a/activitysim/abm/models/non_mandatory_destination.py +++ b/activitysim/abm/models/non_mandatory_destination.py @@ -7,7 +7,7 @@ import pandas as pd from activitysim.abm.models.util import tour_destination -from activitysim.core import estimation, los, tracing, workflow, expressions +from activitysim.core import estimation, expressions, logit, los, tracing, workflow from activitysim.core.configuration.logit import TourLocationComponentSettings from activitysim.core.util import assign_in_place @@ -123,13 +123,17 @@ def non_mandatory_tour_destination( state.settings.downcast_float, ) - assert all( - ~tours["destination"].isna() - ), f"Tours are missing destination: {tours[tours['destination'].isna()]}" + # tours of households skipped under skip_failed_choices have no destination; + # add_table drops them + missing = tours["destination"].isna() & ~tours["household_id"].isin( + logit.skipped_household_ids(state) + ) + assert not missing.any(), f"Tours are missing destination: {tours[missing]}" state.add_table("tours", tours) - if want_sample_table: + # save_sample_df is None if every chooser failed sampling under skip_failed_choices + if want_sample_table and save_sample_df is not None: assert len(save_sample_df.index.get_level_values(0).unique()) == len(choices_df) # save_sample_df.set_index(model_settings['ALT_DEST_COL_NAME'], append=True, inplace=True) state.extend_table(sample_table_name, save_sample_df) diff --git a/activitysim/abm/models/tour_od_choice.py b/activitysim/abm/models/tour_od_choice.py index ad9f527d7..3d80552c5 100644 --- a/activitysim/abm/models/tour_od_choice.py +++ b/activitysim/abm/models/tour_od_choice.py @@ -7,7 +7,7 @@ import pandas as pd from activitysim.abm.models.util import tour_od -from activitysim.core import estimation, los, workflow, expressions +from activitysim.core import estimation, expressions, los, workflow logger = logging.getLogger(__name__) @@ -134,7 +134,8 @@ def tour_od_choice( state.add_table("persons", persons) state.add_table("households", households) - if want_sample_table: + # save_sample_df is None if every chooser failed sampling under skip_failed_choices + if want_sample_table and save_sample_df is not None: assert len(save_sample_df.index.get_level_values(0).unique()) == len(choices_df) state.extend_table(sample_table_name, save_sample_df) diff --git a/activitysim/abm/models/util/tour_destination.py b/activitysim/abm/models/util/tour_destination.py index 11bdb8ecd..2e90645a8 100644 --- a/activitysim/abm/models/util/tour_destination.py +++ b/activitysim/abm/models/util/tour_destination.py @@ -15,6 +15,7 @@ config, estimation, expressions, + logit, los, simulate, tracing, @@ -175,7 +176,9 @@ def _destination_sample( # remember person_id in chosen alts so we can merge with persons in subsequent steps # (broadcasts person_id onto all alternatives sharing the same tour_id index value) - choices[chooser_id_column] = choosers[chooser_id_column] + # reindex: assigning a Series to an empty sample would create rows for + # choosers whose sampling failed + choices[chooser_id_column] = choosers[chooser_id_column].reindex(choices.index) return choices @@ -664,6 +667,10 @@ def destination_presample( n_total_alts=len(full_taz_index) if full_taz_index is not None else None, ) + if state.settings.skip_failed_choices and taz_sample.empty: + # every chooser failed sampling, so there is no TAZ to choose a MAZ for + return taz_sample.rename(columns={DEST_TAZ: alt_dest_col_name}) + # choose a MAZ for each DEST_TAZ choice, choice probability based on MAZ size_term fraction of TAZ total maz_choices = choose_MAZ_for_TAZ( state, @@ -749,7 +756,7 @@ def run_destination_sample( # remember person_id in chosen alts so we can merge with persons in subsequent steps # (broadcasts person_id onto all alternatives sharing the same tour_id index value) - choices[chooser_id_column] = tours[chooser_id_column] + choices[chooser_id_column] = tours[chooser_id_column].reindex(choices.index) return choices @@ -1027,6 +1034,19 @@ def run_tour_destination( trace_label=tracing.extend_trace_label(segment_trace_label, "sample"), ) + if state.settings.skip_failed_choices: + # Failed sampling choices have no alternatives for logsums or + # simulation. Record their households (the sample choosers carry no + # household_id) so add_table drops them with the rest of the household. + failed = ~choosers.index.isin(location_sample_df.index) + if failed.any(): + logit.record_skipped_households( + state, choosers.loc[failed, "household_id"], segment_trace_label + ) + choosers = choosers[~failed] + if choosers.empty: + continue + # - destination_logsums # if LOGSUM_SETTINGS is set to 'None', we skip this step if model_settings.LOGSUM_SETTINGS: diff --git a/activitysim/abm/models/util/tour_od.py b/activitysim/abm/models/util/tour_od.py index 6626dccc2..8d56a750e 100644 --- a/activitysim/abm/models/util/tour_od.py +++ b/activitysim/abm/models/util/tour_od.py @@ -683,6 +683,16 @@ def od_presample( trace_label, ) + alt_dest_col_name = model_settings.ALT_DEST_COL_NAME + chooser_orig_col_name = model_settings.CHOOSER_ORIG_COL_NAME + + if state.settings.skip_failed_choices and orig_MAZ_dest_TAZ_sample.empty: + # every chooser failed sampling, so there is no TAZ to choose a MAZ for + return pd.DataFrame( + columns=[alt_dest_col_name, chooser_orig_col_name, "prob", "pick_count"], + index=orig_MAZ_dest_TAZ_sample.index, + ) + orig_MAZ_dest_TAZ_sample[ORIG_MAZ] = ( orig_MAZ_dest_TAZ_sample[alt_od_col_name].str.split("_").str[0].astype(int) ) @@ -705,8 +715,6 @@ def od_presample( # outputs assert DEST_MAZ in maz_choices - alt_dest_col_name = model_settings.ALT_DEST_COL_NAME - chooser_orig_col_name = model_settings.CHOOSER_ORIG_COL_NAME maz_choices = maz_choices.rename( columns={DEST_MAZ: alt_dest_col_name, ORIG_MAZ: chooser_orig_col_name} ) @@ -1161,6 +1169,21 @@ def run_tour_od( ), ) + if state.settings.skip_failed_choices: + # Failed sampling choices have no alternatives for logsums or + # simulation; record their households so add_table drops them. + failed = ~choosers.index.isin(od_sample_df.index) + if failed.any(): + if "household_id" in choosers: + logit.record_skipped_households( + state, + choosers.loc[failed, "household_id"], + tracing.extend_trace_label(trace_label, segment_name), + ) + choosers = choosers[~failed] + if choosers.empty: + continue + if model_settings.ORIG_FILTER == "original_MAZ > 0": pass elif model_settings.ORIG_FILTER == "external_TAZ > 0": @@ -1232,6 +1255,13 @@ def run_tour_od( if len(choices_list) > 0: choices_df = pd.concat(choices_list) + else: + # every segment was empty, e.g. all of its choosers failed sampling + logger.warning("%s no choices", trace_label) + choices_df = pd.DataFrame( + columns=["choice", origin_col_name, model_settings.DEST_COL_NAME] + + (["logsum"] if want_logsums else []) + ) if len(sample_list) > 0: save_sample_df = pd.concat(sample_list) diff --git a/activitysim/abm/test/test_location_choice.py b/activitysim/abm/test/test_location_choice.py index 9bcdd7711..65a945c06 100644 --- a/activitysim/abm/test/test_location_choice.py +++ b/activitysim/abm/test/test_location_choice.py @@ -5,6 +5,7 @@ import pandas as pd import pandas.testing as pdt +import pytest from activitysim.abm.models import location_choice @@ -58,6 +59,7 @@ def test_estimation_override_preserves_destination_choice_logsum(monkeypatch): state = SimpleNamespace( settings=SimpleNamespace( sample_method="inverse_cdf", + skip_failed_choices=False, trace_hh_id=None, use_explicit_error_terms=False, ) @@ -88,3 +90,69 @@ def test_estimation_override_preserves_destination_choice_logsum(monkeypatch): ) pdt.assert_frame_equal(choices, expected) assert sample is None + + +@pytest.mark.parametrize("sampled_person_ids", [[], [2, 2]]) +def test_location_choice_skips_failed_samples(monkeypatch, sampled_person_ids): + """Only persons with sampled alternatives reach logsums and final choice.""" + persons = pd.DataFrame( + {"segment": [1, 1]}, index=pd.Index([1, 2], name="person_id") + ) + location_sample = pd.DataFrame( + {"alt_dest": [101, 202][: len(sampled_person_ids)]}, + index=pd.Index(sampled_person_ids, name="person_id"), + ) + expected_persons = persons.loc[persons.index.isin(sampled_person_ids)] + sample_model = Mock(return_value=location_sample) + logsums_model = Mock( + return_value=location_sample.assign(**{location_choice.ALT_LOGSUM: 0.0}) + ) + simulate_model = Mock( + return_value=pd.DataFrame({"choice": 101}, index=expected_persons.index) + ) + monkeypatch.setattr(location_choice, "run_location_sample", sample_model) + monkeypatch.setattr(location_choice, "run_location_logsums", logsums_model) + monkeypatch.setattr(location_choice, "run_location_simulate", simulate_model) + shadow_price_calculator = Mock() + shadow_price_calculator.dest_size_terms.return_value = pd.Series( + [1.0, 1.0], index=[101, 202] + ) + model_settings = SimpleNamespace( + ALT_DEST_COL_NAME="alt_dest", + CHOOSER_SEGMENT_COLUMN_NAME="segment", + LOGSUM_SETTINGS="tour_mode_choice.yaml", + SEGMENT_IDS={"workers": 1}, + ) + state = SimpleNamespace( + settings=SimpleNamespace( + skip_failed_choices=True, use_explicit_error_terms=True + ) + ) + + choices, sample = location_choice.run_location_choice( + state=state, + persons_merged_df=persons, + network_los=Mock(), + shadow_price_calculator=shadow_price_calculator, + want_logsums=False, + want_sample_table=True, + estimator=None, + model_settings=model_settings, + chunk_size=0, + chunk_tag="workplace_location", + trace_label="workplace_location", + ) + + if expected_persons.empty: + assert choices.empty + assert sample is None + logsums_model.assert_not_called() + simulate_model.assert_not_called() + else: + pdt.assert_index_equal(choices.index, expected_persons.index) + pdt.assert_frame_equal(logsums_model.call_args.args[2], expected_persons) + pdt.assert_frame_equal(simulate_model.call_args.args[2], expected_persons) + pdt.assert_index_equal( + sample.index.get_level_values("person_id").unique(), + expected_persons.index, + ) diff --git a/activitysim/abm/test/test_misc/test_location_choice_sampling.py b/activitysim/abm/test/test_misc/test_location_choice_sampling.py index 6b21e0329..262781c79 100644 --- a/activitysim/abm/test/test_misc/test_location_choice_sampling.py +++ b/activitysim/abm/test/test_misc/test_location_choice_sampling.py @@ -1,5 +1,8 @@ from __future__ import annotations +from types import SimpleNamespace +from unittest.mock import Mock + import pandas as pd from activitysim.abm.models import location_choice @@ -23,6 +26,58 @@ def get_skim_dict(self, layer): return _DummySkimDict() +def test_location_presample_skips_empty_taz_sample(monkeypatch): + state = workflow.State().default_settings() + state.settings.skip_failed_choices = True + state.settings.use_explicit_error_terms = True + taz_sample = pd.DataFrame( + columns=[location_choice.DEST_TAZ, "prob", "pick_count"], + index=pd.Index([], name="person_id"), + ) + monkeypatch.setattr( + location_choice, "_location_sample", Mock(return_value=taz_sample) + ) + monkeypatch.setattr( + location_choice.shadow_pricing, + "load_shadow_price_calculator", + Mock(return_value=SimpleNamespace(use_shadow_pricing=False)), + ) + choose_maz = Mock() + monkeypatch.setattr( + location_choice.tour_destination, "choose_MAZ_for_TAZ", choose_maz + ) + persons = pd.DataFrame( + {location_choice.HOME_MAZ: [9001], location_choice.HOME_TAZ: [90]}, + index=pd.Index([1001], name="person_id"), + ) + size_terms = pd.DataFrame( + { + "size_term": [1.0, 2.0], + "shadow_price_size_term_adjustment": [1.0, 1.0], + "shadow_price_utility_adjustment": [0.0, 0.0], + }, + index=pd.Index([101, 102], name="zone_id"), + ) + + out = location_choice.location_presample( + state, + "workers", + persons, + _DummyNetworkLos({101: 1, 102: 2}), + size_terms, + estimator=None, + model_settings=SimpleNamespace(ALT_DEST_COL_NAME="zone_id"), + chunk_size=0, + chunk_tag="workplace_location", + trace_label="workplace_location", + ) + + pd.testing.assert_frame_equal( + out, taz_sample.rename(columns={location_choice.DEST_TAZ: "zone_id"}) + ) + choose_maz.assert_not_called() + + def test_location_presample_uses_taz_stable_mapping(monkeypatch): captured = {} diff --git a/activitysim/abm/test/test_misc/test_tour_destination_sampling.py b/activitysim/abm/test/test_misc/test_tour_destination_sampling.py index 2157fdcad..8df3a3273 100644 --- a/activitysim/abm/test/test_misc/test_tour_destination_sampling.py +++ b/activitysim/abm/test/test_misc/test_tour_destination_sampling.py @@ -1,12 +1,14 @@ from __future__ import annotations from types import SimpleNamespace +from unittest.mock import Mock import numpy as np import pandas as pd +import pytest from activitysim.abm.models.util import tour_destination -from activitysim.core import workflow +from activitysim.core import logit, los, workflow class _DummySkimDict: @@ -168,6 +170,70 @@ def fake_choose_maz_for_taz( ) +@pytest.mark.parametrize("use_explicit_error_terms", [False, True], ids=["mnl", "eet"]) +def test_destination_presample_returns_empty_sample_when_every_chooser_failed( + monkeypatch, use_explicit_error_terms +): + """An empty TAZ sample is returned as is, without choosing MAZs. + + Under skip_failed_choices, the TAZ sample is empty when every chooser in the + segment failed sampling. choose_MAZ_for_TAZ raises on an empty sample. + """ + + def fake_destination_sample(*_args, **_kwargs): + # interaction_sample's result when every chooser failed + return pd.DataFrame( + { + tour_destination.DEST_TAZ: pd.Series(dtype="int64"), + "prob": pd.Series(dtype="float32"), + "pick_count": pd.Series(dtype="uint32"), + "person_id": pd.Series(dtype="int64"), + }, + index=pd.Index([], name="tour_id", dtype="int64"), + ) + + monkeypatch.setattr( + tour_destination, "_destination_sample", fake_destination_sample + ) + + state = workflow.State().default_settings() + state.settings.skip_failed_choices = True + state.settings.use_explicit_error_terms = use_explicit_error_terms + choosers = pd.DataFrame( + {"origin": [101, 103]}, + index=pd.Index([7001, 7002], name="tour_id"), + ) + model_settings = type( + "ModelSettings", + (), + { + "ALT_DEST_COL_NAME": "zone_id", + "CHOOSER_ORIG_COL_NAME": "origin", + }, + )() + network_los = _DummyNetworkLos({101: 1, 102: 2, 103: 3}) + destination_size_terms = pd.DataFrame( + {"size_term": [1.0, 2.0, 3.0]}, + index=pd.Index([101, 102, 103], name="zone_id"), + ) + + out = tour_destination.destination_presample( + state, + "segment", + choosers, + model_settings, + network_los, + destination_size_terms, + destination_size_terms, + estimator=None, + trace_label="test_trace", + ) + + assert out.empty + assert out.index.name == "tour_id" + assert list(out.columns) == ["zone_id", "prob", "pick_count", "person_id"] + + def test_choose_maz_for_taz_supports_variable_taz_counts(): state = _DummyState([[0.2, 0.81], [0.1, 0.9]]) @@ -417,3 +483,116 @@ def fake_destination_sample( assert captured["n_total_alts"] == 3 assert captured["alt_dest_col_name"] == "zone_id" assert captured["zone_layer"] is None + + +@pytest.mark.parametrize("sampled_tours", [[], [5, 5, 6]], ids=["empty", "sampled"]) +def test_run_destination_sample_copies_person_id_without_adding_rows( + monkeypatch, sampled_tours +): + """person_id is broadcast onto sampled rows; an empty sample stays empty.""" + tours = pd.DataFrame({"person_id": [7, 8]}, index=pd.Index([5, 6], name="tour_id")) + persons_merged = pd.DataFrame( + {"home_zone_id": [1, 2]}, index=pd.Index([7, 8], name="person_id") + ) + sample = pd.DataFrame( + {"alt_dest": [10, 20, 10][: len(sampled_tours)]}, + index=pd.Index(sampled_tours, name="tour_id", dtype="int64"), + ) + monkeypatch.setattr( + tour_destination, "destination_sample", lambda *args, **kwargs: sample.copy() + ) + + result = tour_destination.run_destination_sample( + SimpleNamespace(settings=SimpleNamespace(want_dest_choice_presampling=True)), + "shopping", + tours, + persons_merged, + SimpleNamespace(CHOOSER_ID_COLUMN="person_id"), + SimpleNamespace(zone_system=los.ONE_ZONE), + None, + None, + None, + chunk_size=0, + trace_label="test", + ) + + assert result.index.tolist() == sampled_tours + assert result["person_id"].tolist() == [7, 7, 8][: len(sampled_tours)] + assert result["alt_dest"].notna().all() + + +class _SizeTerms: + def __init__(self, state, size_term_selector): + self.destination_size_terms = pd.DataFrame( + {"shopping": [1.0, 1.0]}, index=pd.Index([101, 202], name="zone_id") + ) + + def dest_size_terms_df(self, segment_name, trace_label): + return self.destination_size_terms[[segment_name]].rename( + columns={segment_name: "size_term"} + ) + + +@pytest.mark.parametrize( + "sampled_tours", [[], [2, 2]], ids=["all_failed", "one_failed"] +) +def test_run_tour_destination_skips_failed_samples(monkeypatch, sampled_tours): + """Tours without a sample skip logsums and simulation; their households are recorded. + + The sample choosers carry no household_id, so sampling cannot record them. + """ + tours = pd.DataFrame( + { + "household_id": [11, 12], + "person_id": [1, 2], + "tour_type": ["shopping", "shopping"], + }, + index=pd.Index([1, 2], name="tour_id"), + ) + sample = pd.DataFrame( + {"alt_dest": [101, 202][: len(sampled_tours)]}, + index=pd.Index(sampled_tours, name="tour_id", dtype="int64"), + ) + expected_tours = tours.loc[tours.index.isin(sampled_tours)] + logsums_model = Mock(return_value=sample.assign(mode_choice_logsum=0.0)) + simulate_model = Mock( + side_effect=lambda state, segment, choosers, *args, **kwargs: pd.DataFrame( + {"choice": 101}, index=choosers.index + ) + ) + monkeypatch.setattr(tour_destination, "SizeTermCalculator", _SizeTerms) + monkeypatch.setattr( + tour_destination, "run_destination_sample", Mock(return_value=sample) + ) + monkeypatch.setattr(tour_destination, "run_destination_logsums", logsums_model) + monkeypatch.setattr(tour_destination, "run_destination_simulate", simulate_model) + state = workflow.State().default_settings() + state.settings.skip_failed_choices = True + model_settings = SimpleNamespace( + SIZE_TERM_SELECTOR="non_mandatory", + SEGMENTS=["shopping"], + CHOOSER_SEGMENT_COLUMN_NAME="tour_type", + LOGSUM_SETTINGS="tour_mode_choice.yaml", + ALT_DEST_COL_NAME="alt_dest", + ) + + choices, _ = tour_destination.run_tour_destination( + state, + tours, + persons_merged=pd.DataFrame(), + want_logsums=False, + want_sample_table=False, + model_settings=model_settings, + network_los=None, + estimator=None, + trace_label="non_mandatory_tour_destination", + ) + + failed = tours.loc[~tours.index.isin(sampled_tours), "household_id"] + assert logit.skipped_household_ids(state) == set(failed) + assert choices.index.tolist() == expected_tours.index.tolist() + if expected_tours.empty: + logsums_model.assert_not_called() + simulate_model.assert_not_called() + else: + pd.testing.assert_frame_equal(simulate_model.call_args.args[2], expected_tours) diff --git a/activitysim/abm/test/test_misc/test_tour_destination_skip_failed.py b/activitysim/abm/test/test_misc/test_tour_destination_skip_failed.py new file mode 100644 index 000000000..5546a56a7 --- /dev/null +++ b/activitysim/abm/test/test_misc/test_tour_destination_skip_failed.py @@ -0,0 +1,175 @@ +from __future__ import annotations + +import pandas as pd +import pytest + +from activitysim.abm.models import ( + atwork_subtour_destination, + joint_tour_destination, + non_mandatory_destination, +) +from activitysim.core import logit, workflow +from activitysim.core.configuration.logit import TourLocationComponentSettings + + +def _tour_1_failed_sampling(state, tours, *args, **kwargs): + """Stand-in for run_tour_destination when tour 1 had no sample. + + The real function records the household of a tour that failed sampling + and returns choices for the other tours only. + """ + logit.record_skipped_households(state, tours.loc[[1], "household_id"], "sample") + choices = pd.DataFrame( + {"choice": [202], "logsum": [1.5]}, index=pd.Index([2], name="tour_id") + ) + return choices, None + + +@pytest.mark.parametrize( + "step_module, step_name, tour_category, logsum_column", + [ + ( + joint_tour_destination, + "joint_tour_destination", + "joint", + "destination_logsum", + ), + ( + non_mandatory_destination, + "non_mandatory_tour_destination", + "non_mandatory", + None, + ), + ( + non_mandatory_destination, + "non_mandatory_tour_destination", + "non_mandatory", + "destination_logsum", + ), + ], + ids=["joint_with_logsums", "non_mandatory", "non_mandatory_with_logsums"], +) +def test_destination_step_drops_tours_of_skipped_households( + monkeypatch, tmp_path, step_module, step_name, tour_category, logsum_column +): + """A tour skipped during sampling has no destination; add_table drops it. + + Before this was handled, joint_tour_destination saved the tours table + before assigning logsums, and assign_in_place then failed on the dropped + tour. non_mandatory_tour_destination asserted that every tour has a + destination. + """ + (tmp_path / "configs").mkdir() + (tmp_path / "data").mkdir() + state = workflow.State().default_settings() + state.initialize_filesystem(working_dir=tmp_path, configs_dir=("configs",)) + state.settings.skip_failed_choices = True + tours = pd.DataFrame( + { + "household_id": [11, 12], + "person_id": [1, 2], + "tour_category": [tour_category, tour_category], + "destination": [-1, -1], + }, + index=pd.Index([1, 2], name="tour_id"), + ) + state.add_table("tours", tours) + monkeypatch.setattr( + step_module.tour_destination, "run_tour_destination", _tour_1_failed_sampling + ) + model_settings = TourLocationComponentSettings.model_construct( + DEST_CHOICE_LOGSUM_COLUMN_NAME=logsum_column, + DEST_CHOICE_SAMPLE_TABLE_NAME=None, + ALT_DEST_COL_NAME="alt_dest", + ) + + getattr(step_module, step_name)( + state, + tours=state.get_dataframe("tours"), + persons_merged=pd.DataFrame(), + network_los=None, + model_settings=model_settings, + ) + + result = state.get_dataframe("tours") + assert result.index.tolist() == [2] + assert result.loc[2, "destination"] == 202 + if logsum_column: + assert result.loc[2, logsum_column] == 1.5 + + +def _all_tours_failed_sampling(state, tours, *args, **kwargs): + """Stand-in for run_tour_destination when every tour had no sample. + + The real function records the households and returns no choices and no + sample table, even if sample tables are wanted. + """ + logit.record_skipped_households(state, tours["household_id"], "sample") + return pd.DataFrame(columns=["choice", "logsum"]), None + + +@pytest.mark.parametrize( + "step_module, step_name, tour_category", + [ + (joint_tour_destination, "joint_tour_destination", "joint"), + ( + non_mandatory_destination, + "non_mandatory_tour_destination", + "non_mandatory", + ), + (atwork_subtour_destination, "atwork_subtour_destination", "atwork"), + ], + ids=["joint", "non_mandatory", "atwork"], +) +def test_destination_step_without_sample_table_when_all_sampling_failed( + monkeypatch, tmp_path, step_module, step_name, tour_category +): + """With sample tables wanted, a step whose tours all failed sampling saves none. + + A mandatory tour of another household survives, so the run continues + past the step. Before this was handled, the steps read the index of the + missing sample table. + """ + (tmp_path / "configs").mkdir() + (tmp_path / "data").mkdir() + state = workflow.State().default_settings() + state.initialize_filesystem(working_dir=tmp_path, configs_dir=("configs",)) + state.settings.skip_failed_choices = True + state.settings.want_dest_choice_sample_tables = True + tours = pd.DataFrame( + { + "household_id": [11, 12], + "person_id": [1, 2], + "tour_category": [tour_category, "mandatory"], + "destination": [-1, 20], + }, + index=pd.Index([1, 2], name="tour_id"), + ) + state.add_table("tours", tours) + monkeypatch.setattr( + step_module.tour_destination, + "run_tour_destination", + _all_tours_failed_sampling, + ) + model_settings = TourLocationComponentSettings.model_construct( + DEST_CHOICE_LOGSUM_COLUMN_NAME=None, + DEST_CHOICE_SAMPLE_TABLE_NAME="destination_sample", + ALT_DEST_COL_NAME="alt_dest", + # atwork_subtour_destination warns and fills these in if missing + SIZE_TERM_SELECTOR="atwork", + SEGMENTS=["atwork"], + ORIG_ZONE_ID="workplace_zone_id", + ) + + getattr(step_module, step_name)( + state, + tours=state.get_dataframe("tours"), + persons_merged=pd.DataFrame(), + network_los=None, + model_settings=model_settings, + ) + + result = state.get_dataframe("tours") + assert result.index.tolist() == [2] + assert result.loc[2, "destination"] == 20 + assert not state.is_table("destination_sample") diff --git a/activitysim/abm/test/test_misc/test_tour_od_sampling.py b/activitysim/abm/test/test_misc/test_tour_od_sampling.py index 5383e524e..6573c6187 100644 --- a/activitysim/abm/test/test_misc/test_tour_od_sampling.py +++ b/activitysim/abm/test/test_misc/test_tour_od_sampling.py @@ -1,12 +1,14 @@ from __future__ import annotations from types import SimpleNamespace +from unittest.mock import Mock import numpy as np import pandas as pd +import pytest from activitysim.abm.models.util import tour_od -from activitysim.core import workflow +from activitysim.core import logit, workflow class _DummySkimDict: @@ -168,6 +170,69 @@ def fake_choose_maz_for_taz( ) +@pytest.mark.parametrize("use_explicit_error_terms", [False, True], ids=["mnl", "eet"]) +def test_od_presample_returns_empty_sample_when_every_chooser_failed( + monkeypatch, use_explicit_error_terms +): + """An empty OD sample gives an empty result with the usual columns. + + Under skip_failed_choices, the OD sample is empty when every chooser in the + segment failed sampling. choose_MAZ_for_TAZ raises on an empty sample. + """ + alt_od_col_name = tour_od.get_od_id_col(tour_od.ORIG_MAZ, tour_od.DEST_TAZ) + + def fake_od_sample(*_args, **_kwargs): + # interaction_sample's result when every chooser failed + return pd.DataFrame( + { + alt_od_col_name: pd.Series(dtype="object"), + "prob": pd.Series(dtype="float32"), + "pick_count": pd.Series(dtype="uint32"), + }, + index=pd.Index([], name="tour_id", dtype="int64"), + ) + + monkeypatch.setattr(tour_od, "_od_sample", fake_od_sample) + + state = workflow.State().default_settings() + state.settings.skip_failed_choices = True + state.settings.use_explicit_error_terms = use_explicit_error_terms + choosers = pd.DataFrame( + {tour_od.ORIG_TAZ: [1, 3]}, + index=pd.Index([7001, 7002], name="tour_id"), + ) + model_settings = type( + "ModelSettings", + (), + { + "ALT_DEST_COL_NAME": "alt_dest", + "CHOOSER_ORIG_COL_NAME": "origin", + }, + )() + network_los = _DummyNetworkLos({101: 1, 102: 2, 103: 3}) + destination_size_terms = pd.DataFrame( + {"size_term": [1.0, 2.0, 3.0]}, + index=pd.Index([101, 102, 103], name="zone_id"), + ) + + out = tour_od.od_presample( + state, + "segment", + choosers, + model_settings, + network_los, + destination_size_terms, + destination_size_terms, + estimator=None, + chunk_size=0, + trace_label="test_trace", + ) + + assert out.empty + assert out.index.name == "tour_id" + assert list(out.columns) == ["alt_dest", "origin", "prob", "pick_count"] + + def test_od_presample_omits_full_taz_index_for_eet_non_poisson(monkeypatch): captured = {} @@ -316,3 +381,88 @@ def test_choose_maz_for_taz_eet_uses_full_taz_positions_with_origin_key(): ), ) assert state.get_rn_generator().calls == [5] + + +class _SizeTerms: + def __init__(self, state, size_term_selector): + self.destination_size_terms = pd.DataFrame( + {"work": [1.0, 1.0]}, index=pd.Index([101, 202], name="zone_id") + ) + + def dest_size_terms_df(self, segment_name, trace_label): + return self.destination_size_terms[[segment_name]].rename( + columns={segment_name: "size_term"} + ) + + +@pytest.mark.parametrize( + "sampled_tours", [[], [2, 2]], ids=["all_failed", "one_failed"] +) +def test_run_tour_od_skips_failed_samples(monkeypatch, sampled_tours): + """Tours without an OD sample skip logsums and simulation; their households are recorded. + + With every tour failed, run_tour_od returns an empty choice table instead of + raising UnboundLocalError. + """ + tours = pd.DataFrame( + {"household_id": [11, 12], "person_id": [1, 2], "tour_type": ["work", "work"]}, + index=pd.Index([1, 2], name="tour_id"), + ) + persons = pd.DataFrame( + {"is_university": [False, False], "demographic_segment": [0, 0]}, + index=pd.Index([1, 2], name="person_id"), + ) + sample = pd.DataFrame( + {"alt_od": ["101_1", "202_2"][: len(sampled_tours)]}, + index=pd.Index(sampled_tours, name="tour_id", dtype="int64"), + ) + expected_tours = tours.loc[tours.index.isin(sampled_tours)] + logsums_model = Mock(return_value=sample.assign(tour_mode_choice_logsum=0.0)) + simulate_model = Mock( + side_effect=lambda state, segment, choosers, *args, **kwargs: pd.DataFrame( + {"choice": "101_1", "origin": 1, "destination": 101}, index=choosers.index + ) + ) + monkeypatch.setattr(tour_od, "SizeTermCalculator", _SizeTerms) + monkeypatch.setattr( + tour_od.expressions, "annotate_preprocessors", lambda *args, **kwargs: None + ) + monkeypatch.setattr(tour_od, "run_od_sample", Mock(return_value=sample)) + monkeypatch.setattr(tour_od, "run_od_logsums", logsums_model) + monkeypatch.setattr(tour_od, "run_od_simulate", simulate_model) + state = workflow.State().default_settings() + state.settings.skip_failed_choices = True + model_settings = SimpleNamespace( + SIZE_TERM_SELECTOR="od", + ORIG_COL_NAME="origin", + DEST_COL_NAME="destination", + CHOOSER_SEGMENT_COLUMN_NAME="tour_type", + SEGMENTS=["work"], + ORIG_FILTER="original_MAZ > 0", + LOGSUM_SETTINGS="tour_mode_choice.yaml", + ALT_DEST_COL_NAME="alt_od", + ) + + choices, _ = tour_od.run_tour_od( + state, + tours, + persons, + want_logsums=False, + want_sample_table=False, + model_settings=model_settings, + network_los=None, + estimator=None, + chunk_size=0, + trace_hh_id=None, + trace_label="tour_od_choice", + ) + + failed = tours.loc[~tours.index.isin(sampled_tours), "household_id"] + assert logit.skipped_household_ids(state) == set(failed) + assert choices.index.tolist() == expected_tours.index.tolist() + assert {"choice", "origin", "destination"} <= set(choices.columns) + if expected_tours.empty: + logsums_model.assert_not_called() + simulate_model.assert_not_called() + else: + assert simulate_model.call_args.args[2].index.tolist() == [2] diff --git a/activitysim/core/interaction_sample.py b/activitysim/core/interaction_sample.py index d8bfe945d..99c970b48 100644 --- a/activitysim/core/interaction_sample.py +++ b/activitysim/core/interaction_sample.py @@ -394,7 +394,7 @@ def make_sample_choices( assert isinstance(alternatives, pd.DataFrame) assert len(alternatives) == alternative_count - if allow_zero_probs: + if allow_zero_probs or state.settings.skip_failed_choices: zero_probs = probs.sum(axis=1) == 0 if zero_probs.all(): return pd.DataFrame( @@ -897,7 +897,7 @@ def _interaction_sample( else: # eet and poisson: optionally trim choosers with all-zero probs. The # inverse-CDF path handles this inside make_sample_choices - if allow_zero_probs: + if allow_zero_probs or state.settings.skip_failed_choices: non_zero = probs.sum(axis=1) != 0 if not non_zero.any(): return pd.DataFrame( @@ -1159,9 +1159,13 @@ def interaction_sample( # http://pandas.pydata.org/pandas-docs/stable/io.html#id2 if len(result_list) > 1: choices = pd.concat(result_list) + elif result_list: + choices = result_list[0] - assert allow_zero_probs or ( - len(choosers.index) == len(np.unique(choices.index.values)) + assert ( + allow_zero_probs + or state.settings.skip_failed_choices + or len(choosers.index) == len(np.unique(choices.index.values)) ) # keep alts in canonical order so choices based on their probs are stable across runs diff --git a/activitysim/core/interaction_sample_simulate.py b/activitysim/core/interaction_sample_simulate.py index ee052caa6..cbdfd0d34 100644 --- a/activitysim/core/interaction_sample_simulate.py +++ b/activitysim/core/interaction_sample_simulate.py @@ -346,10 +346,7 @@ def _interaction_sample_simulate( ) if allow_zero_probs: - zero_probs = ( - utilities_df.sum(axis=1) - <= utilities_df.shape[1] * logit.UTIL_UNAVAILABLE - ) + zero_probs = logit.no_available_alternatives(utilities_df) if zero_probs.any(): # copied from proabability below, fix when that gets fixed # FIXME this is kind of gnarly, but we force choice of first alt @@ -543,6 +540,11 @@ def interaction_sample_simulate( when household tracing enabled. No tracing occurs if label is empty or None. trace_choice_name: str This is the column label to be used in trace file csv dump of choices + allow_zero_probs : bool, optional + Accept choosers for which no sampled alternative is available. + zero_prob_choice_val : optional + Choice value to return for those choosers. Required when allow_zero_probs + is True, unless skip_choice is True. skip_choice: bool This skips the logit choice step and simply returns the alternatives table with logsums (used in disaggregate accessibility) @@ -575,6 +577,15 @@ def interaction_sample_simulate( trace_label = tracing.extend_trace_label(trace_label, "interaction_sample_simulate") chunk_tag = chunk_tag or trace_label + # Choosers with no available alternative get their first alternative as a + # placeholder, and only zero_prob_choice_val replaces it. Without a value they + # would silently keep an alternative they cannot choose. Logsum-only calls + # (skip_choice) make no choice. + if allow_zero_probs and zero_prob_choice_val is None and not skip_choice: + raise ValueError( + f"{trace_label}: zero_prob_choice_val is required when allow_zero_probs is True" + ) + # Note: when use_explicit_error_terms is True but alts_context is None, EET draws are # keyed to the per-call active alternative count rather than a stable universe, so they # are NOT guaranteed to be consistent across scenarios that differ in alternative diff --git a/activitysim/core/interaction_simulate.py b/activitysim/core/interaction_simulate.py index d07be9407..58112fd2c 100644 --- a/activitysim/core/interaction_simulate.py +++ b/activitysim/core/interaction_simulate.py @@ -924,9 +924,7 @@ def _interaction_simulate( trace_choosers=choosers, ) if allow_zero_probs: - zero_probs = ( - utilities.sum(axis=1) <= utilities.shape[1] * logit.UTIL_UNAVAILABLE - ) + zero_probs = logit.no_available_alternatives(utilities) if zero_probs.any(): # Give unavailable rows a temporary choice before replacing it below. utilities.loc[zero_probs, 0] = logit.UTIL_LARGE_ENOUGH @@ -981,6 +979,13 @@ def _interaction_simulate( # tranche of this choosers alternatives created by cross join of alternatives and choosers # offsets is the offset into model_design df of first row of chooser alternatives offsets = np.arange(len(positions)) * sample_size + + # make_choices marks failed choices with -99 when skip_failed_choices is + # enabled; clip so take() below cannot wrap into another chooser's + # alternatives (the failed chooser's household is skipped anyway) + if state.settings.skip_failed_choices: + positions = positions.clip(lower=0) + # resulting Int64Index has one element per chooser row and is in same order as choosers choices = interaction_utilities.index.take(positions + offsets) @@ -1066,7 +1071,8 @@ def interaction_simulate( allow_zero_probs : bool, optional Accept choosers for which every alternative has zero probability. zero_prob_choice_val : optional - Choice value to return for those choosers when allow_zero_probs is True. + Choice value to return for those choosers. Required when allow_zero_probs + is True. Returns ------- @@ -1077,6 +1083,14 @@ def interaction_simulate( trace_label = tracing.extend_trace_label(trace_label, "interaction_simulate") + # Choosers with no available alternative get their first alternative as a + # placeholder, and only zero_prob_choice_val replaces it. Without a value they + # would silently keep an alternative they cannot choose. + if allow_zero_probs and zero_prob_choice_val is None: + raise ValueError( + f"{trace_label}: zero_prob_choice_val is required when allow_zero_probs is True" + ) + assert len(choosers) > 0 result_list = [] diff --git a/activitysim/core/logit.py b/activitysim/core/logit.py index b3b739f82..77067f94f 100644 --- a/activitysim/core/logit.py +++ b/activitysim/core/logit.py @@ -67,6 +67,43 @@ def n_alts_to_cover_max_id(self) -> int: return self.max_alt_id + 1 +def skipped_household_ids(state: workflow.State) -> set: + """Household ids recorded as skipped under ``skip_failed_choices`` so far.""" + import itertools + + return set( + itertools.chain.from_iterable( + state.get("skipped_household_ids", dict()).values() + ) + ) + + +def record_skipped_households(state: workflow.State, household_ids, trace_label) -> int: + """ + Record households to skip under ``skip_failed_choices``. + + ``household_ids`` may contain missing values and households already recorded; + both are ignored. Returns the total number of skipped households. + """ + num_skipped_households = state.get("num_skipped_households", 0) + skipped = state.get("skipped_household_ids", dict()) + + new_ids = set(pd.Series(household_ids).dropna().unique()) - skipped_household_ids( + state + ) + if new_ids: + skipped.setdefault(trace_label, []).extend(new_ids) + num_skipped_households += len(new_ids) + + # make sure the number of skipped households is consistent with the ids recorded + assert num_skipped_households == sum( + len(hh_list) for hh_list in skipped.values() + ), "Inconsistent number of skipped households and recorded ids" + state.set("num_skipped_households", num_skipped_households) + state.set("skipped_household_ids", skipped) + return num_skipped_households + + def report_bad_choices( state: workflow.State, bad_row_map, @@ -130,40 +167,11 @@ 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", dict()) - - # Get current household IDs and filter out None values - current_hh_ids = set(df[trace_col].dropna().unique()) - - # Get all previously skipped household IDs across all trace_labels - import itertools - - already_skipped = set( - itertools.chain.from_iterable(skipped_household_ids.values()) + num_skipped_households = record_skipped_households( + state, df[trace_col], trace_label ) - - # Find truly new household IDs that haven't been skipped before - new_skipped_hh_ids = current_hh_ids - already_skipped - - # Only process if there are new households to skip - if new_skipped_hh_ids: - # Initialize list for this trace_label if it doesn't exist - if trace_label not in skipped_household_ids: - skipped_household_ids[trace_label] = [] - skipped_household_ids[trace_label].extend(new_skipped_hh_ids) - num_skipped_households += len(new_skipped_hh_ids) - - # make sure the number of skipped households is consistent with the ids recorded - assert num_skipped_households == sum( - len(hh_list) for hh_list in skipped_household_ids.values() - ), "Inconsistent number of skipped households and recorded ids" - state.set("num_skipped_households", num_skipped_households) - state.set("skipped_household_ids", skipped_household_ids) - logger.warning( - f"Skipping {bad_row_map.sum()} bad choices. Total skipped households so far: {num_skipped_households}. Skipped household IDs: {skipped_household_ids}" + f"Skipping {bad_row_map.sum()} bad choices. Total skipped households so far: {num_skipped_households}. Skipped household IDs: {state.get('skipped_household_ids', dict())}" ) return @@ -209,6 +217,18 @@ def utils_to_logsums(utils, exponentiated=False, allow_zero_probs=False): return logsums +def no_available_alternatives(utils: pd.DataFrame) -> pd.Series: + """ + Flag rows in which no alternative is available. + + An alternative is available when its utility exceeds UTIL_MIN, the threshold + validate_utils uses before setting utilities to UTIL_UNAVAILABLE. Testing the + threshold directly is exact for any dtype; comparing a row sum with a + multiple of UTIL_UNAVAILABLE is not. + """ + return pd.Series(~(utils.to_numpy() > UTIL_MIN).any(axis=1), index=utils.index) + + def validate_utils( state: workflow.State, utils, diff --git a/activitysim/core/test/test_interaction_sample.py b/activitysim/core/test/test_interaction_sample.py index 9efae1f53..3dcf0da19 100644 --- a/activitysim/core/test/test_interaction_sample.py +++ b/activitysim/core/test/test_interaction_sample.py @@ -8,6 +8,7 @@ import numpy as np import pandas as pd import pytest +import sharrow as sh from activitysim.core import interaction_sample, workflow from activitysim.core.configuration.base import ComputeSettings @@ -20,6 +21,177 @@ def state() -> workflow.State: return state +@pytest.mark.parametrize("sample_method", [None, "eet", "inverse_cdf"]) +@pytest.mark.parametrize("all_failed", [False, True]) +@pytest.mark.parametrize("explicit_chunk_size", [0, 1]) +def test_interaction_sample_skips_underflowed_probs( + state, tmp_path, sample_method, all_failed, explicit_chunk_size +): + (tmp_path / "configs").mkdir() + (tmp_path / "data").mkdir() + state.initialize_filesystem(working_dir=tmp_path, configs_dir=("configs",)) + state.filesystem.sharrow_cache_dir = tmp_path / "sharrow-cache" + state.set("skim_dataset", sh.Dataset()) + state.settings.sharrow = True + state.settings.chunk_training_mode = "explicit" + state.settings.skip_failed_choices = True + state.settings.use_explicit_error_terms = True + choosers = pd.DataFrame( + { + "household_id": [101, 102, 103], + # at or below UTIL_MIN: no alternative is available, in any dtype + "utility": [-1000.0 if all_failed else 0.0, -1000.0, -1000.0], + }, + index=pd.Index([1, 2, 3], name="person_id"), + ) + alternatives = pd.DataFrame( + {"alt_utility": [0.0, -1.0]}, + index=pd.Index([10, 20], name="alt_id"), + ) + spec = pd.DataFrame( + {"coefficient": [1.0, 1.0]}, + index=pd.Index(["utility", "alt_utility"], name="Expression"), + ) + state.rng().add_channel("person_id", choosers) + state.rng().begin_step("test_underflowed_probs") + + sample = interaction_sample.interaction_sample( + state, + choosers, + alternatives, + spec, + sample_size=2, + alt_col_name="alt_id", + explicit_chunk_size=explicit_chunk_size, + compute_settings=ComputeSettings(sample_method=sample_method), + stable_alt_positions=np.array([0, 2]), + n_total_alts=3, + ) + + assert sample.index.name == "person_id" + assert set(sample.index) == (set() if all_failed else {1}) + assert (sample["prob"] > 0).all() + assert sample["alt_id"].isin(alternatives.index).all() + assert (sample["pick_count"] > 0).all() + skipped_households = { + household_id + for ids in state.get("skipped_household_ids").values() + for household_id in ids + } + assert skipped_households == ({101, 102, 103} if all_failed else {102, 103}) + assert state.get("num_skipped_households") == len(skipped_households) + + +@pytest.fixture(scope="module") +def sharrow_cache_dir(tmp_path_factory): + # shared so the flow for the spec below compiles once per module + return tmp_path_factory.mktemp("sharrow-cache") + + +# Utility of every alternative of the two "poor" choosers. -120 is low but +# available: exp() underflows only in float32. -1000 is at or below UTIL_MIN, +# so no alternative is available in any dtype. +LOW_UTILITY = -120.0 +UNAVAILABLE_UTILITY = -1000.0 + + +@pytest.mark.parametrize( + "utility", [LOW_UTILITY, UNAVAILABLE_UTILITY], ids=["low", "unavailable"] +) +@pytest.mark.parametrize( + "allow_zero_probs", [False, True], ids=["location", "trip_dest"] +) +@pytest.mark.parametrize("skip_failed_choices", [True, False], ids=["skip", "no_skip"]) +@pytest.mark.parametrize( + "sample_method, use_explicit_error_terms", + [("inverse_cdf", False), ("poisson", True)], + ids=["mnl", "eet"], +) +@pytest.mark.parametrize("sharrow", [False, True], ids=["float64", "float32"]) +def test_interaction_sample_zero_probability_outcomes( + request, + tmp_path, + sharrow_cache_dir, + sharrow, + sample_method, + use_explicit_error_terms, + skip_failed_choices, + allow_zero_probs, + utility, +): + """Which choosers are sampled, skipped or dropped, by utility level. + + Sharrow evaluates utilities as float32, plain evaluation as float64. + Overflow protection (the row shift) is on only when neither + skip_failed_choices nor allow_zero_probs turns it off. + """ + unavailable = utility == UNAVAILABLE_UTILITY + shifted = not skip_failed_choices and not allow_zero_probs + if sharrow and not unavailable and not shifted: + request.applymarker( + pytest.mark.xfail( + strict=True, + reason="unshifted float32 exp() underflows below about -103, so " + "an available chooser fails (#1115, #1129)", + ) + ) + + (tmp_path / "configs").mkdir() + (tmp_path / "data").mkdir() + state = workflow.State().default_settings() + state.initialize_filesystem(working_dir=tmp_path, configs_dir=("configs",)) + state.filesystem.sharrow_cache_dir = sharrow_cache_dir + state.set("skim_dataset", sh.Dataset()) + state.settings.check_for_variability = False + state.settings.sharrow = sharrow + state.settings.use_explicit_error_terms = use_explicit_error_terms + state.settings.skip_failed_choices = skip_failed_choices + + choosers = pd.DataFrame( + {"household_id": [101, 102, 103], "utility": [0.0, utility, utility]}, + index=pd.Index([1, 2, 3], name="person_id"), + ) + alternatives = pd.DataFrame( + {"alt_utility": [0.0, -1.0]}, + index=pd.Index([10, 20], name="alt_id"), + ) + spec = pd.DataFrame( + {"coefficient": [1.0, 1.0]}, + index=pd.Index(["utility", "alt_utility"], name="Expression"), + ) + state.rng().add_channel("person_id", choosers) + state.rng().begin_step("test_zero_probability_outcomes") + + sample = interaction_sample.interaction_sample( + state, + choosers, + alternatives, + spec, + sample_size=2, + alt_col_name="alt_id", + allow_zero_probs=allow_zero_probs, + compute_settings=ComputeSettings(sample_method=sample_method), + ) + + if unavailable and not shifted: + # dropped from the sample; recorded as skipped only when it is a failure + expected_sampled = {1} + expected_skipped = set() if allow_zero_probs else {102, 103} + else: + # available choosers, or a forced choice under overflow protection + expected_sampled = {1, 2, 3} + expected_skipped = set() + assert set(sample.index) == expected_sampled + assert (sample["prob"] > 0).all() + assert sample["alt_id"].isin(alternatives.index).all() + skipped = { + household_id + for ids in state.get("skipped_household_ids", {}).values() + for household_id in ids + } + assert skipped == expected_skipped + + def test_interaction_sample_ignores_stable_positions_without_global_eet( state, monkeypatch ): diff --git a/activitysim/core/test/test_interaction_sample_simulate.py b/activitysim/core/test/test_interaction_sample_simulate.py index bb173eeca..6da8ce357 100644 --- a/activitysim/core/test/test_interaction_sample_simulate.py +++ b/activitysim/core/test/test_interaction_sample_simulate.py @@ -6,8 +6,10 @@ import numpy as np import pandas as pd import pytest +import sharrow as sh from activitysim.core import interaction_sample, interaction_sample_simulate, workflow +from activitysim.core.configuration.base import ComputeSettings from activitysim.core.logit import AltsContext @@ -214,6 +216,50 @@ def test_interaction_sample_simulate_eet_unavailable_alternatives(state): assert not choices_eet.isin([2, 3, 4]).any() +def _call_with_unavailable_chooser(state, **kwargs): + """Person 2 has no available alternative; person 1 has two at utility 0.""" + choosers = pd.DataFrame( + {"chooser_ok": [1.0, 0.0]}, + index=pd.Index([1, 2], name="person_id"), + ) + alternatives = pd.DataFrame( + {"alt_id": [1, 2, 1, 2]}, + index=pd.Index([1, 1, 2, 2], name="person_id"), + ) + spec = pd.DataFrame( + {"coefficient": [1.0]}, + index=pd.Index(["-1000 * (1 - chooser_ok)"], name="Expression"), + ) + state.rng().set_base_seed(42) + state.rng().add_channel("person_id", choosers) + state.rng().begin_step("test_zero_prob_choice_val") + + return interaction_sample_simulate.interaction_sample_simulate( + state, + choosers, + alternatives, + spec, + choice_column="alt_id", + allow_zero_probs=True, + **kwargs, + ) + + +def test_interaction_sample_simulate_allow_zero_probs_requires_choice_val(state): + """Without zero_prob_choice_val, a chooser with no available alternative would + silently keep its placeholder (first) alternative.""" + with pytest.raises(ValueError, match="zero_prob_choice_val is required"): + _call_with_unavailable_chooser(state) + + +def test_interaction_sample_simulate_skip_choice_needs_no_choice_val(state): + """Logsum-only calls make no choice, so they need no zero_prob_choice_val.""" + result = _call_with_unavailable_chooser(state, want_logsums=True, skip_choice=True) + + assert result["logsums"].iloc[0] == pytest.approx(np.log(2.0)) + assert result["logsums"].iloc[1] == -np.inf + + def test_interaction_sample_simulate_passes_alts_context_and_alt_nrs_df( state, monkeypatch ): @@ -280,3 +326,155 @@ def fake_make_choices_utility_based( index=choosers.index, ) pd.testing.assert_frame_equal(captured["alt_nrs_df"], expected_alt_nrs) + + +@pytest.fixture(scope="module") +def sharrow_cache_dir(tmp_path_factory): + # shared so the flow for the spec below compiles once per module + return tmp_path_factory.mktemp("sharrow-cache") + + +# low but available: exp() underflows only in float32 +LOW_UTILITY = -120.0 + + +def _simulate_two_trips( + tmp_path, + sharrow_cache_dir, + utilities_a, + utilities_b, + *, + sharrow, + use_explicit_error_terms, + allow_zero_probs, + want_logsums=False, +): + """Trips 1 and 2 choose among sampled alternatives 1..n with the given utilities. + + Sharrow evaluates the utilities as float32, plain evaluation as float64. + """ + (tmp_path / "configs").mkdir() + (tmp_path / "data").mkdir() + state = workflow.State().default_settings() + state.initialize_filesystem(working_dir=tmp_path, configs_dir=("configs",)) + state.filesystem.sharrow_cache_dir = sharrow_cache_dir + state.set("skim_dataset", sh.Dataset()) + state.settings.check_for_variability = False + state.settings.sharrow = sharrow + state.settings.use_explicit_error_terms = use_explicit_error_terms + + choosers = pd.DataFrame( + {"household_id": [101, 102], "chooser_util": [0.0, 0.0]}, + index=pd.Index([1, 2], name="trip_id"), + ) + alternatives = pd.DataFrame( + { + "alt_dest": [ + *range(1, len(utilities_a) + 1), + *range(1, len(utilities_b) + 1), + ], + "alt_util": [*utilities_a, *utilities_b], + }, + index=pd.Index([1] * len(utilities_a) + [2] * len(utilities_b), name="trip_id"), + ) + spec = pd.DataFrame( + {"coefficient": [1.0, 1.0]}, + index=pd.Index(["alt_util", "chooser_util"], name="Expression"), + ) + state.rng().add_channel("trip_id", choosers) + state.rng().begin_step("test_sample_simulate_zero_probs") + + result = interaction_sample_simulate.interaction_sample_simulate( + state, + choosers, + alternatives, + spec, + choice_column="alt_dest", + allow_zero_probs=allow_zero_probs, + zero_prob_choice_val=-1, + want_logsums=want_logsums, + compute_settings=ComputeSettings(protect_columns=["alt_dest"]), + ) + return result, state + + +@pytest.mark.parametrize("num_alts", [2, 15, 20, 24, 30]) +@pytest.mark.parametrize("sharrow", [False, True], ids=["float64", "float32"]) +def test_interaction_sample_simulate_eet_flags_unavailable_rows( + tmp_path, sharrow_cache_dir, sharrow, num_alts +): + """Under EET, a trip with no available sampled alternative gets zero_prob_choice_val. + + This holds whatever the number of alternatives and the utility dtype. + """ + choices, state = _simulate_two_trips( + tmp_path, + sharrow_cache_dir, + [0.0] + [-1000.0] * (num_alts - 1), + [-1000.0] * num_alts, + sharrow=sharrow, + use_explicit_error_terms=True, + allow_zero_probs=True, + ) + + assert choices.tolist() == [1, -1] + assert state.get("num_skipped_households", 0) == 0 + + +@pytest.mark.parametrize("use_explicit_error_terms", [False, True], ids=["mnl", "eet"]) +@pytest.mark.parametrize("sharrow", [False, True], ids=["float64", "float32"]) +def test_interaction_sample_simulate_low_utilities_stay_available( + request, tmp_path, sharrow_cache_dir, sharrow, use_explicit_error_terms +): + """A trip whose sampled alternatives all have utility -120 still chooses one.""" + if sharrow and not use_explicit_error_terms: + request.applymarker( + pytest.mark.xfail( + strict=True, + reason="unshifted float32 exp() underflows below about -103, so " + "MNL returns zero_prob_choice_val for an available trip (#1115)", + ) + ) + + choices, state = _simulate_two_trips( + tmp_path, + sharrow_cache_dir, + [0.0, 0.0], + [LOW_UTILITY, LOW_UTILITY - 1.0], + sharrow=sharrow, + use_explicit_error_terms=use_explicit_error_terms, + allow_zero_probs=True, + ) + + assert choices.isin([1, 2]).all() + assert state.get("num_skipped_households", 0) == 0 + + +@pytest.mark.parametrize("use_explicit_error_terms", [False, True], ids=["mnl", "eet"]) +@pytest.mark.parametrize("sharrow", [False, True], ids=["float64", "float32"]) +def test_interaction_sample_simulate_logsums_of_low_utilities( + request, tmp_path, sharrow_cache_dir, sharrow, use_explicit_error_terms +): + """Destination logsums stay finite when every sampled alternative is low.""" + if sharrow: + request.applymarker( + pytest.mark.xfail( + strict=True, + reason="unshifted float32 exp() underflows below about -103, so " + "the logsum is -inf (#1115)", + ) + ) + + result, state = _simulate_two_trips( + tmp_path, + sharrow_cache_dir, + [0.0, 0.0], + [LOW_UTILITY, LOW_UTILITY], + sharrow=sharrow, + use_explicit_error_terms=use_explicit_error_terms, + allow_zero_probs=False, + want_logsums=True, + ) + + assert result["logsum"].iloc[1] == pytest.approx(np.log(2.0) + LOW_UTILITY) + assert state.get("num_skipped_households", 0) == 0 diff --git a/activitysim/core/test/test_interaction_simulate.py b/activitysim/core/test/test_interaction_simulate.py index e1108da41..6fa131197 100644 --- a/activitysim/core/test/test_interaction_simulate.py +++ b/activitysim/core/test/test_interaction_simulate.py @@ -277,3 +277,118 @@ def test_eval_interaction_utilities_locals_override_global_constants(tmp_path): ) np.testing.assert_allclose(utilities.utility.to_numpy(), df.distance_km.to_numpy()) + + +def test_interaction_simulate_skips_failed_choice_within_own_alternatives(state): + """A chooser with no available alternative is skipped without a bad lookup. + + make_choices marks the failed choice with -99; unclipped, take() wrapped + into another chooser's alternatives or raised IndexError for small chunks. + """ + state.settings.skip_failed_choices = True + choosers = pd.DataFrame( + {"household_id": [101, 102], "chooser_ok": [1.0, 0.0]}, + index=pd.Index([10, 11], name="person_id"), + ) + alternatives = pd.DataFrame( + {"alt_attr": [0.0, 0.0]}, + index=pd.Index([1, 2], name="alt_id"), + ) + # chooser 11: every alternative at -1000, so all probabilities are zero + spec = pd.DataFrame( + {"coefficient": [1.0]}, + index=pd.Index(["-1000 * (1 - chooser_ok)"], name="Expression"), + ) + state.rng().set_base_seed(42) + state.rng().add_channel("person_id", choosers) + state.rng().begin_step("test_skips_failed_choice") + + choices = interaction_simulate.interaction_simulate( + state, choosers, alternatives, spec, sample_size=2 + ) + + assert choices.index.tolist() == [10, 11] + assert choices.isin(alternatives.index).all() + assert state.get("num_skipped_households", 0) == 1 + assert { + household_id + for ids in state.get("skipped_household_ids").values() + for household_id in ids + } == {102} + + +@pytest.mark.parametrize("utility_dtype", [np.float64, np.float32]) +@pytest.mark.parametrize("num_alts", [2, 15, 20, 100]) +@pytest.mark.parametrize("use_explicit_error_terms", [False, True]) +def test_interaction_simulate_allow_zero_probs_flags_unavailable_rows( + state, monkeypatch, utility_dtype, num_alts, use_explicit_error_terms +): + """A chooser with no available alternative gets zero_prob_choice_val. + + This holds for any number of alternatives and either utility dtype + (Sharrow evaluates utilities as float32). Under explicit error terms, the + row-sum test this replaced missed some counts in float64 and most in float32. + """ + eval_interaction_utilities = interaction_simulate.eval_interaction_utilities + + def eval_utilities_as_dtype(*args, **kwargs): + utilities, trace_results = eval_interaction_utilities(*args, **kwargs) + return utilities.astype(utility_dtype), trace_results + + monkeypatch.setattr( + interaction_simulate, "eval_interaction_utilities", eval_utilities_as_dtype + ) + + # chooser 10 can choose alternative 1 only; chooser 11 has no available alternative + choosers = pd.DataFrame( + {"household_id": [101, 102], "chooser_ok": [1.0, 0.0]}, + index=pd.Index([10, 11], name="person_id"), + ) + alternatives = pd.DataFrame( + {"alt_ok": [1.0] + [0.0] * (num_alts - 1)}, + index=pd.Index(range(1, num_alts + 1), name="alt_id"), + ) + spec = pd.DataFrame( + {"coefficient": [1.0]}, + index=pd.Index(["-1000 * (1 - chooser_ok * alt_ok)"], name="Expression"), + ) + + state.settings.use_explicit_error_terms = use_explicit_error_terms + state.rng().set_base_seed(42) + state.rng().add_channel("person_id", choosers) + state.rng().begin_step("test_allow_zero_probs") + + choices = interaction_simulate.interaction_simulate( + state, + choosers, + alternatives, + spec, + sample_size=num_alts, + allow_zero_probs=True, + zero_prob_choice_val=-1, + ) + + assert choices.tolist() == [1, -1] + assert state.get("num_skipped_households", 0) == 0 + + +def test_interaction_simulate_allow_zero_probs_requires_choice_val(state): + """Without zero_prob_choice_val, a chooser with no available alternative would + silently keep its placeholder (first) alternative.""" + choosers = pd.DataFrame( + {"chooser_ok": [1.0, 0.0]}, + index=pd.Index([10, 11], name="person_id"), + ) + alternatives = pd.DataFrame( + {"alt_attr": [0.0, 0.0]}, + index=pd.Index([1, 2], name="alt_id"), + ) + spec = pd.DataFrame( + {"coefficient": [1.0]}, + index=pd.Index(["-1000 * (1 - chooser_ok)"], name="Expression"), + ) + + with pytest.raises(ValueError, match="zero_prob_choice_val is required"): + interaction_simulate.interaction_simulate( + state, choosers, alternatives, spec, sample_size=2, allow_zero_probs=True + ) diff --git a/activitysim/core/test/test_logit.py b/activitysim/core/test/test_logit.py index b98d225d2..e25f05291 100644 --- a/activitysim/core/test/test_logit.py +++ b/activitysim/core/test/test_logit.py @@ -1877,3 +1877,42 @@ def test_alts_context_from_num_alts( assert ctx.min_alt_id == expected_min assert ctx.max_alt_id == expected_max assert ctx.n_alts_to_cover_max_id == expected_n_cover + + +@pytest.mark.parametrize("dtype", [np.float32, np.float64]) +@pytest.mark.parametrize("num_alts", [1, 2, 15, 20, 24, 100]) +def test_no_available_alternatives_is_exact(dtype, num_alts): + """Rows are flagged exactly when every utility is at or below UTIL_MIN. + + The row-sum comparison with num_alts * UTIL_UNAVAILABLE that this replaced + missed some float64 counts and nearly every float32 row. + """ + utils = pd.DataFrame(np.full((3, num_alts), -1000.0, dtype=dtype)) + utils.iloc[0, 0] = 0.0 # one available alternative + utils.iloc[1, :] = -650.0 # all low, but above UTIL_MIN + expected = [False, False, True] + + assert logit.no_available_alternatives(utils).tolist() == expected + # also after validate_utils has replaced -1000 with UTIL_UNAVAILABLE + validated = logit.validate_utils(None, utils.copy(), allow_zero_probs=True) + assert logit.no_available_alternatives(validated).tolist() == expected + + +def test_record_skipped_households_ignores_duplicates_and_missing_ids(): + state = workflow.State().default_settings() + assert logit.skipped_household_ids(state) == set() + + assert logit.record_skipped_households(state, [101, 102, None], "first") == 2 + assert ( + logit.record_skipped_households( + state, pd.Series([102.0, 103.0, np.nan]), "second" + ) + == 3 + ) + assert logit.record_skipped_households(state, [], "third") == 3 + + assert logit.skipped_household_ids(state) == {101, 102, 103} + assert state.get("num_skipped_households") == 3 + assert { + label: set(ids) for label, ids in state.get("skipped_household_ids").items() + } == {"first": {101, 102}, "second": {103}} diff --git a/docs/dev-guide/changes.md b/docs/dev-guide/changes.md index 16361539b..edb0747bc 100644 --- a/docs/dev-guide/changes.md +++ b/docs/dev-guide/changes.md @@ -12,7 +12,34 @@ configurations or code to fail to run correctly. This section describes changes that are implemented in current development branch (i.e., the main branch on GitHub), but not yet released in a stable version of ActivitySim. -No unreleased changes are currently documented. + +### Skip Failed Choices Fixes + +This release fixes several failures in model runs with `skip_failed_choices` +enabled ([#1129](https://github.com/ActivitySim/activitysim/issues/1129), +[PR #1130](https://github.com/ActivitySim/activitysim/pull/1130)): + +- A chooser for which no alternative can be sampled is now skipped together + with its household, like other failed choices. Previously the failure was + logged, but the model step then crashed. This applies to work and school + location choice and to tour destination choice (non-mandatory, joint and + at-work), including steps in which every chooser fails while destination + sample tables are written. +- With `allow_zero_probs`, `interaction_simulate` and + `interaction_sample_simulate` now find choosers with no available + alternative by testing each utility against the availability threshold. The + previous comparison of row sums was not exact and could miss such choosers. + +Code that calls `interaction_simulate` or `interaction_sample_simulate` with +`allow_zero_probs=True` must now also pass `zero_prob_choice_val`, or a +`ValueError` is raised. Previously, choosers with no available alternative +silently kept an alternative they cannot choose. Logsum-only calls to +`interaction_sample_simulate` with `skip_choice=True` do not need a value. All +built-in models already pass one. + +A known limitation remains: because `skip_failed_choices` turns off overflow +protection, valid but very low utilities can still underflow to zero +probability with `float32` utilities, and the household is then skipped. ### Shadow Pricing Re-Simulation Segmentation @@ -154,27 +181,27 @@ to `True` in the `compute_settings` for each model component where it is desired ### Skipping Failed Choices -A new feature is introduced to ActivitySim to skip failed choices during model execution. -By default, ActivitySim will skip any failed choices during a model run, i.e., `skip_failed_choices` is set to `True`. -A failed choice occurs when the computed utilities for all alternatives are zero, or infinite, or nan, which can happen due to -data issues or model specification problems. A warning message is logged when a failed choice is encountered, +A new feature is introduced to ActivitySim to skip failed choices during model execution. +By default, ActivitySim will skip any failed choices during a model run, i.e., `skip_failed_choices` is set to `True`. +A failed choice occurs when the computed utilities for all alternatives are zero, or infinite, or nan, which can happen due to +data issues or model specification problems. A warning message is logged when a failed choice is encountered, and the corresponding household (along with its persons, vehicles, tours, trips, etc) will be excluded from further model steps. At the end of the model run, a summary of all skipped households is provided in the log file for user reference. Users can set -a threshold for the maximum allowed percentage of skipped households `fraction_of_failed_choices_allowed`, -and if the percentage of skipped households exceeds this threshold, -the model run will be terminated with an error. The default threshold is set to `0.001`, 0.1% of households. +a threshold for the maximum allowed percentage of skipped households `fraction_of_failed_choices_allowed`, +and if the percentage of skipped households exceeds this threshold, +the model run will be terminated with an error. The default threshold is set to `0.001`, 0.1% of households. -This feature helps to ensure that the model can continue running even in the presence of data or specification issues, +This feature helps to ensure that the model can continue running even in the presence of data or specification issues, while also providing visibility into such issues that need to be addressed. See more information in ActivitySim's users guide "Skip Failed Choices" and code updates in [PR #1023](https://github.com/ActivitySim/activitysim/pull/1023) #### Potential Impact on Existing Model Runs -With `skip_failed_choices` defaulted to `True` and an allowed failure threshold defaulted as 0.1% of households, -failures that were previously silently masked will now generate warnings, be explicitly skipped, and counted toward the threshold. -As a result, model runs that previously completed "successfully" may now fail earlier, +With `skip_failed_choices` defaulted to `True` and an allowed failure threshold defaulted as 0.1% of households, +failures that were previously silently masked will now generate warnings, be explicitly skipped, and counted toward the threshold. +As a result, model runs that previously completed "successfully" may now fail earlier, surfacing underlying specification or data issues that require attention rather than being silently absorbed. If an agency does not -want to address the underlying issues immediately, they can set `skip_failed_choices` to `False` to maintain the previous model results, +want to address the underlying issues immediately, they can set `skip_failed_choices` to `False` to maintain the previous model results, but it is recommended to review the log file for any warnings about failed choices and address them as soon as possible to ensure model quality. ### Shadow Price Zones Reopening