diff --git a/activitysim/abm/models/joint_tour_destination.py b/activitysim/abm/models/joint_tour_destination.py index 29b117821b..bcae1b4966 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,6 +103,10 @@ 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: diff --git a/activitysim/abm/models/location_choice.py b/activitysim/abm/models/location_choice.py index c955f4a606..f4320a0b16 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 b0c6ea3fb1..5f00b9b319 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,9 +123,12 @@ 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) diff --git a/activitysim/abm/models/util/tour_destination.py b/activitysim/abm/models/util/tour_destination.py index 11bdb8ecdb..2e90645a87 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 6626dccc25..8d56a750e8 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 9bcdd7711f..65a945c061 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 6b21e0329c..262781c793 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 2157fdcadb..8df3a3273d 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 0000000000..eee8fa6962 --- /dev/null +++ b/activitysim/abm/test/test_misc/test_tour_destination_skip_failed.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +import pandas as pd +import pytest + +from activitysim.abm.models import 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 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 5383e524e5..6573c6187f 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 d8bfe945db..99c970b484 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 ee052caa63..cbdfd0d345 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 d07be94070..58112fd2c9 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 b3b739f827..77067f94f4 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 9efae1f53d..3dcf0da196 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 bb173eecad..6da8ce357e 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 e1108da415..6fa131197c 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 b98d225d29..e25f052919 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}}