JointTourFreqCompSettings

activitysim/abm/models/util/tour_frequency.py

@@ -8,7 +8,9 @@
 import pandas as pd
 
 from activitysim.abm.models.util.canonical_ids import set_tour_index
-from activitysim.core import config, workflow
+from activitysim.core import workflow
+from activitysim.core.configuration.base import PreprocessorSettings, PydanticReadable
+from activitysim.core.configuration.logit import LogitComponentSettings
 from activitysim.core.util import reindex
 
 logger = logging.getLogger(__name__)
@@ -322,7 +324,6 @@ def process_atwork_subtours(
     work_tours: pd.DataFrame,
     atwork_subtour_frequency_alts: pd.DataFrame,
 ):
-
     """
     This method processes the atwork_subtour_frequency column that comes
     out of the model of the same name and turns into a DataFrame that
@@ -615,8 +616,33 @@ def process_tours_frequency_composition(
     return tours
 
 
+class JointTourFreqCompContent(PydanticReadable):
+    VALUE_MAP: dict[int, str]
+    COLUMNS: list[str]
+
+
+class JointTourFreqCompAlts(PydanticReadable):
+    PURPOSE: JointTourFreqCompContent
+    COMPOSITION: JointTourFreqCompContent
+
+
+class JointTourFreqCompSettings(LogitComponentSettings):
+    """
+    Settings for joint tour frequency and composition.
+    """
+
+    ALTS_TABLE_STRUCTURE: JointTourFreqCompAlts
+    preprocessor: PreprocessorSettings | None = None
+    ALTS_PREPROCESSOR: PreprocessorSettings | None = None
+
+
 def create_joint_tours(
-    state: workflow.State, tour_counts, tour_category, parent_col="person_id"
+    state: workflow.State,
+    tour_counts,
+    tour_category,
+    parent_col="person_id",
+    model_settings: JointTourFreqCompSettings | None = None,
+    model_settings_file_name: str = "joint_tour_frequency_composition.yaml",
 ):
     """
     This method processes the tour_frequency column that comes
@@ -657,36 +683,17 @@ def create_joint_tours(
     2588676       2         0         0
     2588677       1         1         0
     """
-    model_settings_file_name = "joint_tour_frequency_composition.yaml"
 
-    model_settings = state.filesystem.read_model_settings(model_settings_file_name)
+    if model_settings is None:
+        model_settings = JointTourFreqCompSettings.read_settings_file(
+            state.filesystem, model_settings_file_name
+        )
 
-    alts_table_structure = model_settings.get("ALTS_TABLE_STRUCTURE", None)
-    assert (
-        alts_table_structure is not None
-    ), f"Expected to find ALTS_TABLE_STRUCTURE setting in joint_tour_frequency_composition.yaml"
-
-    tour_type_dict = alts_table_structure.get("PURPOSE", None).get("VALUE_MAP", None)
-    assert (
-        tour_type_dict is not None
-    ), f"Expected to find PURPOSE.VALUE_MAP setting in ALTS_TABLE_STRUCTURE"
-
-    tour_type_cols = alts_table_structure.get("PURPOSE", None).get("COLUMNS", None)
-    assert (
-        tour_type_cols is not None
-    ), f"Expected to find PURPOSE.COLUMNS setting in ALTS_TABLE_STRUCTURE"
-
-    tour_comp_dict = alts_table_structure.get("COMPOSITION", None).get(
-        "VALUE_MAP", None
-    )
-    assert (
-        tour_comp_dict is not None
-    ), f"Expected to find COMPOSITION.VALUE_MAP setting in ALTS_TABLE_STRUCTURE"
-
-    tour_comp_cols = alts_table_structure.get("COMPOSITION", None).get("COLUMNS", None)
-    assert (
-        tour_comp_cols is not None
-    ), f"Expected to find COMPOSITION.COLUMNS setting in ALTS_TABLE_STRUCTURE"
+    alts_table_structure = model_settings.ALTS_TABLE_STRUCTURE
+    tour_type_dict = alts_table_structure.PURPOSE.VALUE_MAP
+    tour_type_cols = alts_table_structure.PURPOSE.COLUMNS
+    tour_comp_dict = alts_table_structure.COMPOSITION.VALUE_MAP
+    tour_comp_cols = alts_table_structure.COMPOSITION.COLUMNS
 
     # reformat with the columns given below
     tours_purp = tour_counts[tour_type_cols].stack().reset_index()