Make global constants available to interaction-based model expressions

activitysim/abm/models/trip_destination.py

@@ -189,7 +189,7 @@ def _destination_sample(
             f"SAMPLE_SIZE set to 0 for {trace_label} because disable_destination_sampling is set"
         )
 
-    locals_dict = state.get_global_constants().copy()
+    locals_dict = {}
     locals_dict.update(model_settings.CONSTANTS)
 
     # size_terms of destination zones are purpose-specific, and trips have various purposes

activitysim/core/expressions.py

@@ -321,8 +321,8 @@ def annotate_tables(
                     "Failed to set skim wrapper targets: %s. Skims wrappers may not be used in expressions.",
                     e,
                 )
-        if locals_dict:
-            locals_d.update(locals_dict)
+
+        locals_d.update(locals_dict or {})
 
         results = compute_columns(
             state,

activitysim/core/flow.py

@@ -808,8 +808,10 @@ def apply_flow(
     """
     if sh is None:
         return None, None
-    if locals_d is None:
-        locals_d = {}
+
+    # Global constants are always available, but can be overridden by locals_d.
+    locals_d = {**state.get_global_constants(), **(locals_d or {})}
+
     with logtime("apply_flow"):
         try:
             flow = get_flow(

activitysim/core/interaction_simulate.py

@@ -100,7 +100,7 @@ def eval_interaction_utilities(
         assert len(spec.columns) == 1
 
         # avoid altering caller's passed-in locals_d parameter (they may be looping)
-        locals_d = locals_d.copy() if locals_d is not None else {}
+        locals_d = dict(locals_d or {})
 
         utilities = None
 
@@ -210,6 +210,9 @@ def replace_in_index_level(mi, level, *repls):
             or estimator
             or (sharrow_enabled == "test" and extra_data is None)
         ):
+            # Global constants are always available, but can be overridden by locals_d.
+            # Sharrow calculations receive them in flow.apply_flow instead.
+            locals_d = {**state.get_global_constants(), **locals_d}
 
             def to_series(x):
                 if np.isscalar(x):

activitysim/core/simulate.py

@@ -655,7 +655,6 @@ def eval_utilities(
         from .flow import apply_flow  # import inside func to prevent circular imports
 
         locals_dict = {}
-        locals_dict.update(state.get_global_constants())
         if locals_d is not None:
             locals_dict.update(locals_d)
         sh_util, sh_flow, sh_tree = apply_flow(

activitysim/core/test/configs/preprocessor.csv

@@ -5,4 +5,5 @@ count persons test,num_persons,persons.groupby('household_id').size().reindex(df
 skim dict test,od_distance,"skim_dict.lookup(df.origin, df.destination, 'DIST')"
 skim wrapper test,od_distance_wrapper,skims2d['DIST']
 sov time,od_sov_time,skims3d['SOV_TIME']
-testing constant from locals_dict,constant_test,test_constant / 2
+testing constant from locals_dict,constant_test,test_constant / 2
+testing global constant,global_constant_test,global_test_constant / 2

activitysim/core/test/test_interaction_sample_simulate.py

@@ -7,7 +7,7 @@
 import pandas as pd
 import pytest
 
-from activitysim.core import interaction_sample_simulate, workflow
+from activitysim.core import interaction_sample, interaction_sample_simulate, workflow
 from activitysim.core.logit import AltsContext
 
 
@@ -18,6 +18,64 @@ def state() -> workflow.State:
     return state
 
 
+def test_global_constants_available_in_sampling_and_simulation(tmp_path):
+    """Global constants are available in both destination-choice substeps."""
+    configs_dir = tmp_path.joinpath("configs")
+    configs_dir.mkdir()
+    configs_dir.joinpath("constants.yaml").write_text(
+        "SAMPLE_SCALE: 3.0\nSIMULATE_SCALE: 2.0\n"
+    )
+    tmp_path.joinpath("data").mkdir()
+
+    state = workflow.State()
+    state.initialize_filesystem(
+        working_dir=tmp_path, configs_dir=("configs",)
+    ).default_settings()
+    state.settings.check_for_variability = False
+
+    choosers = pd.DataFrame(
+        {"chooser_attr": [1.0, 2.0]},
+        index=pd.Index([0, 1], name="person_id"),
+    )
+    alternatives = pd.DataFrame(
+        {"alt_attr": [1.0, 2.0]},
+        index=pd.Index([10, 20], name="alt_id"),
+    )
+
+    # Sampling and simulation use separate specifications in location and
+    # destination choice, so exercise each expression-evaluation path.
+    sample_spec = pd.DataFrame(
+        {"coefficient": [1.0]},
+        index=pd.Index(["alt_attr * SAMPLE_SCALE"], name="Expression"),
+    )
+    sample = interaction_sample.interaction_sample(
+        state,
+        choosers,
+        alternatives,
+        sample_spec,
+        sample_size=0,
+        alt_col_name="alt_id",
+    )
+    sampled_alternatives = sample.join(alternatives, on="alt_id")
+
+    simulate_spec = pd.DataFrame(
+        {"coefficient": [1.0]},
+        index=pd.Index(["alt_attr * SIMULATE_SCALE"], name="Expression"),
+    )
+    results = interaction_sample_simulate.interaction_sample_simulate(
+        state,
+        choosers,
+        sampled_alternatives,
+        simulate_spec,
+        choice_column="alt_id",
+        want_logsums=True,
+        skip_choice=True,
+    )
+
+    expected_logsum = np.logaddexp(2.0, 4.0)
+    np.testing.assert_allclose(results["logsums"], expected_logsum)
+
+
 def test_interaction_sample_simulate_parity(state):
     # Run interaction_sample_simulate with and without explicit error terms and check that results are similar.
 

activitysim/core/test/test_interaction_simulate.py

@@ -1,11 +1,13 @@
 # ActivitySim
 # See full license in LICENSE.txt.
 
+from __future__ import annotations
+
 import numpy as np
 import pandas as pd
 import pytest
 
-from activitysim.core import interaction_simulate, workflow
+from activitysim.core import flow, interaction_simulate, workflow
 
 
 @pytest.fixture
@@ -15,6 +17,39 @@ def state() -> workflow.State:
     return state
 
 
+def test_apply_flow_global_constants_and_local_override(state, monkeypatch):
+    class FakeFlow:
+        name = "test_flow"
+        compiled_recently = False
+        tree = object()
+
+        def dot(self, coefficients, dtype, compile_watch):
+            return np.array([[1.0]])
+
+    captured_locals = {}
+
+    def fake_get_flow(_state, _spec, locals_d, *_args, **_kwargs):
+        captured_locals.update(locals_d)
+        return FakeFlow()
+
+    state.get_global_constants = lambda: {"GLOBAL_SCALE": 2, "GLOBAL_ONLY": 4}
+    monkeypatch.setattr(flow, "sh", object())
+    monkeypatch.setattr(flow, "get_flow", fake_get_flow)
+
+    spec = pd.DataFrame(
+        {"alt": [1.0]}, index=pd.Index(["GLOBAL_SCALE"], name="Expression")
+    )
+    result, _, _ = flow.apply_flow(
+        state,
+        spec,
+        pd.DataFrame({"value": [1.0]}),
+        locals_d={"GLOBAL_SCALE": 3},
+    )
+
+    np.testing.assert_allclose(result, [[1.0]])
+    assert captured_locals == {"GLOBAL_SCALE": 3, "GLOBAL_ONLY": 4}
+
+
 def test_interaction_simulate_explicit_error_terms_parity(state):
     # Run interaction_simulate with and without explicit error terms and check that results are similar.
 
@@ -172,3 +207,73 @@ def test_interaction_simulate_eet_large_utilities(state):
     assert not choices_eet.isna().any()
     # With such a large difference, Alt 1 should be the dominant choice
     assert (choices_eet == 1).all()
+
+
+def test_eval_interaction_utilities_global_constants(tmp_path):
+    # global constants (from constants.yaml) should be available to expressions
+    # evaluated for interaction models (e.g. location choice, destination choice,
+    # tour scheduling), see issue #1015
+
+    configs_dir = tmp_path.joinpath("configs")
+    configs_dir.mkdir()
+    configs_dir.joinpath("constants.yaml").write_text("KM_TO_MILE: 0.621371\n")
+    tmp_path.joinpath("data").mkdir()
+
+    state = workflow.State()
+    state.initialize_filesystem(
+        working_dir=tmp_path, configs_dir=("configs",)
+    ).default_settings()
+    state.settings.check_for_variability = False
+
+    df = pd.DataFrame({"distance_km": [1.0, 10.0]}, index=[0, 1])
+
+    spec = pd.DataFrame(
+        {"coefficient": [1.0]},
+        index=pd.Index(["distance_km * KM_TO_MILE"], name="Expression"),
+    )
+
+    utilities, _ = interaction_simulate.eval_interaction_utilities(
+        state,
+        spec,
+        df,
+        locals_d=None,
+        trace_label="test_global_constants",
+        trace_rows=None,
+    )
+
+    np.testing.assert_allclose(
+        utilities.utility.to_numpy(), df.distance_km.to_numpy() * 0.621371
+    )
+
+
+def test_eval_interaction_utilities_locals_override_global_constants(tmp_path):
+    # values passed in locals_d take precedence over global constants
+
+    configs_dir = tmp_path.joinpath("configs")
+    configs_dir.mkdir()
+    configs_dir.joinpath("constants.yaml").write_text("KM_TO_MILE: 0.621371\n")
+    tmp_path.joinpath("data").mkdir()
+
+    state = workflow.State()
+    state.initialize_filesystem(
+        working_dir=tmp_path, configs_dir=("configs",)
+    ).default_settings()
+    state.settings.check_for_variability = False
+
+    df = pd.DataFrame({"distance_km": [1.0, 10.0]}, index=[0, 1])
+
+    spec = pd.DataFrame(
+        {"coefficient": [1.0]},
+        index=pd.Index(["distance_km * KM_TO_MILE"], name="Expression"),
+    )
+
+    utilities, _ = interaction_simulate.eval_interaction_utilities(
+        state,
+        spec,
+        df,
+        locals_d={"KM_TO_MILE": 1.0},
+        trace_label="test_global_constants_override",
+        trace_rows=None,
+    )
+
+    np.testing.assert_allclose(utilities.utility.to_numpy(), df.distance_km.to_numpy())

activitysim/core/test/test_preprocessing.py

@@ -87,6 +87,7 @@ def check_outputs(tours):
         "od_distance_wrapper",
         "od_sov_time",
         "constant_test",
+        "global_constant_test",
     ]
 
     # check all new columns are added
@@ -109,6 +110,7 @@ def check_outputs(tours):
             "od_distance_wrapper": [0.24, 0.28, 0.57],
             "od_sov_time": [0.78, 0.89, 1.76],
             "constant_test": [21, 21, 21],
+            "global_constant_test": [21, 21, 21],
         }
     ).set_index("tour_id")
     pd.testing.assert_frame_equal(tours[new_cols], exppected_output, check_dtype=False)
@@ -124,7 +126,11 @@ def setup_skims(state: workflow.State):
     return {"skims3d": skims3d, "skims2d": skims2d}
 
 
-def test_preprocessor(state: workflow.State, households, persons, tours):
+def test_preprocessor(state: workflow.State, households, persons, tours, monkeypatch):
+    monkeypatch.setattr(
+        state, "get_global_constants", lambda: {"global_test_constant": 42}
+    )
+
     # adding dataframes to state so they can be accessed in preprocessor
     state.add_table("households", households)
     state.add_table("persons", persons)
@@ -156,7 +162,11 @@ def test_preprocessor(state: workflow.State, households, persons, tours):
     pd.testing.assert_frame_equal(state_tours, original_tours)
 
 
-def test_annotator(state, households, persons, tours):
+def test_annotator(state, households, persons, tours, monkeypatch):
+    monkeypatch.setattr(
+        state, "get_global_constants", lambda: {"global_test_constant": 42}
+    )
+
     # adding dataframes to state so they can be accessed in annotator
     state.add_table("households", households)
     state.add_table("persons", persons)

activitysim/core/test/test_simulate.py

@@ -82,6 +82,33 @@ def test_eval_variables(state, spec, data):
     pdt.assert_frame_equal(result, expected, check_names=False)
 
 
+def test_standard_utilities_global_constants_and_local_override(state):
+    state.get_global_constants = lambda: {"GLOBAL_SCALE": 2}
+    choosers = pd.DataFrame({"value": [1.0, 2.0]})
+    spec = pd.DataFrame(
+        {"alt": [1.0]},
+        index=pd.Index(["@df.value * GLOBAL_SCALE"], name="Expression"),
+    )
+    chunk_sizer = chunk.ChunkSizer(state, "", "", len(choosers))
+
+    utilities = simulate.eval_utilities(
+        state,
+        spec,
+        choosers,
+        chunk_sizer=chunk_sizer,
+    )
+    overridden_utilities = simulate.eval_utilities(
+        state,
+        spec,
+        choosers,
+        locals_d={"GLOBAL_SCALE": 3},
+        chunk_sizer=chunk_sizer,
+    )
+
+    npt.assert_allclose(utilities["alt"], [2.0, 4.0])
+    npt.assert_allclose(overridden_utilities["alt"], [3.0, 6.0])
+
+
 def test_simple_simulate(state, data, spec):
     state.settings.check_for_variability = False