59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
"""A lightweight ADMM-like planner for CrisisOps MVP.
|
|
|
|
This is intentionally simple: given a set of available objects (resources)
|
|
and a set of requests (needs), it performs a basic allocation to maximize
|
|
reliability of critical deliveries while minimizing idle trips in a toy setting.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Dict, List
|
|
|
|
from .core import GraphOfContracts, PlanDelta
|
|
|
|
|
|
class ADMMPlanner:
|
|
def __init__(self):
|
|
# In a real system this would persist state; for MVP we keep in-memory.
|
|
self._state: Dict[str, Dict[str, int]] = {}
|
|
|
|
def reset(self) -> None:
|
|
self._state.clear()
|
|
|
|
def optimize(self, gof: GraphOfContracts, max_iter: int = 5) -> PlanDelta:
|
|
"""Run a toy optimization pass over the GoC to produce a PlanDelta.
|
|
|
|
We allocate available objects to matching morphisms (requests) by a
|
|
simplistic rule: if a source object and target object share a compatible
|
|
type and the requested quantity is available, we create an action.
|
|
"""
|
|
actions: List[Dict[str, object]] = []
|
|
now = time.time()
|
|
|
|
# Example heuristic: pair objects by type and create dispatch actions
|
|
# Primitive: for each morphism, try to assign a matching object as resource.
|
|
for mid, morph in gof.morphisms.items():
|
|
src = gof.objects.get(morph.source_id)
|
|
dst = gof.objects.get(morph.target_id)
|
|
if not src or not dst:
|
|
continue
|
|
# Simple compatibility rule: if both share a category/role (type)
|
|
if src.type == dst.type:
|
|
action = {
|
|
"morphism_id": mid,
|
|
"action": "dispatch",
|
|
"resource_id": src.id,
|
|
"destination_id": dst.id,
|
|
"timestamp": now,
|
|
"details": {
|
|
"src_type": src.type,
|
|
"dst_type": dst.type,
|
|
},
|
|
}
|
|
actions.append(action)
|
|
|
|
# Record plan delta
|
|
plan = PlanDelta(timestamp=now, actions=actions)
|
|
gof.plan_delta = plan
|
|
return plan
|