Spaces:
Running on Zero
Running on Zero
File size: 4,796 Bytes
1676aa7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 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 | """
Deterministic Resource Optimizer.
This is a real integer linear program, solved with PuLP (CBC backend) —
not a heuristic dressed up to look like one. No language model output
ever reaches this file except through the NeedProfile.priority_score
field, which was itself computed by pure Python in need_engine.py.
FORMULATION
-----------
Decision variables:
x[loc][resource] = integer number of units of `resource` assigned to `loc`
0 <= x[loc][resource] <= required[loc][resource] (never over-allocate; no waste)
Constraints:
for each resource type r:
sum over all locations of x[loc][r] <= available[r]
Objective (maximize):
sum over loc, r of priority_score[loc] * x[loc][r]
This says: every unit of any resource sent to a higher-priority location is
worth more than the same unit sent to a lower-priority location, and the
solver is free to trade off between resource types and locations to
maximize total weighted need covered, subject to the hard resource caps.
Re-solving takes milliseconds for problems this size (5-8 locations x 3
resource types), which is what makes the live Human Override panel
(section 13 of the spec) genuinely interactive rather than a fake animation.
"""
from __future__ import annotations
import pulp
from core.schemas import NeedProfile, ResourcePool, AllocationPlan, LocationAllocation
RESOURCE_TYPES = ["medical_teams", "rescue_teams", "supply_trucks"]
_REQUIRED_FIELD = {
"medical_teams": "required_medical_teams",
"rescue_teams": "required_rescue_teams",
"supply_trucks": "required_supply_trucks",
}
def solve_allocation(profiles: list[NeedProfile], pool: ResourcePool) -> AllocationPlan:
if not profiles:
return AllocationPlan(
allocations=[], overall_coverage_pct=0.0,
resources_used=ResourcePool(0, 0, 0), resources_available=pool,
solver_status="NoLocations", objective_value=0.0,
)
prob = pulp.LpProblem("disaster_resource_allocation", pulp.LpMaximize)
x = {}
for p in profiles:
for r in RESOURCE_TYPES:
required = getattr(p, _REQUIRED_FIELD[r])
x[(p.location_id, r)] = pulp.LpVariable(
f"x_{p.location_id}_{r}", lowBound=0, upBound=max(required, 0), cat="Integer"
)
# Objective: maximize total priority-weighted units allocated
prob += pulp.lpSum(
p.priority_score * x[(p.location_id, r)]
for p in profiles for r in RESOURCE_TYPES
)
# Constraints: cannot exceed available pool per resource type
available = {"medical_teams": pool.medical_teams, "rescue_teams": pool.rescue_teams,
"supply_trucks": pool.supply_trucks}
for r in RESOURCE_TYPES:
prob += pulp.lpSum(x[(p.location_id, r)] for p in profiles) <= available[r], f"cap_{r}"
solver = pulp.PULP_CBC_CMD(msg=False)
prob.solve(solver)
status = pulp.LpStatus[prob.status]
allocations = []
total_required_units = 0
total_assigned_units = 0
used = {"medical_teams": 0, "rescue_teams": 0, "supply_trucks": 0}
for p in profiles:
assigned = {r: int(round(x[(p.location_id, r)].value() or 0)) for r in RESOURCE_TYPES}
required = {r: getattr(p, _REQUIRED_FIELD[r]) for r in RESOURCE_TYPES}
for r in RESOURCE_TYPES:
used[r] += assigned[r]
req_total = sum(required.values())
assigned_total = sum(assigned.values())
coverage = round((assigned_total / req_total * 100), 1) if req_total > 0 else 100.0
total_required_units += req_total
total_assigned_units += assigned_total
allocations.append(LocationAllocation(
location_id=p.location_id,
assigned_medical_teams=assigned["medical_teams"],
assigned_rescue_teams=assigned["rescue_teams"],
assigned_supply_trucks=assigned["supply_trucks"],
required_medical_teams=required["medical_teams"],
required_rescue_teams=required["rescue_teams"],
required_supply_trucks=required["supply_trucks"],
coverage_pct=coverage,
unmet_medical=max(0, required["medical_teams"] - assigned["medical_teams"]),
unmet_rescue=max(0, required["rescue_teams"] - assigned["rescue_teams"]),
unmet_supply=max(0, required["supply_trucks"] - assigned["supply_trucks"]),
))
overall_coverage = round((total_assigned_units / total_required_units * 100), 1) if total_required_units > 0 else 100.0
return AllocationPlan(
allocations=allocations,
overall_coverage_pct=overall_coverage,
resources_used=ResourcePool(**used),
resources_available=pool,
solver_status=status,
objective_value=pulp.value(prob.objective) or 0.0,
)
|