66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from itertools import combinations
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from .models import SafetyPolicy, SwarmSolution
|
|
|
|
|
|
class VerificationReport(BaseModel):
|
|
passed: bool
|
|
violations: list[str] = Field(default_factory=list)
|
|
|
|
|
|
def _current_position(robot: Any) -> float:
|
|
current_pos = robot.current_pos
|
|
if current_pos is None:
|
|
current_pos = robot.initial_pos
|
|
assert current_pos is not None
|
|
return float(current_pos)
|
|
|
|
|
|
def verify_swarm_solution(solution: SwarmSolution, policy: SafetyPolicy) -> VerificationReport:
|
|
violations: list[str] = []
|
|
robots = solution.robots
|
|
|
|
if solution.certificate.iterations != len(solution.audit_log):
|
|
violations.append("certificate iteration count does not match audit log")
|
|
|
|
expected_version = solution.shared.version
|
|
if not solution.audit_log:
|
|
violations.append("audit log is empty")
|
|
else:
|
|
for expected_iteration, entry in enumerate(solution.audit_log):
|
|
if entry.iteration != expected_iteration:
|
|
violations.append("audit log iterations are not contiguous")
|
|
break
|
|
if solution.audit_log[-1].shared_version != expected_version:
|
|
violations.append("final shared version does not match audit log")
|
|
|
|
average_position = sum(_current_position(robot) for robot in robots) / len(robots) if robots else 0.0
|
|
if abs(solution.consensus - average_position) > policy.max_travel_distance:
|
|
violations.append("consensus diverged from robot state")
|
|
|
|
for robot in robots:
|
|
current_pos = _current_position(robot)
|
|
if not policy.verify(robot.initial_pos, current_pos):
|
|
violations.append(f"{robot.robot_id}: travel bound exceeded")
|
|
if robot.travel_distance > policy.energy_budget:
|
|
violations.append(f"{robot.robot_id}: energy budget exceeded")
|
|
|
|
for left, right in combinations(robots, 2):
|
|
if abs(_current_position(left) - _current_position(right)) < policy.min_separation:
|
|
violations.append(
|
|
f"{left.robot_id}/{right.robot_id}: separation below {policy.min_separation}"
|
|
)
|
|
|
|
if not solution.certificate.converged:
|
|
violations.append("solver did not converge")
|
|
|
|
if solution.audit_log and solution.shared.delta_clock < len(solution.audit_log[0].applied_deltas):
|
|
violations.append("delta clock regressed")
|
|
|
|
return VerificationReport(passed=not violations, violations=violations)
|