32 lines
1.3 KiB
Python
32 lines
1.3 KiB
Python
from __future__ import annotations
|
|
from typing import Dict, Any, Optional
|
|
from .contract import SafetyContract
|
|
|
|
|
|
class ShadowPlanner:
|
|
"""Toy shadow planner for MVP: suggests a safer delta to the proposed plan.
|
|
|
|
In a real system this would run a separate optimizer. Here we implement a simple heuristic:
|
|
if the plan has a 'speed' parameter, cap it to a safe maximum; otherwise, return a reduced-cost plan.
|
|
"""
|
|
|
|
SAFE_SPEED_MAX = 0.5
|
|
|
|
def propose_safe(self, plan: Dict[str, Any], state: Dict[str, Any], contract: SafetyContract) -> Optional[Dict[str, Any]]:
|
|
new_plan = plan.copy()
|
|
costs = new_plan.get("costs", {})
|
|
action = new_plan.get("action")
|
|
# Simple heuristic: if speed present, cap it; otherwise, throttle time/cost if possible
|
|
if "speed" in new_plan:
|
|
if new_plan["speed"] > self.SAFE_SPEED_MAX:
|
|
new_plan["speed"] = self.SAFE_SPEED_MAX
|
|
new_plan["costs"] = {**costs, "time": max(costs.get("time", 0), 0.1)}
|
|
return new_plan
|
|
# If no speed but time can be reduced, reduce time fractionally
|
|
if costs and costs.get("time", 0) > 0:
|
|
new_plan_costs = dict(costs)
|
|
new_plan_costs["time"] = max(costs["time"] * 0.8, 0.01)
|
|
new_plan["costs"] = new_plan_costs
|
|
return new_plan
|
|
return None
|