update several models

activitysim/abm/models/atwork_subtour_scheduling.py

@@ -13,6 +13,7 @@
 from activitysim.core import config, estimation, expressions, simulate
 from activitysim.core import timetable as tt
 from activitysim.core import tracing, workflow
+from activitysim.core.configuration.base import PydanticReadable
 from activitysim.core.skim_dataset import SkimDataset
 from activitysim.core.skim_dictionary import SkimDict
 from activitysim.core.util import assign_in_place
@@ -22,20 +23,30 @@
 DUMP = False
 
 
+class AtworkSubtourSchedulingSettings(PydanticReadable):
+    """
+    Settings for the `atwork_subtour_scheduling` component.
+    """
+
+    sharrow_skip: bool = True
+    """Skip Sharow"""  # TODO Check this again
+
+
 @workflow.step
 def atwork_subtour_scheduling(
     state: workflow.State,
     tours: pd.DataFrame,
     persons_merged: pd.DataFrame,
     tdd_alts: pd.DataFrame,
     skim_dict: SkimDict | SkimDataset,
+    model_settings: AtworkSubtourSchedulingSettings | None = None,
+    model_settings_file_name: str = "tour_scheduling_atwork.yaml",
+    trace_label: str = "atwork_subtour_scheduling",
 ) -> None:
     """
     This model predicts the departure time and duration of each activity for at work subtours tours
     """
 
-    trace_label = "atwork_subtour_scheduling"
-    model_settings_file_name = "tour_scheduling_atwork.yaml"
     trace_hh_id = state.settings.trace_hh_id
     subtours = tours[tours.tour_category == "atwork"]
 
@@ -44,11 +55,16 @@ def atwork_subtour_scheduling(
         tracing.no_results(trace_label)
         return
 
-    model_settings = state.filesystem.read_model_settings(model_settings_file_name)
+    if model_settings is None:
+        model_settings = AtworkSubtourSchedulingSettings.read_settings_file(
+            state.filesystem,
+            model_settings_file_name,
+        )
+
     estimator = estimation.manager.begin_estimation(state, "atwork_subtour_scheduling")
 
-    model_spec = state.filesystem.read_model_spec(file_name=model_settings["SPEC"])
-    sharrow_skip = model_settings.get("sharrow_skip")
+    model_spec = state.filesystem.read_model_spec(file_name=model_settings.SPEC)
+    sharrow_skip = model_settings.sharrow_skip
     coefficients_df = state.filesystem.read_model_coefficients(model_settings)
     model_spec = simulate.eval_coefficients(
         state, model_spec, coefficients_df, estimator

activitysim/abm/models/initialize.py

@@ -107,7 +107,7 @@ class InitializeLanduseSettings(PydanticReadable):
 @workflow.step
 def initialize_landuse(
     state: workflow.State,
-    model_settings: InitialiseLanduseSettings | None = None,
+    model_settings: InitializeLanduseSettings | None = None,
     model_settings_file_name: str = "initialize_landuse.yaml",
     trace_label: str = "initialize_landuse",
 ) -> None:

activitysim/abm/models/parking_location_choice.py

@@ -16,6 +16,8 @@
     tracing,
     workflow,
 )
+from activitysim.core.configuration.base import PreprocessorSettings
+from activitysim.core.configuration.logit import LogitComponentSettings
 from activitysim.core.interaction_sample_simulate import interaction_sample_simulate
 from activitysim.core.tracing import print_elapsed_time
 from activitysim.core.util import assign_in_place
@@ -281,33 +283,53 @@ def run_parking_destination(
     return trips[parking_location_column_name], save_sample_df
 
 
+class ParkingLocationSettings(LogitComponentSettings, extra="forbid"):
+    """
+    Settings for the `parking_location` component.
+    """
+
+    preprocessor: PreprocessorSettings | None = None
+    """Setting for the preprocessor."""
+
+    ALT_DEST_COL_NAME: str = "parking_zone"
+    """Parking destination column name."""
+
+    TRIP_DEPARTURE_PERIOD: str = "stop_period"
+    """Trip departure time period."""
+
+
 @workflow.step
 def parking_location(
     state: workflow.State,
     trips: pd.DataFrame,
     trips_merged: pd.DataFrame,
     land_use: pd.DataFrame,
     network_los: los.Network_LOS,
+    model_settings: ParkingLocationSettings | None = None,
+    model_settings_file_name: str = "parking_location_choice.yaml",
+    trace_label: str = "parking_location",
 ) -> None:
     """
     Given a set of trips, each trip needs to have a parking location if
     it is eligible for remote parking.
     """
 
-    trace_label = "parking_location"
-    model_settings = state.filesystem.read_model_settings(
-        "parking_location_choice.yaml"
-    )
+    if model_settings is None:
+        model_settings = ParkingLocationSettings.read_settings_file(
+            state.filesystem,
+            model_settings_file_name,
+        )
+
     trace_hh_id = state.settings.trace_hh_id
-    alt_destination_col_name = model_settings["ALT_DEST_COL_NAME"]
+    alt_destination_col_name = model_settings.ALT_DEST_COL_NAME
 
-    preprocessor_settings = model_settings.get("PREPROCESSOR", None)
+    preprocessor_settings = model_settings.preprocessor
 
     trips_df = trips
     trips_merged_df = trips_merged
     land_use_df = land_use
 
-    proposed_trip_departure_period = model_settings["TRIP_DEPARTURE_PERIOD"]
+    proposed_trip_departure_period = model_settings.TRIP_DEPARTURE_PERIOD
     # TODO: the number of skim time periods should be more readily available than this
     n_skim_time_periods = np.unique(
         network_los.los_settings.skim_time_periods.labels

activitysim/abm/models/stop_frequency.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+from typing import Any
 
 import pandas as pd
 
@@ -16,18 +17,39 @@
     tracing,
     workflow,
 )
+from activitysim.core.configuration.base import PreprocessorSettings, PydanticReadable
+from activitysim.core.configuration.logit import LogitComponentSettings
 from activitysim.core.util import assign_in_place
 
 logger = logging.getLogger(__name__)
 
 
+class StopFrequencySettings(LogitComponentSettings, extra="forbid"):
+    """
+    Settings for the `free_parking` component.
+    """
+
+    preprocessor: PreprocessorSettings | None = None
+    """Setting for the preprocessor."""
+
+    SPEC_SEGMENTS: dict[str, Any] = {}
+    # TODO Check this again
+
+    SEGMENT_COL: str = "primary_purpose"
+
+    # CONSTANTS TODO Check this again
+
+
 @workflow.step
 def stop_frequency(
     state: workflow.State,
     tours: pd.DataFrame,
     tours_merged: pd.DataFrame,
     stop_frequency_alts: pd.DataFrame,
     network_los: los.Network_LOS,
+    model_settings: StopFrequencySettings | None = None,
+    model_settings_file_name: str = "stop_frequency.yaml",
+    trace_label: str = "stop_frequency",
 ) -> None:
     """
     stop frequency model
@@ -55,11 +77,13 @@ def stop_frequency(
 
     """
 
-    trace_label = "stop_frequency"
-    model_settings_file_name = "stop_frequency.yaml"
     trace_hh_id = state.settings.trace_hh_id
 
-    model_settings = state.filesystem.read_model_settings(model_settings_file_name)
+    if model_settings is None:
+        model_settings = StopFrequencySettings.read_settings_file(
+            state.filesystem,
+            model_settings_file_name,
+        )
 
     assert not tours_merged.household_id.isnull().any()
     assert not (tours_merged.origin == -1).any()
@@ -69,7 +93,7 @@ def stop_frequency(
     constants = config.get_model_constants(model_settings)
 
     # - run preprocessor to annotate tours_merged
-    preprocessor_settings = model_settings.get("preprocessor", None)
+    preprocessor_settings = model_settings.preprocessor
     if preprocessor_settings:
         # hack: preprocessor adds origin column in place if it does not exist already
         assert "origin" in tours_merged
@@ -99,11 +123,11 @@ def stop_frequency(
         "stop_frequency segments", tours_merged.primary_purpose, value_counts=True
     )
 
-    spec_segments = model_settings.get("SPEC_SEGMENTS")
+    spec_segments = model_settings.SPEC_SEGMENTS
     assert (
         spec_segments is not None
     ), f"SPEC_SEGMENTS setting not found in model settings: {model_settings_file_name}"
-    segment_col = model_settings.get("SEGMENT_COL")
+    segment_col = model_settings.SEGMENT_COL
     assert (
         segment_col is not None
     ), f"SEGMENT_COL setting not found in model settings: {model_settings_file_name}"

activitysim/abm/models/summarize.py

@@ -9,6 +9,7 @@
 import pandas as pd
 
 from activitysim.core import expressions, workflow
+from activitysim.core.configuration.base import PydanticReadable
 from activitysim.core.los import Network_LOS
 
 logger = logging.getLogger(__name__)
@@ -200,6 +201,21 @@ def manual_breaks(
         return bins
 
 
+class SummarizeSettings(PydanticReadable):
+    """
+    Settings for the `summarize` component.
+    """
+
+    SPECIFICATION: str = "summarize.csv"
+    """Filename for the summarize specification (csv) file."""
+
+    OUTPUT: str = "summarize"
+    """Output folder name."""
+
+    EXPORT_PIPELINE_TABLES: bool = True
+    """To export pipeline tables for expression development."""
+
+
 @workflow.step
 def summarize(
     state: workflow.State,
@@ -211,6 +227,9 @@ def summarize(
     trips: pd.DataFrame,
     tours_merged: pd.DataFrame,
     land_use: pd.DataFrame,
+    model_settings: SummarizeSettings | None = None,
+    model_settings_file_name: str = "summarize.yaml",
+    trace_label: str = "summarize",
 ) -> None:
     """
     A standard model that uses expression files to summarize pipeline tables for vizualization.
@@ -224,17 +243,20 @@ def summarize(
     Outputs a seperate csv summary file for each expression;
     outputs starting with '_' are saved as temporary local variables.
     """
-    trace_label = "summarize"
-    model_settings_file_name = "summarize.yaml"
-    model_settings = state.filesystem.read_model_settings(model_settings_file_name)
+
+    if model_settings is None:
+        model_settings = SummarizeSettings.read_settings_file(
+            state.filesystem,
+            model_settings_file_name,
+        )
 
     output_location = (
-        model_settings["OUTPUT"] if "OUTPUT" in model_settings else "summaries"
+        model_settings.OUTPUT if "OUTPUT" in model_settings else "summaries"
     )
     os.makedirs(state.get_output_file_path(output_location), exist_ok=True)
 
     spec = pd.read_csv(
-        state.filesystem.get_config_file_path(model_settings["SPECIFICATION"]),
+        state.filesystem.get_config_file_path(model_settings.SPECIFICATION),
         comment="#",
     )
 
@@ -311,7 +333,7 @@ def summarize(
                         )
 
     # Output pipeline tables for expression development
-    if model_settings["EXPORT_PIPELINE_TABLES"] is True:
+    if model_settings.EXPORT_PIPELINE_TABLES is True:
         pipeline_table_dir = os.path.join(output_location, "pipeline_tables")
         os.makedirs(state.get_output_file_path(pipeline_table_dir), exist_ok=True)
         for name, df in locals_d.items():

activitysim/abm/models/tour_scheduling_probabilistic.py

@@ -8,6 +8,7 @@
 
 from activitysim.abm.models.util import probabilistic_scheduling as ps
 from activitysim.core import chunk, estimation, workflow
+from activitysim.core.configuration.base import PydanticReadable
 
 logger = logging.getLogger(__name__)
 
@@ -69,10 +70,27 @@ def run_tour_scheduling_probabilistic(
     return choices
 
 
+class TourSchedulingProbabilisticSettings(PydanticReadable):
+    """
+    Settings for the `tour_scheduling_probabilistic` component.
+    """
+
+    depart_alt_base: int = 0
+
+    PROBS_SPEC: str = "tour_scheduling_probs.csv"
+    """Filename for the tour scheduling probabilistic specification (csv) file."""
+
+    PROBS_JOIN_COLS: list[str] | None = None
+    """List of columns"""
+
+
 @workflow.step
 def tour_scheduling_probabilistic(
     state: workflow.State,
     tours: pd.DataFrame,
+    model_settings: TourSchedulingProbabilisticSettings | None = None,
+    model_settings_file_name: str = "tour_scheduling_probabilistic.yaml",
+    trace_label: str = "tour_scheduling_probabilistic",
 ) -> None:
     """Makes tour departure and arrival choices by sampling from a probability lookup table
 
@@ -92,15 +110,18 @@ def tour_scheduling_probabilistic(
 
     """
 
-    trace_label = "tour_scheduling_probabilistic"
-    model_settings_file_name = "tour_scheduling_probabilistic.yaml"
-    model_settings = state.filesystem.read_model_settings(model_settings_file_name)
-    depart_alt_base = model_settings.get("depart_alt_base", 0)
+    if model_settings is None:
+        model_settings = TourSchedulingProbabilisticSettings.read_settings_file(
+            state.filesystem,
+            model_settings_file_name,
+        )
+
+    depart_alt_base = model_settings.depart_alt_base
     scheduling_probs_filepath = state.filesystem.get_config_file_path(
-        model_settings["PROBS_SPEC"]
+        model_settings.PROBS_SPEC
     )
     scheduling_probs = pd.read_csv(scheduling_probs_filepath)
-    probs_join_cols = model_settings["PROBS_JOIN_COLS"]
+    probs_join_cols = model_settings.PROBS_JOIN_COLS
     tours_df = tours
 
     # trip_scheduling is a probabilistic model ane we don't support estimation,