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
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540 | class CareHomeDistributor:
""" """
def __init__(
self,
care_homes_file: str = None,
workers_sector="Q",
debug_mode: bool = False,
):
"""
Tool to distribute unallocated people into care homes based on age proportions.
Parameters
----------
care_homes_file
Path to care homes CSV with age proportion data
workers_sector
Sector code for care home workers
debug_mode
Enable detailed tracking and CSV export for development/testing
"""
self.care_homes_file = care_homes_file
self.workers_sector = workers_sector
self.debug_mode = debug_mode
self.debug_data = [] # Store debug information
def _get_mixed_geography_data(self, super_areas):
"""Load and merge care homes data from multiple geography files based on super areas.
Args:
super_areas:
"""
if not super_areas:
return paths.data_path / "input/care_homes/care_homes_sct.csv"
# Detect all unique geographies present in the super areas
geographies = set()
for super_area in super_areas:
if hasattr(super_area, 'areas'):
for area in super_area.areas:
area_name = area.name
if area_name.startswith(('E0', 'W0')):
geographies.add('ew')
elif area_name.startswith('S0'):
geographies.add('sct')
elif area_name.startswith('N0'):
geographies.add('ni')
# If only one geography, return the appropriate single file
if len(geographies) == 1:
geo = list(geographies)[0]
if geo == 'ew':
return paths.data_path / "input/care_homes/care_homes_ew.csv"
elif geo == 'sct':
return paths.data_path / "input/care_homes/care_homes_sct.csv"
elif geo == 'ni':
return paths.data_path / "input/care_homes/care_homes_NI.csv"
# For mixed geographies, we need to create a combined file path indicator
# The actual loading and merging will be handled elsewhere since this method
# was designed to return a single file path
return "MIXED_GEOGRAPHIES"
def distribute_workers_to_care_homes(self, super_areas: SuperAreas):
"""
Args:
super_areas (SuperAreas):
"""
# Auto-detect care homes file on first use if not specified
if self.care_homes_file is None:
mixed_geo_result = self._get_mixed_geography_data(super_areas)
if mixed_geo_result == "MIXED_GEOGRAPHIES":
logger.info("Mixed geographies detected - care home data should be loaded by CareHomes.for_areas")
# For mixed geographies, we don't need to load a specific file
# as the care homes should already be created with proper data
self.care_homes_file = "MIXED_GEOGRAPHIES"
else:
self.care_homes_file = mixed_geo_result
logger.info(f"Distributing workers to care homes...")
care_home_data = [] # Collect data for visualization
for super_area in super_areas:
care_homes = []
for area in super_area.areas:
care_homes.extend(area.care_homes)
if not care_homes:
continue
carers = [
person
for person in super_area.workers
if (
person.sector == "Q"
and person.primary_activity is None
and person.sub_sector is None
)
]
shuffle(carers)
for care_home in care_homes:
worker_ids = [] # Collect worker IDs for each care home
while len(care_home.workers) < care_home.n_workers:
try:
carer = carers.pop()
except Exception:
""" logger.info(
f"Care home in area {care_home.area.name} has not enough workers!"
) """
break
care_home.add(
person=carer,
subgroup_type=care_home.SubgroupType.workers,
activity="primary_activity",
)
care_home.add_to_registered_members(carer.id, care_home.SubgroupType.workers)
carer.lockdown_status = "key_worker"
worker_ids.append(carer.id)
# Log final worker assignment for this care home
# Append care home information for visualization
# Get information about registered members
total_registered = sum(len(members) for members in care_home.registered_members_ids.values())
all_subgroups = list(care_home.registered_members_ids.keys())
# Sample some IDs to display
sampled_ids = []
for subgroup, members in care_home.registered_members_ids.items():
if members:
# Take up to 2 from each subgroup
for member_id in members[:2]:
if subgroup == care_home.SubgroupType.workers:
sampled_ids.append(f"worker:{member_id}")
elif subgroup == care_home.SubgroupType.residents:
sampled_ids.append(f"resident:{member_id}")
elif subgroup == care_home.SubgroupType.visitors:
sampled_ids.append(f"visitor:{member_id}")
else:
sampled_ids.append(f"sg{subgroup}:{member_id}")
sampled_ids = sampled_ids[:5] # Limit to 5 total
care_home_data.append({
"| Care Home ID": care_home.id,
"| Area": care_home.area.name,
"| Total Registered": total_registered,
"| Subgroups": all_subgroups,
"| Sample Registered Member IDs": sampled_ids,
})
# Convert care home data to a DataFrame for easy visualization
df_care_homes = pd.DataFrame(care_home_data)
print("\n===== Care Home Registered Members Summary =====")
print(df_care_homes.head(10)) # Display a sample of 10 care homes for brevity
# Count total workers distributed
total_workers_distributed = sum(len(care_home.workers) for super_area in super_areas for area in super_area.areas for care_home in area.care_homes)
logger.info(f"Distributed {total_workers_distributed} workers to care homes")
def distribute_unallocated_people(self, care_homes, unallocated_people):
"""Distribute unallocated people to care homes based on each care home's stored age proportions.
Only uses people from the same area as each care home.
Args:
care_homes: CareHomes object containing available care homes
unallocated_people: Dict with structure: {area_id: {age_group: {sex: [Person, ...]}}}
Returns:
dict: Summary of people distributed to care homes
"""
logger.info(f"Distributing people to care homes...")
if not care_homes or not unallocated_people:
return {}
total_distributed = 0
distribution_summary = {'by_area': {}, 'by_age_range': {'0_15': 0, '16_24': 0, '25_34': 0, '35_49': 0, '50_64': 0, '65_99': 0}}
# Group care homes by area for efficient processing
care_homes_by_area = {}
for care_home in care_homes.members:
if care_home.size >= care_home.max_residents:
continue # Care home is full
if not hasattr(care_home, 'age_proportions'):
logger.debug(f"Care home {getattr(care_home, 'id', 'unknown')} has no age proportions - skipping")
continue
# Get area identifier from care home
area_id = getattr(care_home.area, 'name', str(getattr(care_home.area, 'id', 'unknown')))
if area_id not in care_homes_by_area:
care_homes_by_area[area_id] = []
care_homes_by_area[area_id].append(care_home)
# Process each area that has both care homes and unallocated people
for area_id, area_care_homes in care_homes_by_area.items():
if area_id not in unallocated_people:
logger.debug(f"No unallocated people in area {area_id} with care homes")
continue
area_people = unallocated_people[area_id]
area_distributed = 0
# Initialize area statistics
area_stats = self._initialize_area_stats(area_id, area_people, area_care_homes)
# Distribute people to care homes in this area
for care_home in area_care_homes:
available_beds = care_home.max_residents - care_home.size
age_proportions = care_home.age_proportions
# Calculate how many people of each age range this care home needs
needed_by_age_range = {}
for age_prop_key, proportion in age_proportions.items():
if proportion > 0:
needed_by_age_range[age_prop_key] = max(1, int(available_beds * proportion))
# Distribute people from this area to this care home
distributed_to_this_home = self._distribute_to_single_care_home_from_area(
care_home, needed_by_age_range, area_people, area_id
)
area_distributed += distributed_to_this_home
total_distributed += distributed_to_this_home
# Update care home stats
area_stats['care_homes'].append({
'id': getattr(care_home, 'id', 'unknown'),
'capacity': care_home.max_residents,
'initial_residents': care_home.size - distributed_to_this_home,
'distributed': distributed_to_this_home,
'final_residents': care_home.size,
'occupancy_rate': care_home.size / care_home.max_residents if care_home.max_residents > 0 else 0
})
# Finalize area statistics
self._finalize_area_stats(area_stats, area_people)
if area_distributed > 0:
distribution_summary['by_area'][area_id] = area_distributed
# Display area report if debug mode is enabled
if self.debug_mode:
self._display_area_report(area_stats)
logger.info(f"Total people distributed to care homes: {total_distributed}")
# Display comprehensive summary if debug mode is enabled
if self.debug_mode:
self._display_final_summary(distribution_summary, total_distributed, care_homes_by_area)
return distribution_summary
def _distribute_to_single_care_home_from_area(self, care_home, needed_by_age_range, area_people, area_id):
"""Distribute people from a specific area to a single care home based on its age range needs.
Args:
care_home: The care home object to populate
needed_by_age_range: Dict of {age_prop_key: count} needed for this care home (e.g., {'prop_0_15': 2, 'prop_65_99': 10})
area_people: Available people from this specific area: {age_group: {sex: [Person, ...]}}
area_id: ID of the area being processed
Returns:
int: Number of people distributed to this care home
"""
distributed_count = 0
# Map age proportion keys to our age groups and age ranges
age_range_mapping = {
'prop_0_15': ('kids', 0, 15),
'prop_16_24': ('young_adults', 16, 24),
'prop_25_34': ('adults', 25, 34),
'prop_35_49': ('adults', 35, 49),
'prop_50_64': ('adults', 50, 64),
'prop_65_99': ('old_adults', 65, 99)
}
# Try to fill the care home according to its age proportions
for age_prop_key, needed_count in needed_by_age_range.items():
if needed_count <= 0 or age_prop_key not in age_range_mapping:
continue
age_group, min_age, max_age = age_range_mapping[age_prop_key]
# Collect available people of this age group from this area only
available_people = []
if age_group in area_people:
for sex in ['m', 'f']:
if sex in area_people[age_group]:
# Filter by specific age range within the age group
matching_people = [p for p in area_people[age_group][sex]
if min_age <= p.age <= max_age]
available_people.extend(matching_people)
if not available_people:
continue
# Randomly select people to distribute
shuffle(available_people)
people_to_distribute = available_people[:min(needed_count, len(available_people))]
# Add people to care home and remove from unallocated lists
for person in people_to_distribute:
if care_home.size < care_home.max_residents:
care_home.add(person, subgroup_type=1) # Add as resident (subgroup_type=1)
distributed_count += 1
# Remove person from this area's unallocated people
self._remove_person_from_area(person, area_people)
if self.debug_mode:
self.debug_data.append({
'person_id': person.id,
'age': person.age,
'sex': person.sex,
'age_range': f"{min_age}-{max_age}",
'care_home_id': getattr(care_home, 'id', 'unknown'),
'area_id': area_id
})
else:
break # Care home is now full
return distributed_count
def _remove_person_from_area(self, person, area_people):
"""Remove a specific person from a single area's unallocated people structure.
Args:
person:
area_people:
"""
# Determine person's age group
age_group = self._get_person_age_group(person)
# Find and remove the person from the appropriate list in this area
if age_group in area_people:
for sex in ['m', 'f']:
if sex in area_people[age_group] and person in area_people[age_group][sex]:
area_people[age_group][sex].remove(person)
return
def _get_person_age_group(self, person):
"""Determine which age group a person belongs to.
Args:
person:
"""
age = person.age
if age <= 15:
return 'kids'
elif age <= 24:
return 'young_adults'
elif age <= 64:
return 'adults'
else:
return 'old_adults'
def _initialize_area_stats(self, area_id, area_people, _area_care_homes):
"""Initialize statistics tracking for an area.
Args:
area_id:
area_people:
_area_care_homes:
"""
# Count initial people by age group and sex
initial_people = {}
total_initial = 0
for age_group in ["kids", "young_adults", "adults", "old_adults"]:
initial_people[age_group] = {}
for sex in ["m", "f"]:
count = len(area_people.get(age_group, {}).get(sex, []))
initial_people[age_group][sex] = count
total_initial += count
return {
'area_id': area_id,
'initial_people': initial_people,
'total_initial_people': total_initial,
'care_homes': [],
'total_distributed': 0,
'remaining_people': {},
'distribution_rate': 0.0
}
def _finalize_area_stats(self, area_stats, area_people):
"""Calculate final statistics for an area.
Args:
area_stats:
area_people:
"""
# Count remaining people
remaining_people = {}
total_remaining = 0
for age_group in ["kids", "young_adults", "adults", "old_adults"]:
remaining_people[age_group] = {}
for sex in ["m", "f"]:
count = len(area_people.get(age_group, {}).get(sex, []))
remaining_people[age_group][sex] = count
total_remaining += count
area_stats['remaining_people'] = remaining_people
area_stats['total_distributed'] = area_stats['total_initial_people'] - total_remaining
area_stats['distribution_rate'] = (area_stats['total_distributed'] / area_stats['total_initial_people']) if area_stats['total_initial_people'] > 0 else 0.0
def _display_area_report(self, stats):
"""Display comprehensive care home distribution report for an area.
Args:
stats:
"""
print(f"\n{'='*100}")
print(f"🏥 CARE HOME DISTRIBUTION REPORT - AREA: {stats['area_id']}")
print(f"{'='*100}")
# Section 1: Care Home Summary
print(f"\n🏢 CARE HOMES SUMMARY:")
print(f"{'Care Home ID':<25} {'Capacity':<10} {'Initial':<10} {'Added':<8} {'Final':<8} {'Occupancy':<12}")
print("-" * 85)
total_capacity = 0
total_initial = 0
total_added = 0
total_final = 0
for ch in stats['care_homes']:
total_capacity += ch['capacity']
total_initial += ch['initial_residents']
total_added += ch['distributed']
total_final += ch['final_residents']
print(f"{ch['id']:<25} {ch['capacity']:<10} {ch['initial_residents']:<10} {ch['distributed']:<8} {ch['final_residents']:<8} {ch['occupancy_rate']:<11.1%}")
print("-" * 85)
print(f"{'TOTAL':<25} {total_capacity:<10} {total_initial:<10} {total_added:<8} {total_final:<8} {total_final/total_capacity if total_capacity > 0 else 0:<11.1%}")
# Section 2: People Distribution by Age Group
print(f"\n👥 PEOPLE DISTRIBUTION BY AGE GROUP:")
print(f"{'Age Group':<15} {'Initial M':<10} {'Initial F':<10} {'Remain M':<10} {'Remain F':<10} {'Distributed':<12} {'Rate':<8}")
print("-" * 85)
for age_group in ["kids", "young_adults", "adults", "old_adults"]:
initial_m = stats['initial_people'][age_group]['m']
initial_f = stats['initial_people'][age_group]['f']
remain_m = stats['remaining_people'][age_group]['m']
remain_f = stats['remaining_people'][age_group]['f']
total_initial_ag = initial_m + initial_f
total_remaining_ag = remain_m + remain_f
distributed_ag = total_initial_ag - total_remaining_ag
rate_ag = distributed_ag / total_initial_ag if total_initial_ag > 0 else 0
print(f"{age_group.replace('_', ' ').title():<15} {initial_m:<10} {initial_f:<10} {remain_m:<10} {remain_f:<10} {distributed_ag:<12} {rate_ag:<7.1%}")
print("-" * 85)
total_remain = sum(stats['remaining_people'][ag]['m'] + stats['remaining_people'][ag]['f'] for ag in stats['remaining_people'])
print(f"{'TOTAL':<15} {'-':<10} {'-':<10} {'-':<10} {'-':<10} {stats['total_distributed']:<12} {stats['distribution_rate']:<7.1%}")
# Section 3: Summary Statistics
print(f"\n📊 AREA SUMMARY:")
print(f"Total Initial People: {stats['total_initial_people']:>6}")
print(f"Total Distributed: {stats['total_distributed']:>6}")
print(f"Total Remaining: {total_remain:>6}")
print(f"Distribution Rate: {stats['distribution_rate']:>5.1%}")
print(f"Total Care Home Capacity: {total_capacity:>6}")
print(f"Care Home Occupancy Rate: {total_final/total_capacity if total_capacity > 0 else 0:>5.1%}")
print(f"\n{'='*100}\n")
def _display_final_summary(self, distribution_summary, total_distributed, care_homes_by_area):
"""Display final summary across all areas.
Args:
distribution_summary:
total_distributed:
care_homes_by_area:
"""
print(f"\n{'='*100}")
print(f"🎯 FINAL CARE HOME DISTRIBUTION SUMMARY")
print(f"{'='*100}")
# Summary by area
print(f"\n📍 DISTRIBUTION BY AREA:")
print(f"{'Area ID':<20} {'Care Homes':<12} {'People Distributed':<18}")
print("-" * 50)
total_care_homes = 0
for area_id, distributed_count in distribution_summary['by_area'].items():
num_care_homes = len(care_homes_by_area.get(area_id, []))
total_care_homes += num_care_homes
print(f"{area_id:<20} {num_care_homes:<12} {distributed_count:<18}")
print("-" * 50)
print(f"{'TOTAL':<20} {total_care_homes:<12} {total_distributed:<18}")
# Age distribution from debug data
if self.debug_data:
print(f"\n📊 DISTRIBUTION BY AGE RANGE:")
age_range_counts = {}
for entry in self.debug_data:
age_range = entry['age_range']
age_range_counts[age_range] = age_range_counts.get(age_range, 0) + 1
print(f"{'Age Range':<15} {'Count':<10} {'Percentage':<12}")
print("-" * 37)
for age_range, count in sorted(age_range_counts.items()):
percentage = count / total_distributed if total_distributed > 0 else 0
print(f"{age_range:<15} {count:<10} {percentage:<11.1%}")
print(f"\n🎉 Successfully distributed {total_distributed} people to care homes across {len(distribution_summary['by_area'])} areas!")
print(f"{'='*100}\n")
|