Updating model components

activitysim/abm/models/auto_ownership.py

@@ -19,9 +19,6 @@ class AutoOwnershipSettings(LogitComponentSettings):
     Settings for the `auto_ownership` component.
     """
 
-    SPEC: str = "auto_ownership.csv"
-    """Filename for the auto ownership specification (csv) file."""
-
 
 @workflow.step
 def auto_ownership_simulate(

activitysim/abm/models/initialize.py

@@ -8,6 +8,7 @@
 
 from activitysim.abm.tables import disaggregate_accessibility, shadow_pricing
 from activitysim.core import chunk, expressions, tracing, workflow
+from activitysim.core.configuration.base import PydanticReadable
 
 # We are using the naming conventions in the mtc_asim.h5 example
 # file for our default list. This provides backwards compatibility
@@ -97,8 +98,19 @@ def annotate_tables(state: workflow.State, model_settings, trace_label, chunk_si
         chunk_sizer.log_df(trace_label, tablename, None)
 
 
+class InitializeLanduseSettings(PydanticReadable):
+    """
+    Settings for the `initialize_landuse` component.
+    """
+
+
 @workflow.step
-def initialize_landuse(state: workflow.State) -> None:
+def initialize_landuse(
+    state: workflow.State,
+    model_settings: InitialiseLanduseSettings | None = None,
+    model_settings_file_name: str = "initialize_landuse.yaml",
+    trace_label: str = "initialize_landuse",
+) -> None:
     """
     Initialize the land use table.
 
@@ -110,19 +122,22 @@ def initialize_landuse(state: workflow.State) -> None:
     -------
     ?
     """
-    trace_label = "initialize_landuse"
-    settings_filename = "initialize_landuse.yaml"
+    # trace_label = "initialize_landuse"
+    # settings_filename = "initialize_landuse.yaml"
 
     with chunk.chunk_log(state, trace_label, base=True) as chunk_sizer:
-        model_settings = state.filesystem.read_settings_file(
-            settings_filename, mandatory=True
-        )
+        if model_settings is None:
+            model_settings = InitializeLanduseSettings.read_settings_file(
+                state.filesystem,
+                model_settings_file_name,
+                mandatory=True,
+            )
 
-        annotate_tables(state, model_settings, trace_label, chunk_sizer)
+            annotate_tables(state, model_settings, trace_label, chunk_sizer)
 
-        # instantiate accessibility (must be checkpointed to be be used to slice accessibility)
-        accessibility = state.get_dataframe("accessibility")
-        chunk_sizer.log_df(trace_label, "accessibility", accessibility)
+            # instantiate accessibility (must be checkpointed to be be used to slice accessibility)
+            accessibility = state.get_dataframe("accessibility")
+            chunk_sizer.log_df(trace_label, "accessibility", accessibility)
 
 
 @workflow.step

activitysim/abm/models/transit_pass_ownership.py

@@ -14,33 +14,49 @@
     tracing,
     workflow,
 )
+from activitysim.core.configuration.base import PreprocessorSettings
+from activitysim.core.configuration.logit import LogitComponentSettings
 
 logger = logging.getLogger("activitysim")
 
 
+class TransitPassOwnershipSettings(LogitComponentSettings):
+    """
+    Settings for the `transit_pass_ownership` component.
+    """
+
+    preprocessor: PreprocessorSettings | None = None
+    """Setting for the preprocessor."""
+
+
 @workflow.step
 def transit_pass_ownership(
     state: workflow.State,
     persons_merged: pd.DataFrame,
     persons: pd.DataFrame,
+    model_settings: TransitPassOwnershipSettings | None = None,
+    model_settings_file_name: str = "transit_pass_ownership.yaml",
+    trace_label: str = "transit_pass_ownership",
 ) -> None:
     """
     Transit pass ownership model.
     """
 
-    trace_label = "transit_pass_ownership"
-    model_settings_file_name = "transit_pass_ownership.yaml"
+    if model_settings is None:
+        model_settings = TransitPassOwnershipSettings.read_settings_file(
+            state.filesystem,
+            model_settings_file_name,
+        )
 
     choosers = persons_merged
     logger.info("Running %s with %d persons", trace_label, len(choosers))
 
-    model_settings = state.filesystem.read_model_settings(model_settings_file_name)
     estimator = estimation.manager.begin_estimation(state, "transit_pass_ownership")
 
     constants = config.get_model_constants(model_settings)
 
     # - preprocessor
-    preprocessor_settings = model_settings.get("preprocessor", None)
+    preprocessor_settings = model_settings.preprocessor
     if preprocessor_settings:
         locals_d = {}
         if constants is not None:
@@ -54,7 +70,7 @@ def transit_pass_ownership(
             trace_label=trace_label,
         )
 
-    model_spec = state.filesystem.read_model_spec(file_name=model_settings["SPEC"])
+    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

activitysim/abm/models/transit_pass_subsidy.py

@@ -50,7 +50,6 @@ def transit_pass_subsidy(
     choosers = persons_merged
     logger.info("Running %s with %d persons", trace_label, len(choosers))
 
-    model_settings = state.filesystem.read_model_settings(model_settings_file_name)
     estimator = estimation.manager.begin_estimation(state, "transit_pass_subsidy")
 
     constants = config.get_model_constants(model_settings)

activitysim/abm/models/vehicle_type_choice.py

@@ -18,6 +18,7 @@
     tracing,
     workflow,
 )
+from activitysim.core.configuration.logit import LogitComponentSettings
 from activitysim.core.interaction_simulate import interaction_simulate
 
 logger = logging.getLogger(__name__)
@@ -473,13 +474,22 @@ def iterate_vehicle_type_choice(
     return all_choices, all_choosers
 
 
+class VehicleTypeChoiceSettings(LogitComponentSettings):
+    """
+    Settings for the `vehicle_type_choice` component.
+    """
+
+
 @workflow.step
 def vehicle_type_choice(
     state: workflow.State,
     persons: pd.DataFrame,
     households: pd.DataFrame,
     vehicles: pd.DataFrame,
     vehicles_merged: pd.DataFrame,
+    model_settings: VehicleTypeChoiceSettings | None = None,
+    model_settings_file_name: str = "vehicle_type_choice.yaml",
+    trace_label: str = "vehicle_type_choice",
 ) -> None:
     """Assign a vehicle type to each vehicle in the `vehicles` table.
 
@@ -519,15 +529,20 @@ def vehicle_type_choice(
     persons : pd.DataFrame
     households : pd.DataFrame
     vehicles : pd.DataFrame
-    vehicles_merged : DataFrame
+    vehicles_merged :pd. DataFrame
+    model_settings : class specifying the model settings
+    model_settings_file_name: filename of the model settings file
+    trace_label: trace label of the vehicle type choice model
     """
-    trace_label = "vehicle_type_choice"
-    model_settings_file_name = "vehicle_type_choice.yaml"
-    model_settings = state.filesystem.read_model_settings(model_settings_file_name)
+    if model_settings is None:
+        model_settings = VehicleTypeChoiceSettings.read_settings_file(
+            state.filesystem,
+            model_settings_file_name,
+        )
 
     estimator = estimation.manager.begin_estimation(state, "vehicle_type")
 
-    model_spec_raw = state.filesystem.read_model_spec(file_name=model_settings["SPEC"])
+    model_spec_raw = 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_raw, coefficients_df, estimator

activitysim/abm/models/work_from_home.py

@@ -27,15 +27,37 @@ class WorkFromHomeSettings(LogitComponentSettings, extra="forbid"):
     """
 
     preprocessor: PreprocessorSettings | None = None
+    """Setting for the preprocessor."""
+
     WORK_FROM_HOME_ALT: int
+    """ """  # TODO
+
     WORK_FROM_HOME_ITERATIONS: int | None = None
-    CHOOSER_FILTER_COLUMN_NAME: str = "is_worker"
+    """Setting to specify the number of iterations."""
+
+    CHOOSER_FILTER_COLUMN_NAME: str = "is_worker" | None
+    """Column name in the dataframe to represent worker."""
+
     WORK_FROM_HOME_CHOOSER_FILTER: str = None
+    """Setting to filter work from home chooser."""
+
     WORK_FROM_HOME_COEFFICIENT_CONSTANT: float = None
+    """Setting to set the work from home coefficient."""
+
     WORK_FROM_HOME_TARGET_PERCENT: float = None
+    """Setting to set work from target percent."""
+
     WORK_FROM_HOME_TARGET_PERCENT_TOLERANCE: float = None
+    """Setting to set work from home target percent tolerance."""
+
     sharrow_skip: bool = False
-    DEST_CHOICE_COLUMN_NAME: str = "workplace_zone_id"
+    """Setting to skip sharrow."""
+
+    DEST_CHOICE_COLUMN_NAME: str = "workplace_zone_id" | None
+    """Column name in persons dataframe to specify the workplace zone id. """
+
+    SPEC: str = "work_from_home.csv"
+    """Filename for the accessibility specification (csv) file."""
 
 
 @workflow.step
@@ -60,7 +82,6 @@ def work_from_home(
         )
 
     choosers = persons_merged
-    model_settings = state.filesystem.read_model_settings(model_settings_file_name)
     chooser_filter_column_name = model_settings.CHOOSER_FILTER_COLUMN_NAME
     choosers = choosers[choosers[chooser_filter_column_name]]
     logger.info("Running %s with %d persons", trace_label, len(choosers))