Config Settings and Documentation

.github/workflows/branch-docs.yml

@@ -40,7 +40,7 @@ jobs:
         id: cache
 
       - name: Update environment
-        run: mamba env update -n docbuild -f conda-environments/docbuild.yml
+        run: mamba env update --verbose -n docbuild -f conda-environments/docbuild.yml
         if: steps.cache.outputs.cache-hit != 'true'
 
       - name: Install activitysim

.gitignore

@@ -5,7 +5,6 @@ sandbox/
 .pytest_cache
 .vagrant
 
-
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

activitysim/abm/misc.py

@@ -4,6 +4,7 @@
 
 import logging
 
+import numpy as np
 import pandas as pd
 
 from activitysim.core import workflow
@@ -16,16 +17,16 @@
 
 
 @workflow.cached_object
-def households_sample_size(state: workflow.State, override_hh_ids):
+def households_sample_size(state: workflow.State, override_hh_ids) -> int:
 
     if override_hh_ids is None:
-        return state.settings, households_sample_size
+        return state.settings.households_sample_size
     else:
-        return 0 if override_hh_ids is None else len(override_hh_ids)
+        return len(override_hh_ids)
 
 
 @workflow.cached_object
-def override_hh_ids(state: workflow.State):
+def override_hh_ids(state: workflow.State) -> np.ndarray | None:
 
     hh_ids_filename = state.settings.hh_ids
     if hh_ids_filename is None:
@@ -63,12 +64,12 @@ def override_hh_ids(state: workflow.State):
 
 
 @workflow.cached_object
-def trace_od(state: workflow.State):
+def trace_od(state: workflow.State) -> tuple[int, int] | None:
 
     od = state.settings.trace_od
 
     if od and not (
-        isinstance(od, (list, tuple))
+        isinstance(od, list | tuple)
         and len(od) == 2
         and all(isinstance(x, int) for x in od)
     ):
@@ -81,12 +82,12 @@ def trace_od(state: workflow.State):
 
 
 @workflow.cached_object
-def chunk_size(state: workflow.State):
+def chunk_size(state: workflow.State) -> int:
     _chunk_size = int(state.settings.chunk_size or 0)
 
     return _chunk_size
 
 
 @workflow.cached_object
-def check_for_variability(state: workflow.State):
+def check_for_variability(state: workflow.State) -> bool:
     return bool(state.settings.check_for_variability)

activitysim/abm/models/accessibility.py

@@ -3,16 +3,36 @@
 from __future__ import annotations
 
 import logging
+from typing import Any
 
 import numba as nb
 import numpy as np
 import pandas as pd
 
 from activitysim.core import assign, chunk, los, workflow
+from activitysim.core.configuration.base import PydanticReadable
 
 logger = logging.getLogger(__name__)
 
 
+class AccessibilitySettings(PydanticReadable):
+    """
+    Settings for aggregate accessibility component.
+    """
+
+    CONSTANTS: dict[str, Any] = {}
+
+    land_use_columns: list[str] = []
+    """Only include the these columns in the computational tables
+
+    Memory usage is reduced by only listing the minimum columns needed by
+    the SPEC, and nothing extra.
+    """
+
+    SPEC: str = "accessibility.csv"
+    """Filename for the accessibility specification (csv) file."""
+
+
 @nb.njit
 def _accumulate_accessibility(arr, orig_zone_count, dest_zone_count):
     assert arr.size == orig_zone_count * dest_zone_count
@@ -144,6 +164,10 @@ def compute_accessibility(
     land_use: pd.DataFrame,
     accessibility: pd.DataFrame,
     network_los: los.Network_LOS,
+    model_settings: AccessibilitySettings | None = None,
+    model_settings_file_name: str = "accessibility.yaml",
+    trace_label: str = "compute_accessibility",
+    output_table_name: str = "accessibility",
 ) -> None:
     """
     Compute accessibility for each zone in land use file using expressions from accessibility_spec
@@ -160,40 +184,44 @@ def compute_accessibility(
     product mutes large differences.  The decay function on the walk accessibility measure is
     steeper than automobile or transit.  The minimum accessibility is zero.
     """
+    if model_settings is None:
+        model_settings = AccessibilitySettings.read_settings_file(
+            state.filesystem, model_settings_file_name
+        )
 
-    trace_label = "compute_accessibility"
-    model_settings = state.filesystem.read_model_settings("accessibility.yaml")
     assignment_spec = assign.read_assignment_spec(
-        state.filesystem.get_config_file_path("accessibility.csv")
+        state.filesystem.get_config_file_path(model_settings.SPEC)
     )
 
     accessibility_df = accessibility
     if len(accessibility_df.columns) > 0:
         logger.warning(
-            f"accessibility table is not empty. Columns:{list(accessibility_df.columns)}"
+            f"accessibility table is not empty. "
+            f"Columns:{list(accessibility_df.columns)}"
         )
         raise RuntimeError("accessibility table is not empty.")
 
-    constants = model_settings.get("CONSTANTS", {})
+    constants = model_settings.CONSTANTS
 
-    # only include the land_use columns needed by spec, as specified by land_use_columns model_setting
-    land_use_columns = model_settings.get("land_use_columns", [])
+    # only include the land_use columns needed by spec,
+    # as specified by land_use_columns model_setting
+    land_use_columns = model_settings.land_use_columns
     land_use_df = land_use
     land_use_df = land_use_df[land_use_columns]
 
     logger.info(
-        f"Running {trace_label} with {len(accessibility_df.index)} orig zones {len(land_use_df)} dest zones"
+        f"Running {trace_label} with {len(accessibility_df.index)} orig zones "
+        f"{len(land_use_df)} dest zones"
     )
 
     accessibilities_list = []
 
     for (
-        i,
+        _i,
         chooser_chunk,
-        chunk_trace_label,
+        _chunk_trace_label,
         chunk_sizer,
     ) in chunk.adaptive_chunked_choosers(state, accessibility_df, trace_label):
-
         accessibilities = compute_accessibilities_for_zones(
             state,
             chooser_chunk,
@@ -211,4 +239,4 @@ def compute_accessibility(
     logger.info(f"{trace_label} computed accessibilities {accessibility_df.shape}")
 
     # - write table to pipeline
-    state.add_table("accessibility", accessibility_df)
+    state.add_table(output_table_name, accessibility_df)

activitysim/abm/models/atwork_subtour_destination.py

@@ -8,6 +8,7 @@
 
 from activitysim.abm.models.util import tour_destination
 from activitysim.core import config, estimation, los, tracing, workflow
+from activitysim.core.configuration.logit import TourLocationComponentSettings
 from activitysim.core.util import assign_in_place
 
 logger = logging.getLogger(__name__)
@@ -20,25 +21,30 @@ def atwork_subtour_destination(
     tours: pd.DataFrame,
     persons_merged: pd.DataFrame,
     network_los: los.Network_LOS,
+    model_settings: TourLocationComponentSettings | None = None,
+    model_settings_file_name: str = "atwork_subtour_destination.yaml",
+    trace_label: str = "atwork_subtour_destination",
 ) -> None:
-    trace_label = "atwork_subtour_destination"
-    model_settings_file_name = "atwork_subtour_destination.yaml"
-    model_settings = state.filesystem.read_model_settings(model_settings_file_name)
+    if model_settings is None:
+        model_settings = TourLocationComponentSettings.read_settings_file(
+            state.filesystem,
+            model_settings_file_name,
+        )
 
     future_settings = {
         "SIZE_TERM_SELECTOR": "atwork",
         "SEGMENTS": ["atwork"],
         "ORIG_ZONE_ID": "workplace_zone_id",
     }
-    model_settings = config.future_model_settings(
+    model_settings = config.future_component_settings(
         model_settings_file_name, model_settings, future_settings
     )
 
     destination_column_name = "destination"
-    logsum_column_name = model_settings.get("DEST_CHOICE_LOGSUM_COLUMN_NAME")
+    logsum_column_name = model_settings.DEST_CHOICE_LOGSUM_COLUMN_NAME
     want_logsums = logsum_column_name is not None
 
-    sample_table_name = model_settings.get("DEST_CHOICE_SAMPLE_TABLE_NAME")
+    sample_table_name = model_settings.DEST_CHOICE_SAMPLE_TABLE_NAME
     want_sample_table = (
         state.settings.want_dest_choice_sample_tables and sample_table_name is not None
     )
@@ -54,8 +60,8 @@ def atwork_subtour_destination(
     if estimator:
         estimator.write_coefficients(model_settings=model_settings)
         # estimator.write_spec(model_settings, tag='SAMPLE_SPEC')
-        estimator.write_spec(model_settings, tag="SPEC")
-        estimator.set_alt_id(model_settings["ALT_DEST_COL_NAME"])
+        estimator.write_spec(file_name=model_settings.SPEC, tag="SPEC")
+        estimator.set_alt_id(model_settings.ALT_DEST_COL_NAME)
         estimator.write_table(
             state.get_injectable("size_terms"), "size_terms", append=False
         )

activitysim/abm/models/atwork_subtour_frequency.py

@@ -16,6 +16,8 @@
     tracing,
     workflow,
 )
+from activitysim.core.configuration.base import PreprocessorSettings, PydanticReadable
+from activitysim.core.configuration.logit import LogitComponentSettings
 
 logger = logging.getLogger(__name__)
 
@@ -26,20 +28,30 @@ def add_null_results(state, trace_label, tours):
     state.add_table("tours", tours)
 
 
+class AtworkSubtourFrequencySettings(LogitComponentSettings, extra="forbid"):
+    """
+    Settings for the `atwork_subtour_frequency` component.
+    """
+
+    preprocessor: PreprocessorSettings | None = None
+    """Setting for the preprocessor."""
+
+
 @workflow.step
 def atwork_subtour_frequency(
     state: workflow.State,
     tours: pd.DataFrame,
     persons_merged: pd.DataFrame,
+    model_settings: AtworkSubtourFrequencySettings | None = None,
+    model_settings_file_name: str = "atwork_subtour_frequency.yaml",
+    trace_label: str = "atwork_subtour_frequency",
 ) -> None:
     """
     This model predicts the frequency of making at-work subtour tours
     (alternatives for this model come from a separate csv file which is
     configured by the user).
     """
 
-    trace_label = "atwork_subtour_frequency"
-    model_settings_file_name = "atwork_subtour_frequency.yaml"
     trace_hh_id = state.settings.trace_hh_id
     work_tours = tours[tours.tour_type == "work"]
 
@@ -48,10 +60,15 @@ def atwork_subtour_frequency(
         add_null_results(state, trace_label, tours)
         return
 
-    model_settings = state.filesystem.read_model_settings(model_settings_file_name)
+    if model_settings is None:
+        model_settings = AtworkSubtourFrequencySettings.read_settings_file(
+            state.filesystem,
+            model_settings_file_name,
+        )
+
     estimator = estimation.manager.begin_estimation(state, "atwork_subtour_frequency")
 
-    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
@@ -72,7 +89,7 @@ def atwork_subtour_frequency(
     constants = config.get_model_constants(model_settings)
 
     # - preprocessor
-    preprocessor_settings = model_settings.get("preprocessor", None)
+    preprocessor_settings = model_settings.preprocessor
     if preprocessor_settings:
         expressions.assign_columns(
             state,

activitysim/abm/models/atwork_subtour_mode_choice.py

@@ -9,6 +9,7 @@
 
 from activitysim.abm.models.util.mode import run_tour_mode_choice_simulate
 from activitysim.core import config, estimation, expressions, los, tracing, workflow
+from activitysim.core.configuration.logit import TourModeComponentSettings
 from activitysim.core.util import assign_in_place
 
 logger = logging.getLogger(__name__)
@@ -20,19 +21,23 @@ def atwork_subtour_mode_choice(
     tours: pd.DataFrame,
     persons_merged: pd.DataFrame,
     network_los: los.Network_LOS,
+    model_settings: TourModeComponentSettings | None = None,
+    model_settings_file_name: str = "tour_mode_choice.yaml",
+    trace_label: str = "atwork_subtour_mode_choice",
 ) -> None:
     """
     At-work subtour mode choice simulate
     """
 
-    trace_label = "atwork_subtour_mode_choice"
-
     trace_hh_id = state.settings.trace_hh_id
 
-    model_settings_file_name = "tour_mode_choice.yaml"
-    model_settings = state.filesystem.read_model_settings(model_settings_file_name)
+    if model_settings is None:
+        model_settings = TourModeComponentSettings.read_settings_file(
+            state.filesystem,
+            model_settings_file_name,
+        )
 
-    logsum_column_name = model_settings.get("MODE_CHOICE_LOGSUM_COLUMN_NAME")
+    logsum_column_name = model_settings.MODE_CHOICE_LOGSUM_COLUMN_NAME
     mode_column_name = "tour_mode"
 
     subtours = tours[tours.tour_category == "atwork"]
@@ -57,7 +62,7 @@ def atwork_subtour_mode_choice(
     )
 
     constants = {}
-    constants.update(config.get_model_constants(model_settings))
+    constants.update(model_settings.CONSTANTS)
 
     skim_dict = network_los.get_default_skim_dict()
 
@@ -149,7 +154,7 @@ def atwork_subtour_mode_choice(
 
     # add cached tvpb_logsum tap choices for modes specified in tvpb_mode_path_types
     if network_los.zone_system == los.THREE_ZONE:
-        tvpb_mode_path_types = model_settings.get("tvpb_mode_path_types")
+        tvpb_mode_path_types = model_settings.tvpb_mode_path_types
         for mode, path_types in tvpb_mode_path_types.items():
             for direction, skim in zip(
                 ["od", "do"], [tvpb_logsum_odt, tvpb_logsum_dot]
@@ -189,12 +194,12 @@ def atwork_subtour_mode_choice(
     state.add_table("tours", tours)
 
     # - annotate tours table
-    if model_settings.get("annotate_tours"):
+    if model_settings.annotate_tours:
         tours = state.get_dataframe("tours")
         expressions.assign_columns(
             state,
             df=tours,
-            model_settings=model_settings.get("annotate_tours"),
+            model_settings=model_settings.annotate_tours,
             trace_label=tracing.extend_trace_label(trace_label, "annotate_tours"),
         )
         state.add_table("tours", tours)