From 1982b84c5d9878fce91b19f52aa47dd902e7bfc1 Mon Sep 17 00:00:00 2001 From: Yue Shuai <48269801+yueshuaing@users.noreply.github.com> Date: Tue, 4 Aug 2026 16:38:31 -0700 Subject: [PATCH 1/6] implement multiprocessing in the current cvm --- src/asim-cvm/configs/settings.yaml | 19 ++++++++ src/asim-cvm/configs/shadow_pricing.yaml | 2 + src/asim-cvm/extensions/cvm_establishments.py | 4 ++ .../extensions/cvm_route_generation.py | 4 ++ src/asim-cvm/run_cvm.py | 45 +++++++++++++------ 5 files changed, 60 insertions(+), 14 deletions(-) create mode 100644 src/asim-cvm/configs/shadow_pricing.yaml diff --git a/src/asim-cvm/configs/settings.yaml b/src/asim-cvm/configs/settings.yaml index 4c135410b..e9581dff4 100644 --- a/src/asim-cvm/configs/settings.yaml +++ b/src/asim-cvm/configs/settings.yaml @@ -102,3 +102,22 @@ models: - write_cvm_trip_matrices - write_tables +multiprocess: True +fail_fast: True +num_processes: 12 + +chunk_size: 0 +chunk_method: hybrid_uss +chunk_training_mode: disabled + +multiprocess_steps: + - name: mp_initialize + begin: household_attractor + - name: mp_route + begin: route_purpose_and_vehicle + slice: + tables: + - establishments + - routes + - name: mp_finalize + begin: write_cvm_trip_matrices diff --git a/src/asim-cvm/configs/shadow_pricing.yaml b/src/asim-cvm/configs/shadow_pricing.yaml new file mode 100644 index 000000000..9ab438378 --- /dev/null +++ b/src/asim-cvm/configs/shadow_pricing.yaml @@ -0,0 +1,2 @@ +# placeholder required by multiprocessing +shadow_pricing_models: \ No newline at end of file diff --git a/src/asim-cvm/extensions/cvm_establishments.py b/src/asim-cvm/extensions/cvm_establishments.py index f60a572a5..1f7c1389b 100644 --- a/src/asim-cvm/extensions/cvm_establishments.py +++ b/src/asim-cvm/extensions/cvm_establishments.py @@ -9,11 +9,15 @@ from activitysim.core import tracing, workflow from activitysim.core.input import read_input_table +from activitysim.abm.models.util import canonical_ids as cid from .cvm_state import State logger = logging.getLogger(__name__) +cid.RANDOM_CHANNELS.append("establishments") +cid.TRACEABLE_TABLES.append("establishments") +cid.CANONICAL_TABLE_INDEX_NAMES.setdefault("establishments", "establishment_id") @workflow.table def establishments(state: State) -> pd.DataFrame: diff --git a/src/asim-cvm/extensions/cvm_route_generation.py b/src/asim-cvm/extensions/cvm_route_generation.py index 7a335d5d8..6832196d7 100644 --- a/src/asim-cvm/extensions/cvm_route_generation.py +++ b/src/asim-cvm/extensions/cvm_route_generation.py @@ -23,6 +23,7 @@ from activitysim.abm.tables.util import simple_table_join from activitysim.core import workflow, tracing +from activitysim.abm.models.util import canonical_ids as cid from .cvm_enum import BusinessTypes from .cvm_enum_tools import as_int_enum @@ -32,6 +33,9 @@ _business_type_offset = int(10 ** np.ceil(np.log10(max(BusinessTypes)))) +cid.RANDOM_CHANNELS.append("routes") +cid.TRACEABLE_TABLES.append("routes") +cid.CANONICAL_TABLE_INDEX_NAMES.setdefault("routes", "route_id") @workflow.step def route_generation( diff --git a/src/asim-cvm/run_cvm.py b/src/asim-cvm/run_cvm.py index 759584b56..3afdcedd0 100644 --- a/src/asim-cvm/run_cvm.py +++ b/src/asim-cvm/run_cvm.py @@ -3,23 +3,40 @@ from pathlib import Path import pytest import activitysim.abm # register components # noqa: F401 +from activitysim.core import mp_tasks +from activitysim.cli.run import INJECTABLES -# TODO: should this just be its own installable package? -ext_dir = str(Path(__file__).parents) -sys.path.insert(0, ext_dir) -extensions = importlib.import_module("extensions") -sys.path = sys.path[1:] +def main(): + # TODO: should this just be its own installable package? + ext_dir = str(Path(__file__).parents) + sys.path.insert(0, ext_dir) + extensions = importlib.import_module("extensions") + sys.path = sys.path[1:] -arguments = sys.argv + arguments = sys.argv -data_dir = tuple(arguments[1].split(",")) -configs_dir = tuple(arguments[2].split(",")) -output_dir = arguments[3] + data_dir = tuple(arguments[1].split(",")) + configs_dir = tuple(arguments[2].split(",")) + output_dir = arguments[3] -state = extensions.cvm_state.State.make_default( - __file__, configs_dir=configs_dir, data_dir=data_dir, output_dir=output_dir -) + state = extensions.cvm_state.State.make_default( + __file__, configs_dir=configs_dir, data_dir=data_dir, output_dir=output_dir + ) -state.logging.config_logger() + state.import_extensions("extensions") + state.logging.config_logger() -state.run(models=state.settings.models) + if state.settings.multiprocess: + injectables = {} + for k in INJECTABLES: + try: + injectables[k] = state.get_injectable(k) + except KeyError: + pass + injectables["settings"] = state.settings + mp_tasks.run_multiprocess(state, injectables) + else: + state.run(models=state.settings.models) + +if __name__ == "__main__": + main() \ No newline at end of file From aad2cc067e23656e4d07544d8a3021400bbfc882 Mon Sep 17 00:00:00 2001 From: Yue Shuai <48269801+yueshuaing@users.noreply.github.com> Date: Wed, 5 Aug 2026 19:02:42 -0700 Subject: [PATCH 2/6] update commercial_accessibility in pipeline --- src/asim-cvm/extensions/cvm_route_generation.py | 1 + src/asim-cvm/extensions/cvm_route_stops.py | 3 ++- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/src/asim-cvm/extensions/cvm_route_generation.py b/src/asim-cvm/extensions/cvm_route_generation.py index 6832196d7..00ddad6f6 100644 --- a/src/asim-cvm/extensions/cvm_route_generation.py +++ b/src/asim-cvm/extensions/cvm_route_generation.py @@ -169,6 +169,7 @@ def route_generation( # sum the two logsums to get the accessibility accessibility_df["accessibility"] = accessibility_df['estab_acc_logsum'] + accessibility_df['household_acc_logsum'] + state.add_table("commercial_accessibility", accessibility_df) # get the accessibility column for the industry join_df = establishments_df.merge( diff --git a/src/asim-cvm/extensions/cvm_route_stops.py b/src/asim-cvm/extensions/cvm_route_stops.py index 26da0bacd..60135c6a6 100644 --- a/src/asim-cvm/extensions/cvm_route_stops.py +++ b/src/asim-cvm/extensions/cvm_route_stops.py @@ -622,7 +622,8 @@ def route_stops( np.exp(accessibility_df['estab_acc_hh_food']) + np.exp(accessibility_df['estab_acc_hh_package']) ) - + state.add_table("commercial_accessibility", accessibility_df) + route_trip_num = 1 cv_trips = [] From 750e849534d33b942993ff0f46cdc20402861ff0 Mon Sep 17 00:00:00 2001 From: Yue Shuai <48269801+yueshuaing@users.noreply.github.com> Date: Thu, 6 Aug 2026 15:58:06 -0700 Subject: [PATCH 3/6] use reproducible per row random generator instead of global seed --- src/asim-cvm/extensions/cvm_route_stops.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/asim-cvm/extensions/cvm_route_stops.py b/src/asim-cvm/extensions/cvm_route_stops.py index 60135c6a6..5193e6b00 100644 --- a/src/asim-cvm/extensions/cvm_route_stops.py +++ b/src/asim-cvm/extensions/cvm_route_stops.py @@ -549,12 +549,14 @@ def _dwell_time( "max_duration" ] - random_dwell_times = scipy.stats.beta.rvs( + random_order = state.get_rn_generator().random_for_df(df) + + random_dwell_times = scipy.stats.beta.pff( + random_order[:,0], a=alpha, b=beta, loc=model_settings.min_dwell_time, scale=max_duration, - size=len(df), ) result_list.append( @@ -750,8 +752,6 @@ def route_stops( nonterminated_routes[model_settings.location_settings_estb.RESULT_COL_NAME] = next_stop_location - np.random.seed(seed=42) - # Choose dwell time nonterminated_routes = _dwell_time( state, From d30cbe4eef48b0d8d14caef93aa6559a6f7b6f98 Mon Sep 17 00:00:00 2001 From: Yue Shuai <48269801+yueshuaing@users.noreply.github.com> Date: Wed, 12 Aug 2026 17:17:22 -0700 Subject: [PATCH 4/6] add commercial_accessibility to multiprocessing --- src/asim-cvm/configs/settings.yaml | 14 +++++++++++++- .../extensions/cvm_establishment_attractor.py | 4 ++++ src/asim-cvm/extensions/cvm_route_stops.py | 2 +- 3 files changed, 18 insertions(+), 2 deletions(-) diff --git a/src/asim-cvm/configs/settings.yaml b/src/asim-cvm/configs/settings.yaml index e9581dff4..25abe5424 100644 --- a/src/asim-cvm/configs/settings.yaml +++ b/src/asim-cvm/configs/settings.yaml @@ -84,7 +84,6 @@ output_tables: - establishments - establishments_all - models: - household_attractor - establishment_attractor @@ -113,6 +112,19 @@ chunk_training_mode: disabled multiprocess_steps: - name: mp_initialize begin: household_attractor + - name: mp_accessibility + begin: cvm_accessibility + slice: + tables: + - commercial_accessibility + exclude: True + - name: mp_route_generation + begin: route_generation + slice: + tables: + - establishments + - name: mp_route_generation_tnc + begin: route_generation_tnc - name: mp_route begin: route_purpose_and_vehicle slice: diff --git a/src/asim-cvm/extensions/cvm_establishment_attractor.py b/src/asim-cvm/extensions/cvm_establishment_attractor.py index 7efc92a96..3de078733 100644 --- a/src/asim-cvm/extensions/cvm_establishment_attractor.py +++ b/src/asim-cvm/extensions/cvm_establishment_attractor.py @@ -200,3 +200,7 @@ def establishment_attractor( # write land use table back to state state.add_table("land_use", land_use) + + # instantiate commercial_accessibility (must be checkpointed to be used + # to slice it) + state.get_dataframe("commercial_accessibility") diff --git a/src/asim-cvm/extensions/cvm_route_stops.py b/src/asim-cvm/extensions/cvm_route_stops.py index 5193e6b00..2c68911ad 100644 --- a/src/asim-cvm/extensions/cvm_route_stops.py +++ b/src/asim-cvm/extensions/cvm_route_stops.py @@ -551,7 +551,7 @@ def _dwell_time( random_order = state.get_rn_generator().random_for_df(df) - random_dwell_times = scipy.stats.beta.pff( + random_dwell_times = scipy.stats.beta.ppf( random_order[:,0], a=alpha, b=beta, From dbb6457947e26f624fa8f4fa7dde1aa40b49caeb Mon Sep 17 00:00:00 2001 From: Yue Shuai <48269801+yueshuaing@users.noreply.github.com> Date: Wed, 12 Aug 2026 17:19:58 -0700 Subject: [PATCH 5/6] register cvm tables in State --- src/asim-cvm/extensions/cvm_establishments.py | 4 ---- .../extensions/cvm_route_generation.py | 5 ----- src/asim-cvm/extensions/cvm_state.py | 22 +++++++++++++++++++ 3 files changed, 22 insertions(+), 9 deletions(-) diff --git a/src/asim-cvm/extensions/cvm_establishments.py b/src/asim-cvm/extensions/cvm_establishments.py index 1f7c1389b..f60a572a5 100644 --- a/src/asim-cvm/extensions/cvm_establishments.py +++ b/src/asim-cvm/extensions/cvm_establishments.py @@ -9,15 +9,11 @@ from activitysim.core import tracing, workflow from activitysim.core.input import read_input_table -from activitysim.abm.models.util import canonical_ids as cid from .cvm_state import State logger = logging.getLogger(__name__) -cid.RANDOM_CHANNELS.append("establishments") -cid.TRACEABLE_TABLES.append("establishments") -cid.CANONICAL_TABLE_INDEX_NAMES.setdefault("establishments", "establishment_id") @workflow.table def establishments(state: State) -> pd.DataFrame: diff --git a/src/asim-cvm/extensions/cvm_route_generation.py b/src/asim-cvm/extensions/cvm_route_generation.py index 00ddad6f6..f7d2a7a75 100644 --- a/src/asim-cvm/extensions/cvm_route_generation.py +++ b/src/asim-cvm/extensions/cvm_route_generation.py @@ -23,7 +23,6 @@ from activitysim.abm.tables.util import simple_table_join from activitysim.core import workflow, tracing -from activitysim.abm.models.util import canonical_ids as cid from .cvm_enum import BusinessTypes from .cvm_enum_tools import as_int_enum @@ -33,10 +32,6 @@ _business_type_offset = int(10 ** np.ceil(np.log10(max(BusinessTypes)))) -cid.RANDOM_CHANNELS.append("routes") -cid.TRACEABLE_TABLES.append("routes") -cid.CANONICAL_TABLE_INDEX_NAMES.setdefault("routes", "route_id") - @workflow.step def route_generation( state: State, diff --git a/src/asim-cvm/extensions/cvm_state.py b/src/asim-cvm/extensions/cvm_state.py index 56c0e2388..6a6628121 100644 --- a/src/asim-cvm/extensions/cvm_state.py +++ b/src/asim-cvm/extensions/cvm_state.py @@ -1,5 +1,6 @@ from activitysim.core.workflow import State as workflow_State from activitysim.core.workflow.state import StateAttr +from activitysim.abm.models.util import canonical_ids as cid from .cvm_settings import CVMSettings @@ -10,3 +11,24 @@ class State(workflow_State): """ settings: CVMSettings = StateAttr(CVMSettings) + + # cvm tables that need RNG channel / traceable-table / canonical + # index registration on every process start (single or multiprocess), + # since checkpoint reloads only tables listed here. + _CVM_TABLES = [ + ("establishments", "establishment_id"), + ("establishments_all", "establishment_id_all"), + ("routes", "route_id"), + ] + + @classmethod + def _register_cvm_tables(cls): + for table_name, index_name in cls._CVM_TABLES: + if table_name not in cid.RANDOM_CHANNELS: + cid.RANDOM_CHANNELS.append(table_name) + if table_name not in cid.TRACEABLE_TABLES: + cid.TRACEABLE_TABLES.append(table_name) + cid.CANONICAL_TABLE_INDEX_NAMES.setdefault(table_name, index_name) + + +State._register_cvm_tables() \ No newline at end of file From f26945d554543eecb797b32073cce0e02e58f9c5 Mon Sep 17 00:00:00 2001 From: Yue Shuai <48269801+yueshuaing@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:31:51 -0700 Subject: [PATCH 6/6] add comments for tnc single-process --- src/asim-cvm/configs/settings.yaml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/asim-cvm/configs/settings.yaml b/src/asim-cvm/configs/settings.yaml index 25abe5424..526c5c9e0 100644 --- a/src/asim-cvm/configs/settings.yaml +++ b/src/asim-cvm/configs/settings.yaml @@ -84,6 +84,7 @@ output_tables: - establishments - establishments_all + models: - household_attractor - establishment_attractor @@ -123,6 +124,8 @@ multiprocess_steps: slice: tables: - establishments + # route_generation_tnc single-process only + # land_use aggregation and duplicate establishment_id under slicing - name: mp_route_generation_tnc begin: route_generation_tnc - name: mp_route