diff --git a/activitysim/abm/models/__init__.py b/activitysim/abm/models/__init__.py index 1fc5ff2825..056659b5e1 100644 --- a/activitysim/abm/models/__init__.py +++ b/activitysim/abm/models/__init__.py @@ -31,6 +31,8 @@ school_escorting, stop_frequency, summarize, + telework_arrangement, + telework_duration, telecommute_frequency, telecommute_status, tour_mode_choice, diff --git a/activitysim/abm/models/settings_checker.py b/activitysim/abm/models/settings_checker.py index c65d0f5772..c89fe1cfd1 100644 --- a/activitysim/abm/models/settings_checker.py +++ b/activitysim/abm/models/settings_checker.py @@ -55,6 +55,8 @@ from activitysim.abm.models.stop_frequency import StopFrequencySettings from activitysim.abm.models.summarize import SummarizeSettings from activitysim.abm.models.telecommute_frequency import TelecommuteFrequencySettings +from activitysim.abm.models.telework_arrangement import TeleworkArrangementSettings +from activitysim.abm.models.telework_duration import TeleworkDurationSettings from activitysim.abm.models.tour_scheduling_probabilistic import ( TourSchedulingProbabilisticSettings, ) @@ -239,6 +241,14 @@ def _construct_message(self) -> str: "settings_cls": TelecommuteFrequencySettings, "settings_file": "telecommute_frequency.yaml", }, + "telework_arrangement": { + "settings_cls": TeleworkArrangementSettings, + "settings_file": "telework_arrangement.yaml", + }, + "telework_duration": { + "settings_cls": TeleworkDurationSettings, + "settings_file": "telework_duration.yaml", + }, "tour_mode_choice_simulate": { "settings_cls": TourModeComponentSettings, "settings_file": "tour_mode_choice.yaml", diff --git a/activitysim/abm/models/telework_arrangement.py b/activitysim/abm/models/telework_arrangement.py new file mode 100644 index 0000000000..823749e7ac --- /dev/null +++ b/activitysim/abm/models/telework_arrangement.py @@ -0,0 +1,151 @@ +# ActivitySim +# See full license in LICENSE.txt. +from __future__ import annotations + +import logging + +import pandas as pd + +from activitysim.core import ( + config, + estimation, + expressions, + simulate, + tracing, + workflow, +) +from activitysim.core.configuration.logit import LogitComponentSettings + +logger = logging.getLogger("activitysim") + + +class TeleworkArrangementSettings(LogitComponentSettings, extra="forbid"): + """ + Settings for the `telework_arrangement` component. + """ + + CHOOSER_FILTER_COLUMN_NAME: str = "is_worker" + """Column name in the dataframe to represent worker.""" + + HAS_IN_HOME_WORK_ACTIVITY_ALT: int = 0 + """The alternative index for having in-home work activity on the simulation day.""" + + +@workflow.step +def telework_arrangement( + state: workflow.State, + persons_merged: pd.DataFrame, + persons: pd.DataFrame, + model_settings: TeleworkArrangementSettings | None = None, + model_settings_file_name: str = "telework_arrangement.yaml", + trace_label: str = "telework_arrangement", +) -> None: + """ + This model predicts the telework arrangement on the simulation day for all workers. + The alternatives are whether or not a worker has in-home telework activities on the simulation day: + The result is a new column in the persons table, "has_in_home_work_activity": True or False + + Parameters + ---------- + state : workflow.State + persons_merged : DataFrame + This represents the 'choosers' table for this component. + persons : DataFrame + The original persons table is referenced so the telework arrangement column + can be appended to it. + model_settings : TeleworkArrangementSettings, optional + The settings used in this model component. If not provided, they are + loaded out of the configs directory YAML file referenced by + the `model_settings_file_name` argument. + model_settings_file_name : str, default "telework_arrangement.yaml" + This is where model setting are found if `model_settings` is not given + explicitly. The same filename is also used to write settings files to + the estimation data bundle in estimation mode. + trace_label : str, default "telework_arrangement" + This label is used for various tracing purposes. + """ + + if model_settings is None: + model_settings = TeleworkArrangementSettings.read_settings_file( + state.filesystem, + model_settings_file_name, + ) + + chooser_filter_column_name = model_settings.CHOOSER_FILTER_COLUMN_NAME + choosers = persons_merged[persons_merged[chooser_filter_column_name]] + + logger.info("Running %s with %d persons", trace_label, len(choosers)) + + estimator = estimation.manager.begin_estimation(state, "telework_arrangement") + + constants = config.get_model_constants(model_settings) + + expressions.annotate_preprocessors( + state, + df=choosers, + locals_dict=constants, + skims=None, + model_settings=model_settings, + trace_label=trace_label, + ) + + model_spec = state.filesystem.read_model_spec(file_name=model_settings.SPEC) + coefficients_df = state.filesystem.read_model_coefficients(model_settings) + model_spec = simulate.eval_coefficients( + state, model_spec, coefficients_df, estimator + ) + nest_spec = config.get_logit_model_settings(model_settings) + + if estimator: + estimator.write_model_settings(model_settings, model_settings_file_name) + estimator.write_spec(model_settings) + estimator.write_coefficients(coefficients_df, model_settings) + estimator.write_choosers(choosers) + + choices = simulate.simple_simulate( + state, + choosers=choosers, + spec=model_spec, + nest_spec=nest_spec, + locals_d=constants, + trace_label=trace_label, + trace_choice_name="telework_arrangement", + estimator=estimator, + compute_settings=model_settings.compute_settings, + ) + + has_in_home_work_activity_alt = model_settings.HAS_IN_HOME_WORK_ACTIVITY_ALT + choices = choices == has_in_home_work_activity_alt + + if estimator: + estimator.write_choices(choices) + choices = estimator.get_survey_values( + choices, + "persons", + "has_in_home_work_activity", + ) + estimator.write_override_choices(choices) + estimator.end_estimation() + + persons["has_in_home_work_activity"] = ( + choices.reindex(persons.index).fillna(0).astype(bool) + ) + + state.add_table("persons", persons) + + tracing.print_summary( + "telework_arrangement.has_in_home_work_activity", + persons.has_in_home_work_activity, + value_counts=True, + ) + + if state.settings.trace_hh_id: + state.tracing.trace_df(persons, label=trace_label, warn_if_empty=True) + + expressions.annotate_tables( + state, + locals_dict=constants, + skims=None, + model_settings=model_settings, + trace_label=trace_label, + ) diff --git a/activitysim/abm/models/telework_duration.py b/activitysim/abm/models/telework_duration.py new file mode 100644 index 0000000000..2707a83c16 --- /dev/null +++ b/activitysim/abm/models/telework_duration.py @@ -0,0 +1,302 @@ +# ActivitySim +# See full license in LICENSE.txt. +from __future__ import annotations + +import logging +from typing import Literal + +import pandas as pd +from pydantic import model_validator + +from activitysim.core import ( + config, + estimation, + expressions, + logit, + simulate, + tracing, + workflow, +) +from activitysim.core.configuration.logit import LogitComponentSettings + +logger = logging.getLogger("activitysim") + + +class TeleworkDurationSettings(LogitComponentSettings, extra="forbid"): + """ + Settings for the `telework_duration` component. + """ + + CHOICE_MODEL: Literal["PROBABILISTIC", "MNL"] = "PROBABILISTIC" + """Choice model type to use for telework duration.""" + + CHOOSER_FILTER_COLUMN_NAME: str = "has_in_home_work_activity" + """Column name in chooser table to represent workers with in-home work activity on the simulation day.""" + + DURATION_CATEGORY_COLUMN_NAME: str = "telework_duration_category" + """Persons column for the chosen telework duration category.""" + + DURATION_HOURS_COLUMN_NAME: str = "telework_duration_hours" + """Persons column for telework duration in hours.""" + + ALTS: str = "telework_duration_alts.csv" + """Alternatives file with duration category to hour mapping.""" + + ALT_NAME_COLUMN: str = "alt" + """Alternatives file column containing category names.""" + + ALT_DURATION_COLUMN: str = "duration_hours" + """Alternatives file column containing duration values in hours.""" + + SPEC: str | None = None + """MNL utility specification file. Required only when CHOICE_MODEL is MNL.""" + + COEFFICIENTS: str | None = None + """MNL coefficients file. Required only when CHOICE_MODEL is MNL.""" + + LOGIT_TYPE: Literal["MNL", "NL"] = "MNL" + """Logit type when running MNL mode.""" + + NESTS: dict | None = None + """Nest settings for NL mode, if ever used.""" + + PROBS_SPEC: str | None = None + """Probabilistic choice lookup table. Required only when CHOICE_MODEL is PROBABILISTIC.""" + + PROBS_JOIN_COLS: list[str] | None = None + """Columns to join choosers to probability table.""" + + CONSTANTS: dict = {} + """Named constants usable in preprocessors and expressions.""" + + preprocessor: dict | list[dict] | None = None + """Chooser preprocessor settings.""" + + @model_validator(mode="after") + def validate_choice_specific_files(self): + if self.CHOICE_MODEL == "MNL": + if not self.SPEC: + raise ValueError("SPEC is required when CHOICE_MODEL is 'MNL'.") + if not self.COEFFICIENTS: + raise ValueError("COEFFICIENTS is required when CHOICE_MODEL is 'MNL'.") + elif self.CHOICE_MODEL == "PROBABILISTIC": + if not self.PROBS_SPEC: + raise ValueError( + "PROBS_SPEC is required when CHOICE_MODEL is 'PROBABILISTIC'." + ) + return self + + +def _load_alternatives(state: workflow.State, model_settings: TeleworkDurationSettings): + alts = simulate.read_model_alts(state, model_settings.ALTS, set_index=None) + + alt_name_col = model_settings.ALT_NAME_COLUMN + alt_duration_col = model_settings.ALT_DURATION_COLUMN + if alt_name_col not in alts.columns or alt_duration_col not in alts.columns: + raise RuntimeError( + "telework_duration alternatives file must include " + f"'{alt_name_col}' and '{alt_duration_col}' columns" + ) + + alts = alts[[alt_name_col, alt_duration_col]].copy() + alts[alt_name_col] = alts[alt_name_col].astype(str) + alts = alts.drop_duplicates(subset=[alt_name_col]) + return alts + + +def _simulate_probabilistic( + state: workflow.State, + choosers: pd.DataFrame, + model_settings: TeleworkDurationSettings, + trace_label: str, +) -> pd.Series: + probs = pd.read_csv( + state.filesystem.get_config_file_path(model_settings.PROBS_SPEC), comment="#" + ) + probs_join_cols = model_settings.PROBS_JOIN_COLS or [] + + if probs_join_cols: + chooser_probs = pd.merge( + choosers.reset_index(), + probs, + on=probs_join_cols, + how="left", + ).set_index(choosers.index.name) + else: + if probs.shape[0] != 1: + raise RuntimeError( + "telework_duration probabilistic mode requires a single-row PROBS_SPEC " + "when PROBS_JOIN_COLS is not provided" + ) + chooser_probs = pd.concat([probs] * len(choosers), ignore_index=True) + chooser_probs.index = choosers.index + + prob_cols = [c for c in probs.columns if c not in probs_join_cols] + if not prob_cols: + raise RuntimeError( + "telework_duration probabilistic mode found no probability columns" + ) + + chooser_probs = chooser_probs[prob_cols].fillna(0) + row_sums = chooser_probs.sum(axis=1) + if (row_sums <= 0).any(): + raise RuntimeError( + "telework_duration probabilistic mode found choosers with no positive " + "probability mass" + ) + + chooser_probs = chooser_probs.div(row_sums, axis=0) + + choices, _ = logit.make_choices( + state, + chooser_probs, + trace_label=trace_label, + trace_choosers=choosers, + ) + + category_choices = pd.Series(prob_cols).loc[choices].astype(str) + category_choices.index = choices.index + + return category_choices + + +@workflow.step +def telework_duration( + state: workflow.State, + persons_merged: pd.DataFrame, + persons: pd.DataFrame, + model_settings: TeleworkDurationSettings | None = None, + model_settings_file_name: str = "telework_duration.yaml", + trace_label: str = "telework_duration", +) -> None: + """ + Simulate daily in-home work duration for workers with in-home work activity. + + This model applies only to workers where `has_in_home_work_activity` is True + from the telework arrangement model. It supports either: + - probabilistic sampling from `PROBS_SPEC`, or + - MNL simulation from `SPEC` and `COEFFICIENTS`. + """ + + if model_settings is None: + model_settings = TeleworkDurationSettings.read_settings_file( + state.filesystem, + model_settings_file_name, + ) + + chooser_filter_col = model_settings.CHOOSER_FILTER_COLUMN_NAME + + choosers = persons_merged[persons_merged[chooser_filter_col]] + + logger.info("Running %s with %d persons", trace_label, len(choosers)) + + category_col = model_settings.DURATION_CATEGORY_COLUMN_NAME + duration_col = model_settings.DURATION_HOURS_COLUMN_NAME + alts = _load_alternatives(state, model_settings) + category_dtype = pd.api.types.CategoricalDtype( + categories=alts[model_settings.ALT_NAME_COLUMN].tolist() + [""], + ordered=False, + ) + + # Default values for non-eligible persons. + persons[category_col] = pd.Series( + pd.Categorical([""] * len(persons), dtype=category_dtype), + index=persons.index, + ) + persons[duration_col] = 0.0 + + if choosers.empty: + state.add_table("persons", persons) + tracing.print_summary(category_col, persons[category_col], value_counts=True) + tracing.print_summary(duration_col, persons[duration_col], value_counts=True) + return + + estimator = estimation.manager.begin_estimation(state, "telework_duration") + constants = config.get_model_constants(model_settings) + + expressions.annotate_preprocessors( + state, + df=choosers, + locals_dict=constants, + skims=None, + model_settings=model_settings, + trace_label=trace_label, + ) + + choice_model = model_settings.CHOICE_MODEL + + if choice_model == "MNL": + model_spec = state.filesystem.read_model_spec(file_name=model_settings.SPEC) + coefficients_df = state.filesystem.read_model_coefficients(model_settings) + model_spec = simulate.eval_coefficients( + state, model_spec, coefficients_df, estimator + ) + nest_spec = config.get_logit_model_settings(model_settings) + + if estimator: + estimator.write_model_settings(model_settings, model_settings_file_name) + estimator.write_spec(model_settings) + estimator.write_coefficients(coefficients_df, model_settings) + estimator.write_choosers(choosers) + + raw_choices = simulate.simple_simulate( + state, + choosers=choosers, + spec=model_spec, + nest_spec=nest_spec, + locals_d=constants, + trace_label=trace_label, + trace_choice_name=category_col, + estimator=estimator, + compute_settings=model_settings.compute_settings, + ) + category_choices = pd.Series( + model_spec.columns[raw_choices.values], index=raw_choices.index + ).astype(category_dtype) + else: + if estimator: + estimator.write_model_settings(model_settings, model_settings_file_name) + estimator.write_spec(model_settings, tag="PROBS_SPEC") + estimator.write_choosers(choosers) + category_choices = _simulate_probabilistic( + state, + choosers, + model_settings, + trace_label, + ).astype(category_dtype) + + alt_to_duration = alts.set_index(model_settings.ALT_NAME_COLUMN)[ + model_settings.ALT_DURATION_COLUMN + ] + + if estimator: + estimator.write_choices(category_choices) + category_choices = estimator.get_survey_values( + category_choices, + "persons", + category_col, + ) + category_choices = category_choices.astype(category_dtype) + estimator.write_override_choices(category_choices) + estimator.end_estimation() + + duration_choices = category_choices.map(alt_to_duration).fillna(0.0).astype(float) + + persons.loc[category_choices.index, category_col] = category_choices + persons.loc[duration_choices.index, duration_col] = duration_choices + + state.add_table("persons", persons) + + tracing.print_summary(category_col, persons[category_col], value_counts=True) + tracing.print_summary(duration_col, persons[duration_col], value_counts=True) + + if state.settings.trace_hh_id: + state.tracing.trace_df(persons, label=trace_label, warn_if_empty=True) + + expressions.annotate_tables( + state, + locals_dict=constants, + skims=None, + model_settings=model_settings, + trace_label=trace_label, + ) diff --git a/activitysim/abm/test/test_misc/test_telework_arrangement.py b/activitysim/abm/test/test_misc/test_telework_arrangement.py new file mode 100644 index 0000000000..c50784986d --- /dev/null +++ b/activitysim/abm/test/test_misc/test_telework_arrangement.py @@ -0,0 +1,393 @@ +import numpy as np +import pandas as pd +import pytest +import openmatrix as omx + +from activitysim.abm.models import telework_arrangement as model +from activitysim.core import workflow, los + + +class DummyFileSystem: + def read_model_spec(self, file_name): + return pd.DataFrame({"alt0": [1.0], "alt1": [0.0]}, index=["1"]) + + def read_model_coefficients(self, model_settings): + return pd.DataFrame({"value": [1.0]}, index=["coef_a"]) + + +class DummyState: + def __init__(self): + self.filesystem = DummyFileSystem() + self.settings = type("Settings", (), {"trace_hh_id": False})() + self.added_tables = {} + + def add_table(self, name, table): + self.added_tables[name] = table.copy() + + +def _settings(filter_col="is_worker", true_alt=0): + return type( + "ModelSettings", + (), + { + "CHOOSER_FILTER_COLUMN_NAME": filter_col, + "HAS_IN_HOME_WORK_ACTIVITY_ALT": true_alt, + "SPEC": "telework_arrangement.csv", + "compute_settings": None, + }, + )() + + +def test_telework_arrangement_monkeypatch(monkeypatch): + state = DummyState() + + persons = pd.DataFrame(index=pd.Index([1, 2, 3], name="person_id")) + persons_merged = pd.DataFrame( + { + "is_worker": [True, False, True], + "some_var": [10, 20, 30], + }, + index=persons.index, + ) + + called = {"annotate": False, "annotate_tables": False, "choosers_index": None} + + monkeypatch.setattr( + model.estimation.manager, "begin_estimation", lambda *a, **k: None + ) + monkeypatch.setattr(model.config, "get_model_constants", lambda *_: {"CONST": 1}) + monkeypatch.setattr(model.config, "get_logit_model_settings", lambda *_: None) + + def fake_annotate_preprocessors(*args, **kwargs): + called["annotate"] = True + assert kwargs["locals_dict"] == {"CONST": 1} + + def fake_annotate_tables(*args, **kwargs): + called["annotate_tables"] = True + assert kwargs["locals_dict"] == {"CONST": 1} + + def fake_eval_coefficients(state, spec, coefficients, estimator): + return spec + + def fake_simple_simulate(*args, **kwargs): + choosers = kwargs["choosers"] + called["choosers_index"] = choosers.index.tolist() + # alt 0 => True, alt 1 => False + return pd.Series([0, 1], index=choosers.index) + + monkeypatch.setattr( + model.expressions, "annotate_preprocessors", fake_annotate_preprocessors + ) + monkeypatch.setattr(model.expressions, "annotate_tables", fake_annotate_tables) + monkeypatch.setattr(model.simulate, "eval_coefficients", fake_eval_coefficients) + monkeypatch.setattr(model.simulate, "simple_simulate", fake_simple_simulate) + monkeypatch.setattr(model.tracing, "print_summary", lambda *a, **k: None) + + model.telework_arrangement( + state=state, + persons_merged=persons_merged, + persons=persons.copy(), + model_settings=_settings(filter_col="is_worker", true_alt=0), + ) + + assert called["annotate"] + assert called["annotate_tables"] + assert called["choosers_index"] == [1, 3] + + out = state.added_tables["persons"] + assert out["has_in_home_work_activity"].dtype == bool + assert out["has_in_home_work_activity"].to_dict() == { + 1: True, + 2: False, + 3: False, + } + + +@pytest.fixture(scope="session") +def example_root(tmp_path_factory): + root = tmp_path_factory.mktemp("example") + config_dir = root / "configs" + config_dir.mkdir() + + data_dir = root / "data" + data_dir.mkdir() + + return root + + +@pytest.fixture(scope="module") +def model_settings(example_root, state): + + model_settings = model.TeleworkArrangementSettings.read_settings_file( + state.filesystem, "telework_arrangement.yaml" + ) + + return model_settings + + +@pytest.fixture(scope="module") +def state(example_root, coeffs_configs_csv, configs_csv) -> workflow.State: + + settings = """ + input_table_list: + - tablename: households + - tablename: persons + - tablename: land_use + """ + + network_los_yaml = """ + zone_system: 2 + taz_skims: skims*.omx + skim_time_periods: + time_window: 1440 + period_minutes: 30 + periods: [12] + labels: &skim_time_period_labels ['AM'] + """ + + skim_matrix = np.array( + [ + [0.42, 0.89, 4.33, 10.31, 9.98], + [0.89, 0.39, 3.76, 10.05, 9.72], + [4.19, 3.61, 0.85, 10.02, 9.69], + [10.57, 9.99, 9.81, 0.16, 0.37], + [10.19, 9.61, 9.43, 0.37, 0.16], + ] + ) + + telework_arrangement_settings = """ + CHOOSER_FILTER_COLUMN_NAME: is_worker + HAS_IN_HOME_WORK_ACTIVITY_ALT: 0 + SPEC: telework_arrangement.csv + COEFFICIENTS: telework_arrangement_coeffs.csv + LOGIT_TYPE: MNL + """ + + telework_arrangement_yaml = example_root / "configs" / "telework_arrangement.yaml" + telework_arrangement_yaml.write_text(telework_arrangement_settings) + + settings_file = example_root / "configs" / "settings.yaml" + settings_file.write_text(settings) + + yaml_file = example_root / "configs" / "network_los.yaml" + yaml_file.write_text(network_los_yaml) + + telework_arrangement_coeffs = ( + example_root / "configs" / "telework_arrangement_coeffs.csv" + ) + telework_arrangement_coeffs.write_text(coeffs_configs_csv) + + telework_arrangement = example_root / "configs" / "telework_arrangement.csv" + telework_arrangement.write_text(configs_csv) + + skims = omx.open_file(example_root / "data" / "skims.omx", "w") + skims["DIST"] = skim_matrix + taz_equivs = [2103, 2104, 2115, 2142, 2144] + skims.create_mapping("zone_number", taz_equivs) + skims.close() + + state = workflow.State.make_default(example_root) + + return state + + +@pytest.fixture(scope="module") +def persons() -> pd.DataFrame: + persons = pd.DataFrame( + { + "person_id": [ + 2664688, + 2664689, + 2668012, + 2668013, + 2701577, + 2701578, + 2860810, + 2860811, + 2865544, + 2865545, + 2865546, + ], + "household_id": [ + 1080351, + 1080351, + 1081684, + 1081684, + 1094369, + 1094369, + 1156249, + 1156249, + 1158612, + 1158612, + 1158612, + ], + "member_id": [1, 2, 1, 2, 1, 2, 1, 2, 1, 2, 3], + "sex": [1, 2, 2, 1, 1, 2, 2, 2, 1, 1, 2], + "maz_seqid": [ + 22660.0, + 22660.0, + 22670.0, + 22670.0, + 22734.0, + 22734.0, + 22803.0, + 22803.0, + 22799.0, + 22799.0, + 22799.0, + ], + "zone_id": [ + 2103.0, + 2103.0, + 2104.0, + 2104.0, + 2115.0, + 2115.0, + 2144.0, + 2144.0, + 2142.0, + 2142.0, + 2142.0, + ], + "is_worker": [ + False, + True, + True, + True, + False, + True, + True, + True, + True, + True, + False, + ], + "home_zone_id": [ + 22660.0, + 22660.0, + 22670.0, + 22670.0, + 22734.0, + 22734.0, + 22803.0, + 22803.0, + 22799.0, + 22799.0, + 22799.0, + ], + "workplace_zone_id": [-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1], + } + ) + persons = persons.set_index("person_id") + + return persons + + +@pytest.fixture(scope="module") +def households() -> pd.DataFrame: + households = pd.DataFrame( + { + "household_id": [1156249, 1080351, 1094369, 1081684, 1158612], + "adjinc": [1010145, 1054606, 1073449, 1031452, 1080470], + "hht": [7.0, 1.0, 1.0, 1.0, 2.0], + "maz": [22803.0, 22660.0, 22734.0, 22670.0, 22799.0], + "taz": [2144.0, 2103.0, 2115.0, 2104.0, 2142.0], + "auto_ownership": [2, 2, 2, 2, 2], + } + ) + households = households.set_index("household_id") + + return households + + +@pytest.fixture(scope="module") +def land_use() -> pd.DataFrame: + land_use = pd.DataFrame( + { + "MAZ": [22660, 22670, 22734, 22799, 22803], + "TAZ": [2103, 2104, 2115, 2142, 2144], + } + ) + + return land_use + + +@pytest.fixture(scope="module") +def configs_csv(): + csv_content = """Label,Description,Expression,has_in_home_work_activity,no_in_home_work_activity +util_acs,alternative specific constant,1,,coef_acs_no_in_home_work +util_female,female,sex==2,coef_female_has_in_home_work, +""" + return csv_content + + +@pytest.fixture(scope="module") +def coeffs_configs_csv(): + csv_content = """coefficient_name,value,constrain +coef_acs_no_in_home_work,0.1,F +coef_female_has_in_home_work,0.5,F +""" + return csv_content + + +@pytest.fixture(scope="module") +def network_los(state, persons, households, land_use) -> los.Network_LOS: + + land_use["zone_id"] = land_use["MAZ"] + land_use.set_index("zone_id", inplace=True) + households["home_zone_id"] = households["maz"] + + state.add_table("persons", persons) + state.add_table("households", households) + state.add_table("land_use", land_use) + + persons_merged = pd.merge( + persons.reset_index(), households, on="household_id", how="left" + ) + persons_merged = pd.merge( + persons_merged, land_use.rename(columns={"TAZ": "taz"}), on="taz", how="left" + ) + + persons_merged["home_zone_id"] = persons_merged["MAZ"] + persons_merged["TAZ"] = persons_merged["taz"] + persons_merged.set_index("person_id", inplace=True) + + state.add_table("persons_merged", persons_merged) + + network_los = los.Network_LOS(state) + + network_los.maz_taz_df = land_use[["MAZ", "TAZ"]] + + network_los.skim_dicts["taz"] = network_los.create_skim_dict("taz") + network_los.skim_dicts["maz"] = network_los.create_skim_dict("maz") + + return network_los + + +def test_telework_arrangement_real(state, model_settings, network_los): + + persons_merged = state.get_dataframe("persons_merged").copy() + + model.telework_arrangement( + state=state, + persons_merged=persons_merged, + persons=state.get_dataframe("persons").copy(), + model_settings=model_settings, + ) + + out = state.get_dataframe("persons")["has_in_home_work_activity"] + + assert out.dtype == bool + assert out.to_dict() == { + 2664688: False, + 2664689: True, + 2668012: False, + 2668013: False, + 2701577: False, + 2701578: True, + 2860810: True, + 2860811: False, + 2865544: True, + 2865545: False, + 2865546: False, + } diff --git a/activitysim/abm/test/test_misc/test_telework_duration.py b/activitysim/abm/test/test_misc/test_telework_duration.py new file mode 100644 index 0000000000..5319df0511 --- /dev/null +++ b/activitysim/abm/test/test_misc/test_telework_duration.py @@ -0,0 +1,279 @@ +from pathlib import Path + +import pandas as pd +import pytest + +from activitysim.abm.models import telework_duration as model +from activitysim.abm.models.settings_checker import try_load_and_check_spec_coefs +from activitysim.core import workflow + + +class DummyFileSystem: + def __init__(self, probs_path: Path): + self.probs_path = probs_path + + def _resolve_path(self, file_name): + candidate = Path(file_name) + if candidate.exists(): + return candidate + tmp_candidate = Path("/tmp") / candidate.name + if tmp_candidate.exists(): + return tmp_candidate + return candidate + + def read_model_alts(self, state, file_name, set_index=None): + return pd.DataFrame( + { + "alt": ["short", "long"], + "duration_hours": [2.0, 4.0], + } + ) + + def read_model_spec(self, file_name, *args, **kwargs): + path = self._resolve_path(file_name) + if not path.exists(): + return pd.DataFrame() + return pd.read_csv(path, comment="#") + + def read_model_coefficients(self, file_name, *args, **kwargs): + path = self._resolve_path(file_name) + if not path.exists(): + return pd.DataFrame(columns=["coefficient_name", "value"]) + df = pd.read_csv(path, comment="#") + if "coefficient_name" in df.columns: + df = df.set_index("coefficient_name") + return df + + def get_config_file_path(self, file_name): + assert file_name == "telework_duration_probs.csv" + return self.probs_path + + +class DummyState: + def __init__(self, probs_path: Path): + self.filesystem = DummyFileSystem(probs_path) + self.settings = type("Settings", (), {"trace_hh_id": False})() + self.added_tables = {} + + def add_table(self, name, table): + self.added_tables[name] = table.copy() + + +def _settings(): + return type( + "ModelSettings", + (), + { + "CHOOSER_FILTER_COLUMN_NAME": "has_in_home_work_activity", + "DURATION_CATEGORY_COLUMN_NAME": "telework_duration_category", + "DURATION_HOURS_COLUMN_NAME": "telework_duration_hours", + "ALTS": "telework_duration_alts.csv", + "ALT_NAME_COLUMN": "alt", + "ALT_DURATION_COLUMN": "duration_hours", + "PROBS_SPEC": "telework_duration_probs.csv", + "PROBS_JOIN_COLS": None, + "CHOICE_MODEL": "PROBABILISTIC", + "compute_settings": None, + }, + )() + + +def test_telework_duration_probabilistic_skips_mnl_spec_validation(monkeypatch): + state = DummyState(Path("/tmp/telework_duration_probs.csv")) + + settings = model.TeleworkDurationSettings( + CHOICE_MODEL="PROBABILISTIC", + PROBS_SPEC="telework_duration_probs.csv", + ) + + errors = try_load_and_check_spec_coefs( + model_name="telework_duration", + model_settings=settings, + state=state, + ) + + assert errors == [] + + +def test_telework_duration_probabilistic_maps_choice_to_duration_monkeypatch( + tmp_path, monkeypatch +): + probs_path = tmp_path / "telework_duration_probs.csv" + probs_path.write_text("short,long\n0.2,0.8\n") + + state = DummyState(probs_path) + persons = pd.DataFrame(index=pd.Index([1, 2], name="person_id")) + persons_merged = pd.DataFrame( + { + "has_in_home_work_activity": [True, False], + }, + index=persons.index, + ) + + called = {"choosers_index": None} + + monkeypatch.setattr( + model.estimation.manager, "begin_estimation", lambda *a, **k: None + ) + monkeypatch.setattr(model.config, "get_model_constants", lambda *_: {}) + monkeypatch.setattr( + model.expressions, "annotate_preprocessors", lambda *a, **k: None + ) + monkeypatch.setattr(model.expressions, "annotate_tables", lambda *a, **k: None) + monkeypatch.setattr(model.tracing, "print_summary", lambda *a, **k: None) + monkeypatch.setattr( + model.simulate, + "read_model_alts", + lambda *a, **k: pd.DataFrame( + {"alt": ["short", "long"], "duration_hours": [2.0, 4.0]} + ), + ) + + def fake_make_choices(state, chooser_probs, trace_label, trace_choosers): + called["choosers_index"] = trace_choosers.index.tolist() + return pd.Series([1], index=trace_choosers.index), None + + monkeypatch.setattr(model.logit, "make_choices", fake_make_choices) + + model.telework_duration( + state=state, + persons_merged=persons_merged, + persons=persons, + model_settings=_settings(), + ) + + assert called["choosers_index"] == [1] + + out = state.added_tables["persons"] + assert out["telework_duration_category"].astype(str).to_dict() == { + 1: "long", + 2: "", + } + assert out["telework_duration_hours"].to_dict() == { + 1: 4.0, + 2: 0.0, + } + + +@pytest.fixture(scope="module") +def real_example_root(tmp_path_factory): + root = tmp_path_factory.mktemp("telework_duration_real") + config_dir = root / "configs" + config_dir.mkdir() + (root / "data").mkdir() + + (config_dir / "settings.yaml").write_text("input_table_list: []\n") + (config_dir / "telework_duration.yaml").write_text( + "CHOOSER_FILTER_COLUMN_NAME: has_in_home_work_activity\n" + "DURATION_CATEGORY_COLUMN_NAME: telework_duration_category\n" + "DURATION_HOURS_COLUMN_NAME: telework_duration_hours\n" + "ALTS: telework_duration_alts.csv\n" + "ALT_NAME_COLUMN: alt\n" + "ALT_DURATION_COLUMN: duration_hours\n" + "PROBS_SPEC: telework_duration_probs.csv\n" + ) + (config_dir / "telework_duration_alts.csv").write_text( + "alt,duration_hours\n" "short,2.0\n" "long,4.0\n" + ) + (config_dir / "telework_duration_probs.csv").write_text("short,long\n0.0,1.0\n") + (config_dir / "telework_duration_mnl.yaml").write_text( + "CHOOSER_FILTER_COLUMN_NAME: has_in_home_work_activity\n" + "DURATION_CATEGORY_COLUMN_NAME: telework_duration_category\n" + "DURATION_HOURS_COLUMN_NAME: telework_duration_hours\n" + "ALTS: telework_duration_alts.csv\n" + "ALT_NAME_COLUMN: alt\n" + "ALT_DURATION_COLUMN: duration_hours\n" + "CHOICE_MODEL: MNL\n" + "SPEC: telework_duration_mnl.csv\n" + "COEFFICIENTS: telework_duration_mnl_coeffs.csv\n" + "LOGIT_TYPE: MNL\n" + ) + (config_dir / "telework_duration_mnl.csv").write_text( + "Label,Description,Expression,short,long\n" + "util_asc,,1,coef_acs_short,\n" + "util_female,,sex==2,coef_female_short,\n" + ) + (config_dir / "telework_duration_mnl_coeffs.csv").write_text( + "coefficient_name,value,constrain\n" + "coef_acs_short,-0.1,F\n" + "coef_female_short,0.5,F\n" + ) + + return root + + +@pytest.fixture(scope="module") +def real_state(real_example_root): + return workflow.State.make_default(real_example_root) + + +def test_telework_duration_probabilistic(real_state): + model_settings = model.TeleworkDurationSettings.read_settings_file( + real_state.filesystem, "telework_duration.yaml" + ) + + persons = pd.DataFrame(index=pd.Index([1, 2, 3], name="person_id")) + persons_merged = pd.DataFrame( + { + "has_in_home_work_activity": [True, False, True], + "sex": [1, 2, 2], + }, + index=persons.index, + ) + + real_state.add_table("persons", persons.copy()) + + model.telework_duration( + state=real_state, + persons_merged=persons_merged, + persons=persons, + model_settings=model_settings, + ) + + out = real_state.get_dataframe("persons") + assert out["telework_duration_category"].astype(str).to_dict() == { + 1: "long", + 2: "", + 3: "long", + } + assert out["telework_duration_hours"].to_dict() == { + 1: 4.0, + 2: 0.0, + 3: 4.0, + } + + +def test_telework_duration_mnl(real_state): + model_settings = model.TeleworkDurationSettings.read_settings_file( + real_state.filesystem, "telework_duration_mnl.yaml" + ) + + persons = pd.DataFrame(index=pd.Index([1, 2, 3], name="person_id")) + persons_merged = pd.DataFrame( + { + "has_in_home_work_activity": [True, False, True], + "sex": [1, 2, 2], + }, + index=persons.index, + ) + + real_state.add_table("persons", persons.copy()) + + model.telework_duration( + state=real_state, + persons_merged=persons_merged, + persons=persons, + model_settings=model_settings, + ) + + out = real_state.get_dataframe("persons") + assert out["telework_duration_category"].astype(str).to_dict() == { + 1: "long", + 2: "", + 3: "long", + } + assert out["telework_duration_hours"].to_dict() == { + 1: 4.0, + 2: 0.0, + 3: 4.0, + } diff --git a/activitysim/core/util.py b/activitysim/core/util.py index 58bc50be46..f383ddd2fb 100644 --- a/activitysim/core/util.py +++ b/activitysim/core/util.py @@ -678,7 +678,9 @@ def drop_unused_columns( pattern = r"[a-zA-Z_][a-zA-Z0-9_]*" unique_variables_in_spec = set( - spec.reset_index()["Expression"].apply(lambda x: re.findall(pattern, x)).sum() + spec.reset_index()["Expression"] + .apply(lambda x: re.findall(pattern, x) if isinstance(x, str) else []) + .sum() ) unique_variables_in_spec |= set(additional_columns or []) diff --git a/activitysim/estimation/larch/simple_simulate.py b/activitysim/estimation/larch/simple_simulate.py index 2ec8a574f3..aef1874e2b 100644 --- a/activitysim/estimation/larch/simple_simulate.py +++ b/activitysim/estimation/larch/simple_simulate.py @@ -311,6 +311,34 @@ def telecommute_status_model( ) +def telework_arrangement_model( + name="telework_arrangement", + edb_directory="output/estimation_data_bundle/{name}/", + return_data=False, +): + return simple_simulate_model( + name=name, + edb_directory=edb_directory, + return_data=return_data, + choices={ + True: 1, + False: 2, + }, # True is has in-home work, false is does not have in-home work, names match spec positions + ) + + +def telework_duration_model( + name="telework_duration", + edb_directory="output/estimation_data_bundle/{name}/", + return_data=False, +): + return simple_simulate_model( + name=name, + edb_directory=edb_directory, + return_data=return_data, + ) + + def mandatory_tour_frequency_model( name="mandatory_tour_frequency", edb_directory="output/estimation_data_bundle/{name}/",