1010from activitysim .core import config
1111from activitysim .core import inject
1212from activitysim .core import pipeline
13+ from activitysim .core import chunk
1314from activitysim .core import mem
1415
1516from activitysim .core import los
1819logger = logging .getLogger (__name__ )
1920
2021
21- # class AccessibilitySkims(object):
22- # """
23- # Wrapper for skim arrays to facilitate use of skims by accessibility model
24- #
25- # Parameters
26- # ----------
27- # skims : 2D array
28- # omx: open omx file object
29- # this is only used to load skims on demand that were not preloaded
30- # length: int
31- # number of zones in skim to return in skim matrix
32- # in case the skims contain additional external zones that should be trimmed out so skim
33- # array is correct shape to match (flattened) O-D tiled columns in the od dataframe
34- # transpose: bool
35- # whether to transpose the matrix before flattening. (i.e. act as a D-O instead of O-D skim)
36- # """
37- #
38- # def __init__(self, skim_dict, orig_zones, dest_zones, transpose=False):
39- #
40- # logger.info(f"init AccessibilitySkims with {len(dest_zones)} dest zones {len(orig_zones)} orig zones")
41- #
42- # assert len(orig_zones) <= len(dest_zones)
43- # assert np.isin(orig_zones, dest_zones).all()
44- # assert len(np.unique(orig_zones)) == len(orig_zones)
45- # assert len(np.unique(dest_zones)) == len(dest_zones)
46- #
47- # self.skim_dict = skim_dict
48- # self.transpose = transpose
49- #
50- # num_skim_zones = skim_dict.get_skim_info('omx_shape')[0]
51- # if num_skim_zones == len(orig_zones) and skim_dict.offset_mapper.offset_series is None:
52- # # no slicing required because whatever the offset_int, the skim data aligns with zone list
53- # self.map_data = False
54- # else:
55- #
56- # logger.debug("AccessibilitySkims - applying offset_mapper")
57- #
58- # skim_index = list(range(num_skim_zones))
59- # orig_map = skim_dict.offset_mapper.map(orig_zones)
60- # dest_map = skim_dict.offset_mapper.map(dest_zones)
61- #
62- # # (we might be sliced multiprocessing)
63- # # assert np.isin(skim_index, orig_map).all()
64- #
65- # out_of_bounds = ~np.isin(skim_index, dest_map)
66- # # if out_of_bounds.any():
67- # # print(f"{(out_of_bounds).sum()} skim zones not in dest_map")
68- # # print(f"dest_zones {dest_zones}")
69- # # print(f"dest_map {dest_map}")
70- # # print(f"skim_index {skim_index}")
71- # assert not out_of_bounds.any(), \
72- # f"AccessibilitySkims {(out_of_bounds).sum()} skim zones not in dest_map: {np.ix_(out_of_bounds)[0]}"
73- #
74- # self.map_data = True
75- # self.orig_map = orig_map
76- # self.dest_map = dest_map
77- #
78- # def __getitem__(self, key):
79- # """
80- # accessor to return flattened skim array with specified key
81- # flattened array will have length length*length and will match tiled OD df used by assign
82- #
83- # this allows the skim array to be accessed from expressions as
84- # skim['DISTANCE'] or skim[('SOVTOLL_TIME', 'MD')]
85- # """
86- #
87- # data = self.skim_dict.get(key).data
88- #
89- # if self.transpose:
90- # data = data.transpose()
91- #
92- # if self.map_data:
93- # # slice skim to include only orig rows and dest columns
94- # # 2-d boolean slicing in numpy is a bit tricky
95- # # data = data[orig_map, dest_map] # <- WRONG!
96- # # data = data[orig_map, :][:, dest_map] # <- RIGHT
97- # # data = data[np.ix_(orig_map, dest_map)] # <- ALSO RIGHT
98- #
99- # data = data[self.orig_map, :][:, self.dest_map]
100- #
101- # return data.flatten()
22+ def compute_accessibilities_for_zones (
23+ accessibility_df ,
24+ land_use_df ,
25+ assignment_spec ,
26+ constants ,
27+ network_los ,
28+ trace_od ,
29+ trace_label ):
10230
103-
104- @inject .step ()
105- def compute_accessibility (accessibility , network_los , land_use , trace_od ):
106-
107- """
108- Compute accessibility for each zone in land use file using expressions from accessibility_spec
109-
110- The actual results depend on the expressions in accessibility_spec, but this is initially
111- intended to permit implementation of the mtc accessibility calculation as implemented by
112- Accessibility.job
113-
114- Compute measures of accessibility used by the automobile ownership model.
115- The accessibility measure first multiplies an employment variable by a mode-specific decay
116- function. The product reflects the difficulty of accessing the activities the farther
117- (in terms of round-trip travel time) the jobs are from the location in question. The products
118- to each destination zone are next summed over each origin zone, and the logarithm of the
119- product mutes large differences. The decay function on the walk accessibility measure is
120- steeper than automobile or transit. The minimum accessibility is zero.
121- """
122-
123- trace_label = 'compute_accessibility'
124- model_settings = config .read_model_settings ('accessibility.yaml' )
125- assignment_spec = assign .read_assignment_spec (config .config_file_path ('accessibility.csv' ))
126-
127- accessibility_df = accessibility .to_frame ()
128-
129- logger .info ("Running %s with %d dest zones" % (trace_label , len (accessibility_df )))
130-
131- constants = config .get_model_constants (model_settings )
132-
133- land_use_columns = model_settings .get ('land_use_columns' , [])
134- land_use_df = land_use .to_frame ()
135- land_use_df = land_use_df [land_use_columns ]
136-
137- # don't assume they are the same: accessibility may be sliced if we are multiprocessing
13831 orig_zones = accessibility_df .index .values
13932 dest_zones = land_use_df .index .values
14033
14134 orig_zone_count = len (orig_zones )
14235 dest_zone_count = len (dest_zones )
14336
144- logger .info ("Running %s with %d dest zones %d orig zones" %
145- (trace_label , dest_zone_count , orig_zone_count ))
37+ logger .info ("Running %s with %d orig zones %d dest zones" %
38+ (trace_label , orig_zone_count , dest_zone_count ))
14639
14740 # create OD dataframe
14841 od_df = pd .DataFrame (
@@ -160,36 +53,34 @@ def compute_accessibility(accessibility, network_los, land_use, trace_od):
16053
16154 # merge land_use_columns into od_df
16255 od_df = pd .merge (od_df , land_use_df , left_on = 'dest' , right_index = True ).sort_index ()
56+ chunk .log_df (trace_label , "od_df" , od_df )
16357
16458 locals_d = {
16559 'log' : np .log ,
16660 'exp' : np .exp ,
16761 'network_los' : network_los ,
16862 }
63+ locals_d .update (constants )
16964
17065 skim_dict = network_los .get_default_skim_dict ()
17166 locals_d ['skim_od' ] = skim_dict .wrap ('orig' , 'dest' ).set_df (od_df )
17267 locals_d ['skim_do' ] = skim_dict .wrap ('dest' , 'orig' ).set_df (od_df )
17368
17469 if network_los .zone_system == los .THREE_ZONE :
175- locals_d ['tvpb' ] = TransitVirtualPathBuilder (network_los )
176-
177- if constants is not None :
178- locals_d .update (constants )
70+ locals_d ['tvpb' ] = network_los .tvpb
17971
18072 results , trace_results , trace_assigned_locals \
181- = assign .assign_variables (assignment_spec , od_df , locals_d , trace_rows = trace_od_rows )
73+ = assign .assign_variables (assignment_spec , od_df , locals_d ,
74+ trace_rows = trace_od_rows , trace_label = trace_label , chunk_log = True )
18275
76+ chunk .log_df (trace_label , "results" , results )
77+
78+ # accessibility_df = accessibility_df.copy()
18379 for column in results .columns :
18480 data = np .asanyarray (results [column ])
18581 data .shape = (orig_zone_count , dest_zone_count ) # (o,d)
18682 accessibility_df [column ] = np .log (np .sum (data , axis = 1 ) + 1 )
18783
188- logger .info ("{trace_label} added {len(results.columns} columns" )
189-
190- # - write table to pipeline
191- pipeline .replace_table ("accessibility" , accessibility_df )
192-
19384 if trace_od :
19485
19586 if not trace_od_rows .any ():
@@ -208,3 +99,102 @@ def compute_accessibility(accessibility, network_los, land_use, trace_od):
20899
209100 if trace_assigned_locals :
210101 tracing .write_csv (trace_assigned_locals , file_name = "accessibility_locals" )
102+
103+ return (accessibility_df )
104+
105+
106+ def accessibility_calc_row_size (accessibility_df , land_use_df , assignment_spec , network_los , trace_label ):
107+ """
108+ rows_per_chunk calculator for accessibility
109+ """
110+
111+ sizer = chunk .RowSizeEstimator (trace_label )
112+
113+ # if there are skims, and zone_system is THREE_ZONE, and there are any
114+ # then we want to estimate the per-row overhead tvpb skims
115+ # (do this first to facilitate tracing of rowsize estimation below)
116+ if network_los .zone_system == los .THREE_ZONE :
117+ # DISABLE_TVPB_OVERHEAD
118+ logger .debug ("disable calc_row_size for THREE_ZONE with tap skims" )
119+ return 0
120+
121+ land_use_rows = len (land_use_df .index )
122+ land_use_columns = len (land_use_df .columns )
123+ od_columns = 2
124+
125+ # assignment spec has one row per value to assign
126+ # count number of unique persistent assign_variables targets simultaneously resident during spec eval
127+ # (since dict overwrites recurring targets, only count unique targets)
128+ def is_persistent (target ):
129+ return not (assign .is_throwaway (target ) or assign .is_temp_scalar (target ))
130+ num_spec_values = len ([target for target in assignment_spec .target .unique () if is_persistent (target )])
131+
132+ sizer .add_elements (land_use_rows * od_columns , 'od_df' )
133+
134+ # each od_df joins to all land_use zones
135+ sizer .add_elements (land_use_rows * land_use_columns , 'land_use_choosers' )
136+
137+ # and then we assign_variables to joined land_use from assignment_spec
138+ sizer .add_elements (land_use_rows * num_spec_values , 'spec_values' )
139+
140+ row_size = sizer .get_hwm ()
141+ return row_size
142+
143+
144+ @inject .step ()
145+ def compute_accessibility (land_use , accessibility , network_los , chunk_size , trace_od ):
146+
147+ """
148+ Compute accessibility for each zone in land use file using expressions from accessibility_spec
149+
150+ The actual results depend on the expressions in accessibility_spec, but this is initially
151+ intended to permit implementation of the mtc accessibility calculation as implemented by
152+ Accessibility.job
153+
154+ Compute measures of accessibility used by the automobile ownership model.
155+ The accessibility measure first multiplies an employment variable by a mode-specific decay
156+ function. The product reflects the difficulty of accessing the activities the farther
157+ (in terms of round-trip travel time) the jobs are from the location in question. The products
158+ to each destination zone are next summed over each origin zone, and the logarithm of the
159+ product mutes large differences. The decay function on the walk accessibility measure is
160+ steeper than automobile or transit. The minimum accessibility is zero.
161+ """
162+
163+ trace_label = 'compute_accessibility'
164+ model_settings = config .read_model_settings ('accessibility.yaml' )
165+ assignment_spec = assign .read_assignment_spec (config .config_file_path ('accessibility.csv' ))
166+
167+ accessibility_df = accessibility .to_frame ()
168+ if len (accessibility_df .columns ) > 0 :
169+ logger .warning (f"accessibility table is not empty. Columns:{ list (accessibility_df .columns )} )
170+ raise RuntimeError (f"accessibility table is not empty." )
171+
172+ constants = config .get_model_constants (model_settings )
173+
174+ # only include the land_use columns needed by spec, as specified by land_use_columns model_setting
175+ land_use_columns = model_settings .get ('land_use_columns' , [])
176+ land_use_df = land_use .to_frame ()
177+ land_use_df = land_use_df [land_use_columns ]
178+
179+ logger .info (f"Running { trace_label } { len (accessibility_df .index )} { len (land_use_df )} )
180+
181+ row_size = \
182+ chunk_size and accessibility_calc_row_size (accessibility_df , land_use_df ,
183+ assignment_spec , network_los , trace_label )
184+
185+ accessibilities_list = []
186+
187+ for i , chooser_chunk , chunk_trace_label in \
188+ chunk .adaptive_chunked_choosers (accessibility_df , chunk_size , row_size , trace_label ):
189+
190+ accessibilities = \
191+ compute_accessibilities_for_zones (chooser_chunk , land_use_df , assignment_spec ,
192+ constants , network_los , trace_od , trace_label )
193+ accessibilities_list .append (accessibilities )
194+
195+ accessibility_df = pd .concat (accessibilities_list )
196+
197+ logger .info (f"{ trace_label } { accessibility_df .shape } )
198+
199+ # - write table to pipeline
200+ pipeline .replace_table ("accessibility" , accessibility_df )
0 commit comments