mp rng reset, disabling one new test for no

Author: janzill

activitysim/abm/models/location_choice.py

@@ -1040,7 +1040,8 @@ def iterate_location_choice(
             logger.debug(
                 f"{trace_label} resetting random number generator offsets for iteration {iteration}"
             )
-            state.get_rn_generator().reset_offsets_for_step(state.current_model_name)
+            # handle MP by only resetting offsets for all persons in this process.
+            state.get_rn_generator().reset_offsets_for_df(persons_merged_df)
 
         choices_df_, save_sample_df = run_location_choice(
             state,

activitysim/core/random.py

@@ -9,8 +9,8 @@
 import numpy as np
 import pandas as pd
 
-from activitysim.core.util import reindex
 from activitysim.core.exceptions import DuplicateLoadableObjectError, TableIndexError
+from activitysim.core.util import reindex
 
 from .tracing import print_elapsed_time
 
@@ -445,11 +445,9 @@ def get_channel_for_df(self, df):
             raise TableIndexError("No channel with index name '%s'" % df.index.name)
         return self.channels[channel_name]
 
-    # step handling
-
     def reset_offsets_for_step(self, step_name):
         """
-        Reset offsets for all channels for a new step
+        Reset offsets for all channels for a step
 
         Parameters
         ----------
@@ -462,6 +460,24 @@ def reset_offsets_for_step(self, step_name):
         for c in self.channels:
             self.channels[c].row_states["offset"] = 0
 
+    def reset_offsets_for_df(self, df):
+        """
+        Reset offsets for all choosers in df if the channel for a step
+
+        Parameters
+        ----------
+        step_name : str
+            pipeline step name for this step
+        df : pandas.DataFrame
+            df with index name and values corresponding to a registered channel
+        """
+        channel = self.get_channel_for_df(df)
+        channel.row_states.loc[df.index, "offset"] = 0
+        logger.info(
+            f"RNG: resetting random number generator offsets for channel '{channel.channel_name}' for {len(df)} rows"
+            + f" with index name '{df.index.name}'. Total lenght df: {len(channel.row_states)}"
+        )
+
     def begin_step(self, step_name):
         """
         Register that the pipeline has entered a new step and that global and channel streams

activitysim/examples/production_semcog/test/test_semcog.py

@@ -99,8 +99,9 @@ def test_semcog_eet():
     run_test_semcog(multiprocess=False, use_explicit_error_terms=True)
 
 
-# TODO: currently running into problems with escort trips that park at
-# university. Need to check extensions.
+# TODO: currently running into problems some trips, looks like
+# trip_purpose_and_destination, might need to look into resetting
+# RNGs there. Leaving this test disabled for now.
 # def test_semcog_mp_eet():
 #     run_test_semcog(multiprocess=True, use_explicit_error_terms=True)