18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190 | def execute_allocation_strategy(population,
venues,
household_distributor,
strategy_file: str = "data/households/allocation_strategy.yaml",
export_debug_csv: bool = False):
"""
Execute a unified allocation strategy from YAML configuration.
This function orchestrates BOTH household and venue allocations in a single
sequence defined in the YAML file.
Args:
population: PopulationManager
venues: VenueManager
household_distributor: HouseholdDistributor
strategy_file (str, optional): Path to YAML strategy file (relative or absolute). Default is "data/households/allocation_strategy.yaml".
Returns:
dict: Complete statistics for all steps
"""
logger.info("=" * 60)
logger.info("Executing Unified Allocation Strategy")
logger.info("=" * 60)
# Resolve template variables, then fall back to data/ prefix for bare relative paths
strategy_file = pr.resolve(strategy_file)
if not os.path.isabs(strategy_file):
if not os.path.exists(strategy_file):
strategy_file = f"data/{strategy_file}"
# Load strategy configuration
logger.info(f"Loading allocation strategy from {strategy_file}")
with open(strategy_file, 'r') as f:
strategy = yaml.safe_load(f)
# Check if enabled
if not strategy.get('enabled', True):
logger.info("Unified strategy is disabled, skipping")
return {}
# Get steps
steps = strategy.get('steps', [])
if not steps:
logger.warning("No allocation steps defined in strategy")
return {}
logger.info(f"Found {len(steps)} allocation steps")
logger.info("")
# Execute each step
all_stats = {}
step_number = 1
for step_config in steps:
step_type = step_config.get('type')
step_name = step_config.get('name', f'Step {step_number}')
if step_type not in ['household', 'venue', 'household_excess', 'household_overflow', 'household_promotion', 'resident_linked']:
logger.warning(f"Unknown step type '{step_type}' for step '{step_name}', skipping")
continue
logger.info("=" * 60)
logger.info(f"Step {step_number}: {step_name} ({step_type})")
logger.info("=" * 60)
description = step_config.get('description')
if description:
logger.info(f"Description: {description}")
logger.info("")
# Execute based on type
if step_type == 'household':
stats = _execute_household_step(step_config, household_distributor)
elif step_type == 'venue':
stats = _execute_venue_step(step_config, population, venues, household_distributor)
elif step_type == 'household_excess':
stats = _execute_household_excess_step(step_config, household_distributor)
elif step_type == 'household_overflow':
stats = _execute_household_overflow_step(step_config, household_distributor)
elif step_type == 'household_promotion':
stats = _execute_household_promotion_step(step_config, household_distributor)
elif step_type == 'resident_linked':
stats = _execute_resident_linked_step(step_config, population, venues, household_distributor)
all_stats[step_name] = {
'type': step_type,
'step_number': step_number,
**stats
}
step_number += 1
logger.info("")
# Print overall summary
logger.info("=" * 60)
logger.info("UNIFIED ALLOCATION STRATEGY SUMMARY")
logger.info("=" * 60)
logger.info("")
total_household_alloc = 0
total_venue_alloc = 0
total_excess_alloc = 0
total_overflow_alloc = 0
for step_name, stats in all_stats.items():
logger.info(f"{stats['step_number']}. {step_name} ({stats['type']}):")
if stats['type'] == 'household':
households_created = stats.get('households_created', 0)
people_allocated = stats.get('people_allocated_this_round', 0)
logger.info(f" Households: {households_created:,}")
logger.info(f" People: {people_allocated:,}")
total_household_alloc += people_allocated
elif stats['type'] == 'venue':
allocated = stats.get('allocated', 0)
venues_count = stats.get('venues', 0)
logger.info(f" Venues: {venues_count}")
logger.info(f" People: {allocated:,}")
total_venue_alloc += allocated
elif stats['type'] == 'household_excess':
people_added = stats.get('people_added', 0)
households_modified = stats.get('households_modified', 0)
logger.info(f" Households modified: {households_modified:,}")
logger.info(f" People added: {people_added:,}")
total_excess_alloc += people_added
elif stats['type'] == 'household_overflow':
people_added = stats.get('people_added', 0)
households_modified = stats.get('households_modified', 0)
logger.info(f" Households modified: {households_modified:,}")
logger.info(f" People added (overflow): {people_added:,}")
total_overflow_alloc += people_added
elif stats['type'] == 'household_promotion':
people_added = stats.get('people_added', 0)
households_promoted = stats.get('households_promoted', 0)
logger.info(f" Households promoted: {households_promoted:,}")
logger.info(f" People added (promotion): {people_added:,}")
total_overflow_alloc += people_added # Count with overflow
logger.info("")
logger.info("Overall Totals:")
# Get household count from VenueManager
all_households = household_distributor.venue_manager.get_venues_by_type("household")
logger.info(f" Total households: {len(all_households):,}")
logger.info(f" People in households (initial): {total_household_alloc:,}")
logger.info(f" People added to households (excess): {total_excess_alloc:,}")
logger.info(f" People added to households (overflow): {total_overflow_alloc:,}")
logger.info(f" People in venues: {total_venue_alloc:,}")
# Optional: Log resident_linked total if needed
total_resident_linked = sum(s.get('total_links', 0) for s in all_stats.values() if s['type'] == 'resident_linked')
if total_resident_linked > 0:
logger.info(f" Resident-linked connections: {total_resident_linked:,}")
logger.info(f" Total allocated: {len(household_distributor.allocated_people):,}")
logger.info(f" Remaining unallocated: {household_distributor.get_available_people_count():,}")
total_pop = len(population.get_all_people())
alloc_pct = (len(household_distributor.allocated_people) / total_pop * 100) if total_pop > 0 else 0
logger.info(f" Allocation rate: {alloc_pct:.1f}%")
logger.info("=" * 60)
# Optionally export unallocated people (skipped for large worlds —
# builds a DataFrame across every unplaced person).
if export_debug_csv:
household_distributor.export_unallocated_people_to_csv()
return all_stats
|