# Autonomous Swarm Urban Car Search: First 30 Minutes Developer Wiki
> A fast onboarding guide to the decentralized multi-robot car search system, navigating Ant Colony Optimization, multi-layer stigmergy, vision pipelines, fault recovery, and simulation execution.
## Context Links
- [Agent index](https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/llms.txt)
- [Human interactive wiki](https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601)
- [GitHub repository](https://github.com/ivaziz/swarm_car_search)
## Repository Metadata
- Repository: ivaziz/swarm_car_search
- Generated: 2026-10-19T09:57:29.544Z
- Updated: 2026-10-19T12:26:44.858Z
- Runtime: Antigravity CLI
- Format: First 30 Minutes
- Pages: 6
## Page Index
- 01. [Start Here: System Mental Model & 30-Minute Reading Order](https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/pages/01-start-here-system-mental-model-30-minute-reading-order.md) - Core mental model for the decentralized multi-robot car search system, team subsystem boundaries between Swarm and SLAM, essential glossary terminology, and the recommended code-reading order.
- 02. [First Execution, Scenarios & Runtime Boundaries](https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/pages/02-first-execution-scenarios-runtime-boundaries.md) - Hands-on guide to launching deterministic simulation scenarios (A through F), inspecting generated visual artifacts (ASCII HUD, SVG snapshots, Matplotlib panels), and understanding environment setup constraints and dependencies.
- 03. [ACO Multi-Objective Scoring & Tri-Layer Stigmergy](https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/pages/03-aco-multi-objective-scoring-tri-layer-stigmergy.md) - Detailed mechanics of the decentralized Ant Colony Optimization engine, the 8-factor score formulation, Boltzmann Softmax action selection, tri-layer pheromone evaporation dynamics, and recursive Bayesian target probability updates.
- 04. [Perception Pipeline & Subsystem Contracts](https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/pages/04-perception-pipeline-subsystem-contracts.md) - The 5-stage target identification pipeline covering raycast FOV occlusion, Levenshtein fuzzy license plate OCR, appearance verification, temporal tracking, and the abstract ISLAMProvider interface contract decoupling Swarm from SLAM.
- 05. [Autonomous Fault Detection & Dynamic Reallocation](https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/pages/05-autonomous-fault-detection-dynamic-reallocation.md) - Telemetry monitoring for silent heartbeat timeouts, battery exhaustion, and kinematic stalls, coupled with dynamic task revocation, avoidance pheromone deposition over disabled units, and immediate peer reallocation.
- 06. [Beyond the First 30 Minutes: 3D Gazebo & Extension Points](https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/pages/06-beyond-the-first-30-minutes-3d-gazebo-extension-points.md) - Synthesis of initial learnings, bridging from 2D discrete simulation to containerized ROS 2 / Gazebo Fortress 3D worlds, running Monte Carlo benchmark evaluations against baselines, and architectural extension hooks.
## Source File Index
- `configs/aco_params.yaml`
- `configs/default_config.yaml`
- `docker/Dockerfile`
- `docs/GLOSSARY.md`
- `docs/SETUP_AND_EXECUTION.md`
- `docs/thesis_chapter_swarm.md`
- `evaluation/ai_benchmarks.py`
- `interfaces/perception_interface.py`
- `interfaces/slam_interface.py`
- `perception/matching_engine.py`
- `perception/plate_recognizer.py`
- `perception/vehicle_detector.py`
- `README.md`
- `requirements.txt`
- `run_gazebo_gui.sh`
- `scripts/main.py`
- `scripts/run_evaluation.py`
- `scripts/run_simulation.py`
- `simulation/scenarios.py`
- `swarm/aco/aco_config.py`
- `swarm/aco/aco_engine.py`
- `swarm/aco/task_scorer.py`
- `swarm/communication/local_comm.py`
- `swarm/core/robot_agent.py`
- `swarm/core/swarm_coordinator.py`
- `swarm/failure_handling/failure_detector.py`
- `swarm/failure_handling/recovery_manager.py`
- `swarm/pheromone/pheromone_map.py`
- `swarm/pheromone/target_probability_map.py`
- `swarm/task_allocation/task_manager.py`
- `tests/test_failure_handling.py`
- `visualization/swarm_visualizer.py`
---
## 01. Start Here: System Mental Model & 30-Minute Reading Order
> Core mental model for the decentralized multi-robot car search system, team subsystem boundaries between Swarm and SLAM, essential glossary terminology, and the recommended code-reading order.
- Page Markdown: https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/pages/01-start-here-system-mental-model-30-minute-reading-order.md
- Generated: 2026-10-19T09:57:29.538Z
### Source Files
- `README.md`
- `scripts/main.py`
- `docs/GLOSSARY.md`
- `configs/default_config.yaml`
- `swarm/core/swarm_coordinator.py`
Relevant source files
The following files were used as context for generating this wiki page:
- [README.md](README.md)
- [scripts/main.py](scripts/main.py)
- [docs/GLOSSARY.md](docs/GLOSSARY.md)
- [configs/default_config.yaml](configs/default_config.yaml)
- [configs/aco_params.yaml](configs/aco_params.yaml)
- [interfaces/slam_interface.py](interfaces/slam_interface.py)
- [docs/slam_integration_guide.md](docs/slam_integration_guide.md)
- [docs/architecture.md](docs/architecture.md)
- [swarm/core/swarm_coordinator.py](swarm/core/swarm_coordinator.py)
- [swarm/core/robot_agent.py](swarm/core/robot_agent.py)
- [swarm/states/robot_states.py](swarm/states/robot_states.py)
- [swarm/aco/task_scorer.py](swarm/aco/task_scorer.py)
- [swarm/aco/aco_engine.py](swarm/aco/aco_engine.py)
- [swarm/task_allocation/allocation_strategy.py](swarm/task_allocation/allocation_strategy.py)
- [swarm/failure_handling/recovery_manager.py](swarm/failure_handling/recovery_manager.py)
- [perception/matching_engine.py](perception/matching_engine.py)
# Start Here: System Mental Model & 30-Minute Reading Order
The **SwarmX** codebase (`ivaziz/swarm_car_search`) implements an autonomous multi-robot coordination architecture designed for locating missing and stolen vehicles in complex outdoor urban environments. Built as a robotics graduation project, the system unites bio-inspired Ant Colony Optimization (ACO), indirect spatial memory (stigmergy), Bayesian target belief modeling, and multi-stage computer vision to coordinate a decentralized fleet of ground robots under noisy real-world communication and hardware constraints.
This guide provides the foundational mental model for navigating the repository. It details the strict engineering boundary separating the Swarm and SLAM subsystems, standardizes core robotics and swarm intelligence terminology, and charts an efficient 30-minute code reading path for onboarding engineers, reviewers, and evaluators.
---
## 1. System Mental Model
The core architecture operates as a **decentralized feedback loop**: mobile robots explore an unknown urban street network, ingest sensor data, infer geometric frontiers, share environmental cues stigmergically, and probabilistically allocate search tasks without requiring a single point of failure.
```mermaid
graph TB
subgraph SensingAndSLAM["1. Sensing & State Estimation Tier (ISLAMProvider)"]
Sensors["Sensors: 2D LiDAR / IMU / Encoders / Camera"] --> SLAMImpl["SLAM Provider (MockSLAMProvider / Gazebo Bridge)"]
SLAMImpl --> SLAMSnapshot["SLAMState\n- Pose2D (x, y, theta)\n- OccupancyGrid2D (free / occupied / -1 unknown)\n- FrontierCluster List"]
end
subgraph SwarmCore["2. Swarm Coordination & Stigmergy Tier"]
SLAMSnapshot --> SwarmCoord["SwarmCoordinator\n(Orchestration Engine)"]
SwarmCoord --> StigmergyMap["Tri-Layer Pheromone Map\n- Exploration Field (tau_e, rho=0.01)\n- Recruitment Field (tau_s, rho=0.05)\n- Avoidance Field (tau_a, rho=0.30)"]
SwarmCoord --> TaskPool["TaskManager\n(Frontier Ingestion & Deduplication)"]
SwarmCoord --> BayesMap["TargetProbabilityMap\n(Bayesian Prior & Spatial Memory)"]
end
subgraph DecisionAndAlloc["3. Bio-Inspired Task Allocation Tier"]
TaskPool --> Scorer["TaskScorer\nMulti-Objective Objective Function\n(Info Gain, Target Prior, Distance, Congestion)"]
StigmergyMap --> Scorer
BayesMap --> Scorer
Scorer --> Boltzmann["ACOEngine\nScale-Invariant Boltzmann Softmax Allocator\nP(task_i | robot_j)"]
Boltzmann --> Dispatch["Task Assignment Dispatch"]
end
subgraph RobotAgentExec["4. Onboard Robot Agent & Kinematics Tier"]
Dispatch --> Agent["RobotAgent\n- RobotStateMachine (FSM: IDLE, NAVIGATING, ...)\n- Differential Drive Kinematics (v, omega)\n- Collision Avoidance Halting"]
end
subgraph PerceptionTier["5. Perception & Verification Tier"]
Agent --> RaycastFOV["SimulatedVehicleDetector\n(Raycast Line-of-Sight & 90 deg FoV)"]
RaycastFOV --> OCRMatcher["MatchingEngine\n- Fuzzy OCR (Levenshtein Distance)\n- Vehicle Appearance Verifier\n- Multi-Frame Temporal Voting (k >= 3)"]
OCRMatcher -->|MatchEvent: Confirmed| TargetAlert["Mission Event: TARGET_FOUND\nDeposit tau_s Recruitment Pheromones"]
TargetAlert --> SwarmCoord
end
subgraph FaultToleranceTier["6. Dynamic Fault Tolerance Tier"]
Agent --> HealthMon["FailureDetector\n(Heartbeat Timeouts, Low Battery, Motion Stall)"]
HealthMon --> RecMgr["RecoveryManager\n- Release Orphaned Tasks\n- Deposit Avoidance at Crash Site\n- Reassign Tasks to Peers"]
RecMgr --> SwarmCoord
end
```
### The Six Foundational Architectural Concepts
1. **Frontier-Based Exploration via Decoupled SLAM**:
Robots discover candidate exploration regions by finding frontiers—the borders where known free navigable space meets unmapped unknown space. The Swarm system never calculates maps directly; it receives extracted `FrontierCluster` centroids from an `ISLAMProvider`.
2. **Tri-Layer Stigmergic Pheromones ($\tau_e, \tau_s, \tau_a$)**:
Instead of flooding the wireless network with centralized path broadcasts, robots communicate indirectly via 2D spatial scalar fields:
- **Exploration Layer ($\tau_e$)**: Slow evaporation rate ($\rho_e = 0.01$) acts as long-term memory of covered street corridors.
- **Recruitment Layer ($\tau_s$)**: Medium evaporation rate ($\rho_s = 0.05$) attracts neighboring robots toward candidate target sightings.
- **Avoidance Layer ($\tau_a$)**: Fast evaporation rate ($\rho_a = 0.30$) temporarily repels robots from areas occupied by peers or disabled wreckage to prevent search overlap and traffic jams.
3. **Multi-Objective Task Scoring**:
Candidate tasks are evaluated from each robot’s unique spatial position using a scale-invariant objective function balancing positive incentives (information gain, Bayesian priors, urgency, exploration pheromone) against negative costs (Euclidean distance, peer congestion, collision risk, and redundant visits).
4. **Boltzmann Softmax Action Selection**:
To prevent all robots from greedily collapsing onto the single highest-scoring frontier, tasks are sampled probabilistically using temperature-parameterized Boltzmann Softmax selection. Logits are normalized against the maximum candidate score to guarantee numerical stability.
5. **Multi-Stage Computer Vision Verification**:
Target identification uses a 5-stage pipeline: field-of-view raycasting, plate cropping, fuzzy Levenshtein OCR matching against an optical confusion map ($8 \leftrightarrow B, 0 \leftrightarrow O$), vehicle appearance verification, and multi-frame temporal voting requiring $k \ge 3$ consistent frames before triggering a mission alarm.
6. **Dynamic Fault Recovery**:
Robots monitor peer heartbeats, motion stalling, and battery drainage. When a robot fails, its assigned tasks are immediately returned to the unassigned pool, a high avoidance pheromone ($\tau_a$) is stamped at the failure coordinate, and surviving peers reallocate the task dynamically.
Sources: [README.md:28-76](), [docs/architecture.md:5-70]()
---
## 2. Team Subsystem Boundaries: Swarm vs. SLAM
To ensure clean engineering separation across the project team, responsibilities are strictly split between the Swarm team and the SLAM team.
```
+---------------------------------------------------------------------------------------+
| SUBSYSTEM RESPONSIBILITY MATRIX |
+---------------------------------------------------------------------------------------+
| SLAM Subsystem (Saqr & Mahmoud) | Swarm Subsystem (Mohammed & Ibrahim) |
+-----------------------------------------------+---------------------------------------+
| - Sensor Driver Integration (2D LiDAR, IMU) | - ACO Task Scorer & Boltzmann Engine |
| - EKF / Odometry Fusion (Wheel Encoders) | - Tri-Layer Stigmergic Pheromone Map |
| - 2D Occupancy Grid Generation (data: -1/0/100)| - Task Lifecycle & Deduplication |
| - Obstacle Inflation & Raycasting | - Dynamic Failure Detection & Recovery|
| - Frontier Extraction & Clustering | - Robot State Machine (FSM) & Kinematics|
| - ISLAMProvider Concrete Implementation | - 5-Stage Vision & Plate OCR Matching|
+---------------------------------------------------------------------------------------+
```
### The Architectural Invariant
> **Architectural Invariant**: The Swarm coordination subsystem **never** implements SLAM internally, never fuses raw wheel encoder or IMU data, and never accesses raw sensor buffers directly. All spatial understanding flows through the standardized [`ISLAMProvider`](interfaces/slam_interface.py) abstract interface.
### The Contract: `interfaces/slam_interface.py`
The SLAM subsystem communicates with the Swarm coordinator by providing a `SLAMState` snapshot containing standardized dataclasses:
```python
# interfaces/slam_interface.py:12-100
@dataclass(frozen=True)
class Pose2D:
x: float
y: float
theta: float # Heading in radians [-pi, pi]
@dataclass
class OccupancyGrid2D:
width: int
height: int
resolution: float # Meters per cell edge
origin: Pose2D
data: List[int] # -1: Unknown, 0: Free space, 100: Occupied obstacle
@dataclass(frozen=True)
class FrontierCluster:
cluster_id: str
centroid_x: float
centroid_y: float
size: int
bounding_box: Tuple[float, float, float, float]
@dataclass
class SLAMState:
robot_id: str
timestamp: float
pose: Pose2D
linear_velocity: float
angular_velocity: float
occupancy_grid: OccupancyGrid2D
frontiers: List[FrontierCluster]
exploration_ratio: float
navigation_cost_map: Optional[List[float]] = None
class ISLAMProvider(ABC):
@abstractmethod
def get_latest_state(self, robot_id: str) -> SLAMState: ...
@abstractmethod
def is_healthy(self, robot_id: str) -> bool: ...
```
In headless Python tests and benchmarks, this contract is fulfilled by `MockSLAMProvider`, which simulates raycasting and frontier clustering over an `UrbanEnvironment`. In 3D Gazebo Fortress simulations, it is fulfilled by `GazeboSLAMBridge` subscribing to ROS 2 topics.
Sources: [interfaces/slam_interface.py:1-100](), [docs/slam_integration_guide.md:10-33](), [docs/architecture.md:15-28]()
---
## 3. Essential Glossary & Key Terminology
The table below summarizes the core terminology used across code modules, configuration files, and thesis documentation.
| Term | Domain | Code Reference | Definition & System Role |
| :--- | :--- | :--- | :--- |
| **Ant Colony Optimization (ACO)** | Swarm Intelligence | `swarm/aco/aco_engine.py` | Bio-inspired metaheuristic where artificial ants (robots) sample candidate search frontiers based on heuristic attractiveness and pheromone trails. |
| **Stigmergy** | Swarm Intelligence | `swarm/pheromone/` | Indirect coordination via continuous 2D scalar fields ($\tau_e, \tau_s, \tau_a$), eliminating the need for centralized trajectory negotiation. |
| **Multi-Robot Task Allocation (MRTA)** | Robotics / AI | `swarm/task_allocation/` | Computational problem of assigning frontier clusters and target verification goals to distributed robots to maximize global coverage. |
| **Boltzmann / Softmax Selection** | Metaheuristics | `swarm/aco/aco_engine.py` | Temperature-scaled probability distribution $P(t_i) \propto \exp(Q_i / T)$ preventing greedy clustering onto identical search sectors. |
| **Frontier-Based Exploration** | SLAM / Mapping | `interfaces/slam_interface.py` | Exploration strategy navigating toward frontier boundaries between known free space (`0`) and unknown territory (`-1`). |
| **2D Occupancy Grid** | Mapping | `interfaces/slam_interface.py` | Discretized spatial grid where cells indicate unknown (`-1`), free (`0`), or obstacle (`100`) status. |
| **Differential Drive Kinematics** | Motion Control | `simulation/sim_robot.py` | Non-holonomic two-wheel motion model constrained by linear forward velocity $v$ and yaw angular velocity $\omega$. |
| **Pose2D** | Geometry / SLAM | `interfaces/slam_interface.py` | 3-DOF coordinate tuple $(x, y, \theta)$ expressing position and orientation in the global frame. |
| **Levenshtein Distance** | Computer Vision | `perception/matching_engine.py` | Minimum single-character edit operations between OCR string and target watchlist plate, normalized to $[0.0, 1.0]$. |
| **Multi-Frame Temporal Consensus** | Computer Vision | `perception/temporal_tracker.py` | Aggregates successive sightings of stationary vehicles, requiring $k \ge 3$ frames and character voting to eliminate transient OCR glitches. |
| **Dynamic Urgency Escalation** | Swarm Intelligence | `swarm/aco/task_scorer.py` | Multiplier ($2.2\times$ to $3.5\times$) boosting task priority when candidate visual cues indicate a likely stolen vehicle match. |
| **ROS 2 & Gazebo Fortress** | Infrastructure | `ros2_ws/`, `run_gazebo_gui.sh` | Middleware (ROS 2 Humble) and 3D rigid-body dynamic simulator providing realistic sensor streams and physics. |
Sources: [docs/GLOSSARY.md:7-92](), [README.md:55-76]()
---
## 4. The 30-Minute Code Reading Order
To understand how high-level swarm decisions translate into physical robot motion, follow this sequence. It starts with system interfaces, traces the top-level coordination tick, inspects the ACO allocation mathematics, examines the onboard agent state machine, and concludes with computer vision and failure handling.
```
+-------------------------------------------------------------------------------------------+
| 30-MINUTE CODE READING ROADMAP |
+-------------------------------------------------------------------------------------------+
| Stage | Target Files | Focus Area | Time |
+-------+--------------------------------------------------+-------------------------+------+
| 1 | README.md, scripts/main.py, slam_interface.py | Architecture & Contract | 5m |
| 2 | swarm/core/swarm_coordinator.py | Central Coordination | 7m |
| 3 | swarm/aco/task_scorer.py & aco_engine.py | Multi-Objective Scoring | 6m |
| 4 | swarm/core/robot_agent.py & robot_states.py | FSM & Robot Execution | 6m |
| 5 | perception/matching_engine.py & recovery_mgr.py | Vision & Fault Recovery | 6m |
+-------------------------------------------------------------------------------------------+
```
### Stage 1 (0–5 min): System Entry Point & Global Contracts
* **Files**: [`README.md`](README.md), [`scripts/main.py`](scripts/main.py:1-127), [`interfaces/slam_interface.py`](interfaces/slam_interface.py:1-100)
* **What to inspect**:
1. `scripts/main.py`: The single CLI launcher supporting `sim`, `eval`, `test`, and `info`.
2. `interfaces/slam_interface.py`: The dataclasses `Pose2D`, `OccupancyGrid2D`, `FrontierCluster`, `SLAMState`, and the abstract class `ISLAMProvider`.
* **Key takeaway**: Any subsystem component can be mocked or swapped as long as it adheres to `ISLAMProvider`.
### Stage 2 (5–12 min): The Swarm Coordination Loop
* **File**: [`swarm/core/swarm_coordinator.py`](swarm/core/swarm_coordinator.py:174-467)
* **What to inspect**:
1. Inspect `SwarmCoordinator.step_coordination(timestamp, dt)`:
- Step 1: Advance physics, SLAM, and spatial search memory.
- Step 2: Ingest peer heartbeats and trigger failure detection.
- Step 3: Ingest frontier clusters from active robots' SLAM states into `TaskManager`.
- Step 4: Allocate pending tasks to idle or preemptible robots via `ACOAllocationStrategy`.
- Step 5: Evaluate multi-stage perception and check for target confirmation.
- Step 6: Deposit exploration and avoidance pheromones, followed by discrete evaporation.
* **Key takeaway**: The coordinator synchronizes the swarm tick while delegating local decisions to individual agents.
### Stage 3 (12–18 min): Bio-Inspired Task Allocation & Stigmergy
* **Files**: [`swarm/aco/task_scorer.py`](swarm/aco/task_scorer.py:37-148), [`swarm/aco/aco_engine.py`](swarm/aco/aco_engine.py:103-176), [`configs/aco_params.yaml`](configs/aco_params.yaml:1-36)
* **What to inspect**:
1. `task_scorer.py`: Inspect `explain_score(...)`. Notice the 8 normalized components:
$$\text{Score} = \text{NetUtility} \cdot \left(\frac{\tau^{\alpha \cdot 0.5} \cdot G^{\delta}}{C^{\varepsilon \cdot 0.5}}\right) \cdot \text{BatteryFactor}$$
where $\tau$ is recruitment pheromone, $G$ is urgency, and $C$ is distance cost.
2. `aco_engine.py`: Inspect `select_task(...)`. Look at how relative logits prevent numeric overflow during Boltzmann sampling:
```python
# swarm/aco/aco_engine.py:157-159
weights = [math.exp(((s / max_score) - 1.0) / temp) for s in scores]
```
* **Key takeaway**: The scoring function blends empirical robotics costs (distance, battery) with natural ant foraging dynamics.
### Stage 4 (18–24 min): Robot Agent, FSM, and Kinematics
* **Files**: [`swarm/core/robot_agent.py`](swarm/core/robot_agent.py:116-150), [`swarm/states/robot_states.py`](swarm/states/robot_states.py:9-60), [`swarm/states/state_machine.py`](swarm/states/state_machine.py)
* **What to inspect**:
1. `robot_states.py`: Review `RobotState` values (`IDLE`, `NAVIGATING`, `SEARCHING`, `VEHICLE_DETECTED`, `VERIFYING`, `REPORTING`, `FAILED`).
2. `robot_agent.py`: Trace `assign_task(...)` and notice how incoming `TARGET_CONFIRMATION` tasks preempt generic exploration tasks.
* **Key takeaway**: Each physical robot maintains an autonomous onboard finite state machine, allowing it to navigate and sense independently even during communication blackouts.
```mermaid
stateDiagram-v2
[*] --> IDLE
IDLE --> NAVIGATING: Task Assigned
NAVIGATING --> SEARCHING: Arrived at Centroid
SEARCHING --> NAVIGATING: Reassigned New Frontier
SEARCHING --> VEHICLE_DETECTED: Camera FoV Trigger
NAVIGATING --> VEHICLE_DETECTED: Sighting in Transit
VEHICLE_DETECTED --> VERIFYING: Run OCR & Appearance Matcher
VERIFYING --> REPORTING: Watchlist Match Confirmed
VERIFYING --> SEARCHING: Non-Target Decoy
REPORTING --> COMPLETED: Mission Finalized
NAVIGATING --> FAILED: Stall / Heartbeat / Battery Depleted
SEARCHING --> FAILED: Critical Failure
FAILED --> [*]
```
### Stage 5 (24–30 min): Perception Pipeline & Dynamic Fault Recovery
* **Files**: [`perception/matching_engine.py`](perception/matching_engine.py:95-215), [`swarm/failure_handling/recovery_manager.py`](swarm/failure_handling/recovery_manager.py:63-120)
* **What to inspect**:
1. `matching_engine.py`: Look at `evaluate_multi_stage_match(...)`, which fuses vehicle classification ($C_{\text{veh}}$), plate localization ($C_{\text{plate}}$), Levenshtein OCR ($C_{\text{ocr}} \times S_{\text{plate}}$), RGB appearance ($C_{\text{app}}$), and temporal consensus ($C_{\text{temp}}$).
2. `recovery_manager.py`: Trace `handle_failure(...)`. See how orphaned tasks are marked `PENDING`, high avoidance pheromone ($\tau_a = 12.0$) is deposited over the disabled vehicle, and the task is immediately evaluated for dynamic reassignment to healthy peers.
* **Key takeaway**: The swarm is resilient to physical attrition and sensor noise, recovering mission progress in $< 0.2\,\text{s}$ without operator intervention.
Sources: [swarm/core/swarm_coordinator.py:174-467](), [swarm/aco/task_scorer.py:37-148](), [swarm/aco/aco_engine.py:103-176](), [swarm/core/robot_agent.py:116-150](), [swarm/states/robot_states.py:9-60](), [swarm/failure_handling/recovery_manager.py:63-120]()
---
## 5. System Configuration & Runtime Execution
The system supports two execution paradigms:
1. **Zero-Dependency Headless Python**: Fast Monte Carlo simulation and unit tests using Python's standard library.
2. **Containerized ROS 2 & Gazebo Fortress**: High-fidelity 3D physics with LiDAR and camera rendering accessible via noVNC web browser.
### Key Configuration Parameters
```yaml
# configs/default_config.yaml:11-32
swarm:
num_robots: 3
heartbeat_interval_sec: 1.0
failure_timeout_sec: 5.0
reassignment_cooldown_sec: 2.0
navigation:
max_linear_velocity: 1.5 # m/s
max_angular_velocity: 1.0 # rad/s
waypoint_tolerance: 0.5 # meters
perception:
detection_confidence_threshold: 0.70
ocr_confidence_threshold: 0.80
evidence_capture_enabled: true
```
```yaml
# configs/aco_params.yaml:5-36
objective_function_weights:
alpha: 1.2 # Positive recruitment pheromone
beta: 1.5 # Information gain / frontier size
gamma: 1.8 # Target vehicle prior probability
delta: 1.0 # Mission urgency multiplier
epsilon: 1.3 # Travel distance penalty
zeta: 2.0 # Robot congestion penalty
theta: 1.5 # Avoidance / recent visit penalty
pheromone_dynamics:
evaporation_rates:
exploration: 0.01 # Long-term memory
success: 0.05 # Recruitment decay
avoidance: 0.30 # Rapid congestion release
stochastic_selection:
temperature: 1.0
epsilon_greedy: 0.05
```
### Common CLI Invocations
All repository capabilities are launched through [`scripts/main.py`](scripts/main.py):
```bash
# Display system banner and architecture overview
python3 scripts/main.py info
# Run all unit and integration tests (96+ passing)
python3 scripts/main.py test
# Launch autonomous simulation with 4 robots for 100 steps
python3 scripts/main.py sim --robots 4 --steps 100 --output-dir output/simulation_run
# Execute comparative Monte Carlo benchmark against Greedy & Random Walk baselines
python3 scripts/main.py eval --trials 5 --steps 60 --robots 4 --output-dir output/benchmark_run
```
Sources: [configs/default_config.yaml:5-32](), [configs/aco_params.yaml:5-36](), [scripts/main.py:85-124]()
---
## 6. Summary
The `ivaziz/swarm_car_search` architecture decouples high-level bio-inspired swarm intelligence from low-level robotics mapping. By maintaining a clean boundary through `ISLAMProvider`, Mohammed & Ibrahim's decentralized ACO task allocation and stigmergic pheromone maps operate seamlessly alongside Saqr & Mahmoud's SLAM state estimators. Following the 30-minute reading order will give any developer or evaluator an end-to-end understanding of how frontier ingestion, multi-objective scoring, Boltzmann selection, and dynamic fault recovery collaborate to find targets in challenging urban environments.
---
## 02. First Execution, Scenarios & Runtime Boundaries
> Hands-on guide to launching deterministic simulation scenarios (A through F), inspecting generated visual artifacts (ASCII HUD, SVG snapshots, Matplotlib panels), and understanding environment setup constraints and dependencies.
- Page Markdown: https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/pages/02-first-execution-scenarios-runtime-boundaries.md
- Generated: 2026-10-19T09:25:36.528Z
### Source Files
- `scripts/run_simulation.py`
- `simulation/scenarios.py`
- `requirements.txt`
- `docs/SETUP_AND_EXECUTION.md`
- `visualization/swarm_visualizer.py`
Relevant source files
The following files were used as context for generating this wiki page:
- [scripts/run_simulation.py](scripts/run_simulation.py)
- [simulation/scenarios.py](simulation/scenarios.py)
- [requirements.txt](requirements.txt)
- [docs/SETUP_AND_EXECUTION.md](docs/SETUP_AND_EXECUTION.md)
- [visualization/swarm_visualizer.py](visualization/swarm_visualizer.py)
- [scripts/main.py](scripts/main.py)
- [evaluation/ai_benchmarks.py](evaluation/ai_benchmarks.py)
# First Execution, Scenarios & Runtime Boundaries
This guide provides a comprehensive walkthrough for onboarding, configuring, and executing deterministic simulation missions in **SwarmX** (`ivaziz/swarm_car_search`). The repository simulates a decentralized multi-robot fleet deployed across an urban street grid to detect, track, verify, and report a target vehicle using Ant Colony Optimization (ACO) task allocation, multi-layer stigmergic pheromone maps, and a multi-stage optical perception pipeline.
Whether executing on a minimal standard Python runtime without third-party packages or within a full containerized ROS 2 and Gazebo Fortress environment, this document outlines the exact boundary constraints, CLI flags, deterministic benchmark scenarios (A through F, as well as extended scenarios G and H), and inspection procedures for generated visual artifacts including ASCII HUDs, pure-Python SVG vector snapshots, and publication-grade Matplotlib diagnostic dashboards.
---
## 1. Runtime Boundaries & System Architecture
SwarmX is designed around a **zero-dependency core** philosophy: standard system Python is sufficient to execute the multi-agent coordination logic, simulated kinematics, sensor raycasting, perception verification, and vector visualization. High-fidelity graphical rendering and containerized physical physics form optional outer layers.
```mermaid
flowchart TD
subgraph Core ["Zero-Dependency Pure-Python Core (Python >= 3.9)"]
CLI["CLI Entry Points\nscripts/main.py\nscripts/run_simulation.py"]
Coord["Swarm Coordinator\nswarm/core/swarm_coordinator.py"]
ACO["ACO Task Allocation & Stigmergy\nswarm/aco/ & swarm/pheromone/"]
FSM["Robot Agents & Kinematics\nswarm/core/robot_agent.py\nswarm/states/"]
Perception["Perception & OCR Pipeline\nperception/vehicle_detector.py\nperception/matching_engine.py"]
VizPure["Zero-Dependency Visualizers\nrender_ascii_map (Terminal HUD)\nrender_svg_map (Pure-Python SVG)"]
end
subgraph OptionalEnhancements ["Optional Enhancement Layer (requirements.txt)"]
Matplotlib["Matplotlib >= 3.7.0 & NumPy >= 1.24.0\nrender_matplotlib_dashboard\n3-Panel Mission Analytics"]
Pillow["Pillow >= 10.0.0\nswarm_mission_animation.gif\nMulti-Frame Animated GIF"]
PyTest["PyTest >= 7.0.0\nAutomated Verification Suite"]
end
subgraph ContainerizedLayer ["Containerized 3D Physical Simulation (Docker)"]
ROS2["ROS 2 Humble Node Orchestrator\nros2_ws/src/swarm_gazebo"]
Gazebo["Ignition Gazebo Fortress\nurban_search_world.world"]
Bridge["ros_gz_bridge\n/clock, /cmd_vel, /odom, /scan, /camera"]
VNC["Web GUI (Port 6080: noVNC)\nNative Screen Sharing (Port 5900: VNC)"]
end
CLI --> Coord
Coord --> ACO
Coord --> FSM
FSM --> Perception
Coord --> VizPure
Coord -.-> Matplotlib
Matplotlib -.-> Pillow
CLI -.-> ROS2
ROS2 <--> Bridge <--> Gazebo
Gazebo --> VNC
```
### Dependency Isolation & Capabilities Matrix
The runtime dynamically inspects library availability at import time and gracefully falls back to core features without crashing.
| Runtime Tier | External Dependencies | Supported Features | Primary Entry Point |
| :--- | :--- | :--- | :--- |
| **Core Minimal** | *None* (Python standard library only) | 2D kinematic simulation, ACO allocation, tri-layer pheromone maps, OCR fuzzy matching, terminal ASCII HUD, pure-Python SVG export. | `python3 scripts/run_simulation.py` |
| **Visualization Tier** | `matplotlib>=3.7.0`, `numpy>=1.24.0`, `pillow>=10.0.0` | High-resolution 3-panel dashboard (`.png`), telemetry timeline curves, belief heatmaps, animated GIF generation. | `python3 scripts/run_simulation.py` |
| **Testing & Benchmark** | `pytest>=7.0.0`, `pytest-cov>=4.0.0` | Execution of all unit and integration tests across SLAM, ACO, perception, and visualizer modules. | `pytest -v` or `python3 scripts/main.py test` |
| **Containerized 3D** | Docker Desktop / Docker Engine (Ubuntu 22.04 LTS, ROS 2 Humble, Gazebo Fortress) | Full rigid-body physics, 3D LiDAR pointclouds, synthetic camera sensors, noVNC web streaming (port 6080), native VNC (port 5900). | `./run_gazebo_gui.sh` or `./run_simulation.sh` |
Sources: [requirements.txt:1-28](), [docs/SETUP_AND_EXECUTION.md:8-35](), [visualization/swarm_visualizer.py:22-44]()
---
## 2. Quickstart & Simulation Execution
Missions can be launched either via the unified command launcher (`scripts/main.py`) or directly through the specialized simulation runner (`scripts/run_simulation.py`).
### Unified CLI Entry Point
The master CLI dispatches subcommands for simulation, benchmarking, testing, and architecture inspection:
```bash
# Display system architecture, developer specifications, and module inventory
python3 scripts/main.py info
# Execute the automated verification test suite
python3 scripts/main.py test
# Launch an autonomous swarm simulation run
python3 scripts/main.py sim --scenario easy_target --robots 4 --steps 120
```
Sources: [scripts/main.py:20-65](), [scripts/main.py:91-124]()
### Direct Simulation Runner Options
`scripts/run_simulation.py` accepts parameters controlling the physical grid, fleet composition, target characteristics, fault injection, and output frequency:
```bash
python3 scripts/run_simulation.py \
--scenario easy_target \
--robots 4 \
--steps 100 \
--output-dir output/simulation_run \
--render-every 10 \
--seed 42
```
#### CLI Configuration Parameter Reference
| Flag | Type | Default | Description |
| :--- | :---: | :---: | :--- |
| `--scenario` | `str` | `"easy_target"` | Named deterministic scenario (`easy_target`, `hidden_target`, `target_recruitment`, `failure_recovery`, `false_positive`, `multiple_vehicles`, or `custom`). |
| `--width` | `float` | `50.0` | Urban grid width in meters. |
| `--height` | `float` | `50.0` | Urban grid height in meters. |
| `--resolution` | `float` | `1.0` | Grid occupancy resolution in meters per cell. |
| `--robots` | `int` | `4` | Number of autonomous search robots deployed in the fleet. |
| `--steps` | `int` | `100` | Maximum discrete simulation execution ticks. |
| `--dt` | `float` | `0.2` | Kinematic and physics integration timestep in seconds ($0.2\,\text{s} = 5\,\text{Hz}$). |
| `--target-x` | `float` | `25.0` | World X coordinate for custom target placement. |
| `--target-y` | `float` | `25.0` | World Y coordinate for custom target placement. |
| `--target-plate` | `str` | `"DXB-88392"` | Expected license plate string for target vehicle identification. |
| `--inject-failure-tick`| `int` | `None` | Simulation tick index at which catastrophic failure is triggered on a robot. |
| `--fail-robot` | `str` | `"robot_1"` | Identifier of the specific robot to disable during fault injection. |
| `--output-dir` | `str` | `"output/simulation_run"` | Target filesystem directory for saving SVG frames, PNG dashboards, and reports. |
| `--render-every` | `int` | `10` | Frequency in ticks for writing visual frames to disk ($10 \times 0.2\,\text{s} = 2.0\,\text{s}$). |
| `--seed` | `int` | `42` | Random number generator seed for deterministic repeatability. |
Sources: [scripts/run_simulation.py:47-67]()
---
## 3. Deterministic Scenarios (A through F & Extensions)
To evaluate multi-robot search behavior under controlled, reproducible conditions, SwarmX provides pre-configured benchmark scenarios implemented in `simulation/scenarios.py`.
### The Canonical Urban Grid & Fleet Initial Conditions
Unless overridden, all scenarios instantiate a canonical $50\,\text{m} \times 50\,\text{m}$ urban environment featuring perimeter boundary walls and four solid building blocks dividing the world into cardinal avenues and a central plaza:
- **South-West Block**: $[10.0, 22.5] \times [10.0, 22.5]\,\text{m}$
- **South-East Block**: $[27.5, 40.0] \times [10.0, 22.5]\,\text{m}$
- **North-West Block**: $[10.0, 22.5] \times [27.5, 40.0]\,\text{m}$
- **North-East Block**: $[27.5, 40.0] \times [27.5, 40.0]\,\text{m}$
The four robots spawn at obstacle-free corner intersections oriented toward the interior corridors:
1. `robot_1`: $(6.0, 6.0, 0.0\,\text{rad})$ — South-West corner facing East.
2. `robot_2`: $(6.0, 44.0, -\frac{\pi}{2}\,\text{rad})$ — North-West corner facing South.
3. `robot_3`: $(44.0, 6.0, \frac{\pi}{2}\,\text{rad})$ — South-East corner facing North.
4. `robot_4`: $(44.0, 44.0, \pi\,\text{rad})$ — North-East corner facing West.
Sources: [simulation/scenarios.py:37-54](), [simulation/scenarios.py:71-77]()
```
(0, 50) +---------------------------------------+ (50, 50)
| R2 [v] Avenue N [<] R4|
| +-------+ +-------+ |
| | NW | | NE | |
| | Block | | Block | |
| +-------+ +-------+ |
| Ave W Central Plaza Ave E |
| (25, 25) |
| +-------+ +-------+ |
| | SW | | SE | |
| | Block | | Block | |
| +-------+ +-------+ |
| R1 [>] Avenue S [^] R3|
(0,0) +---------------------------------------+ (50, 0)
```
### Comparative Scenario Matrix
| Scenario Name | Canonical ID | Target Location | Vehicle Distribution | Max Steps / Seed | Primary Evaluation Objective |
| :--- | :---: | :---: | :--- | :---: | :--- |
| `easy_target` | **Scenario A** | $(25.0, 25.0)$ Central Plaza | 1 target sedan (`DXB-88392`) | 500 / 100 | Baseline convergence, unoccluded visual detection, fast discovery. |
| `hidden_target` | **Scenario B** | $(45.0, 35.0)$ East Corridor | 1 target sedan (`DXB-88392`) | 600 / 105 | Raycast occlusion by buildings, peripheral navigation. |
| `multiple_decoys` / `false_positive` | **Scenario C** | $(25.0, 25.0)$ Central Hub | 1 target sedan + 3 civilian decoys (`AD-10294`, `SHJ-44910`, `DXB-99112`) | 500 / 110 | Rejection of non-target vehicles via fuzzy OCR plate matching. |
| `noisy_ocr` | **Scenario D** | $(25.0, 35.0)$ North Lane | 1 target sedan (`DXB-88392`) | 500 / 115 | Optical noise, character ambiguity, multi-frame temporal voting. |
| `failure_recovery` | **Scenario E** | $(25.0, 25.0)$ Central Plaza | 1 target sedan (`DXB-88392`) | 500 / 120 | Robot 1 fails at $t=4.0\,\text{s}$ (tick 20); task reclamation & ACO re-allocation. |
| `cooperative_encounter` / `target_recruitment` | **Scenario F** | $(25.0, 35.0)$ North Avenue | 1 target sedan (`DXB-88392`) | 600 / 125 | Multi-robot convergence, stigmergic recruitment pheromone ($\tau_s$). |
| `far_target` *(Extended)* | **Scenario G** | $(35.0, 25.0)$ East Sector | 1 target sedan (`DXB-88392`) | 1500 / 130 | Distant travel; all 4 robots spawn clustered in South-West corner. |
| `low_prior_target` *(Extended)* | **Scenario H** | $(6.0, 25.0)$ West Lane | 1 target sedan (`DXB-88392`) | 1000 / 135 | Deceptive high prior ($0.90$) at center; target hidden in low prior ($0.20$) zone. |
Sources: [simulation/scenarios.py:56-282]()
### Detailed Scenario Specifications
#### Scenario A (`easy_target`): Central Plaza Baseline
Target vehicle `DXB-88392` (dark gray sedan) is parked in the center intersection at $(25.0, 25.0)$, inside a designated `central_plaza` zone $([20, 20] \times [30, 30]\,\text{m})$ with an initial prior probability of $0.85$. Robots moving inward along avenues achieve rapid line-of-sight.
Sources: [simulation/scenarios.py:79-98]()
```python
# simulation/scenarios.py:81-98
target = TargetVehicle(
target_id="target_car_alpha",
plate_number=target_plate,
vehicle_class="sedan",
color="dark_gray",
position=(25.0, 25.0),
is_stolen=True,
)
return ScenarioConfig(
name=name,
description="Scenario A: Target easy to detect in central intersection.",
target_plate=target_plate,
vehicles=[target],
start_poses=default_starts,
max_steps=500,
seed=100,
zones=[UrbanZone("central_plaza", (20.0, 20.0, 30.0, 30.0), 0.85)],
)
```
#### Scenario B (`hidden_target`): East Corridor Occlusion
The target is positioned at $(45.0, 35.0)$ tucked along the narrow eastern building corridor. Robots crossing the central avenue cannot observe it due to raycast obstruction by the North-East building block until reaching coordinates $x \ge 40.0\,\text{m}$.
Sources: [simulation/scenarios.py:100-119]()
#### Scenario C (`multiple_vehicles` / `false_positive`): Decoy Rejection
To evaluate precision against false alarms, three non-stolen civilian vehicles are placed in the environment alongside the stolen car:
- `decoy_suv_1`: White SUV, plate `AD-10294` at $(10.0, 25.0)$
- `decoy_sedan_2`: Red sedan, plate `SHJ-44910` at $(40.0, 25.0)$
- `decoy_truck_3`: Blue truck, plate `DXB-99112` at $(25.0, 10.0)$
The perception pipeline must classify and dismiss these vehicles as negative candidates without halting the global search.
Sources: [simulation/scenarios.py:121-164]()
#### Scenario D (`noisy_ocr`): Optical Degradation
The target is placed at $(25.0, 35.0)$. The plate recognizer introduces optical noise and character substitutions, requiring the temporal tracker to accumulate multiple consecutive observations to reach the composite confirmation threshold ($C_{\text{target}} \ge 0.80$).
Sources: [simulation/scenarios.py:166-185]()
#### Scenario E (`failure_recovery`): Dynamic Fault Injection
Robot `robot_1` experiences a catastrophic simulated motor failure at tick 20 ($t = 4.0\,\text{s}$). The swarm coordinator marks `robot_1` as an obstacle, revokes its pending tasks, and reallocates them to surviving members using ACO utility bidding.
Sources: [simulation/scenarios.py:187-208]()
#### Scenario F (`cooperative_encounter` / `target_recruitment`): Multi-Robot Convergence
Target placed at $(25.0, 35.0)$ on the North avenue. Upon initial candidate sighting by the first robot, recruitment pheromone ($\tau_s$) is deposited on the shared stigmergy field, prompting nearby peers to alter trajectories for multi-angle visual confirmation.
Sources: [simulation/scenarios.py:210-229]()
---
## 4. Visual Artifacts & Diagnostic Telemetry
During and upon conclusion of a simulation run, `SwarmVisualizer` (`visualization/swarm_visualizer.py`) generates artifacts across three complementary formats.
### 1. Terminal ASCII HUD (`render_ascii_map`)
Printed directly to `sys.stdout` during execution and in the post-mission summary, requiring zero external libraries. It maps physical space onto a character matrix with an active status header and legend.
```text
=== Swarm Urban Search Map [t=24.0s | Active: 4 | Target: SEARCHING] ===
#.................................................#
#.R...............................................#
#.................................................#
#..........#############...#############..........#
#..........#############...#############..........#
#..........#############...#############..........#
#..........#############...#############..........#
#..........#############...#############..........#
#....................?............................#
#....................V............................#
#.......................................R.........#
#..........#############...#############..........#
#..........#############...#############..........#
#..........#############...#############..........#
#..........#############...#############..........#
#..........#############...#############..........#
#.................................................#
#........R........................................#
#............................................R....#
#.................................................#
Legend: [#] Obstacle [.] Free Space [R] Robot [X] Failed Robot [?] Frontier Task [V] Vehicle [!] Target Confirmed
```
Sources: [visualization/swarm_visualizer.py:418-480]()
### 2. Pure-Python SVG Snapshots (`render_svg_map`)
Self-contained vector files saved to `/frame_XXXX.svg` and `final_mission_summary.svg` with zero third-party dependencies:
- **Urban Canvas**: `#16161D` background with `#1F1F28` bounds and `#3C3C4D` obstacle boxes.
- **Frontier Tasks**: Rendered as dashed circles (`#FFD600` for pending, `#00E5FF` for assigned).
- **Vehicles**: Squares (`#FF3D00` for wanted candidates, `#00E676` for verified target) annotated with license plate labels.
- **Robot Fleets**: Directed circles colored by finite state machine status (`STATE_COLOR_MAP`), with orientation vectors indicating instantaneous heading. Failed robots are struck through with a red cross (`#F44336`).
Sources: [visualization/swarm_visualizer.py:46-59](), [visualization/swarm_visualizer.py:481-576]()
### 3. Publication-Grade Matplotlib Dashboard (`render_matplotlib_dashboard`)
When `matplotlib` and `numpy` are installed, SwarmX automatically configures a headless `Agg` backend (`MPLCONFIGDIR=/tmp/matplotlib_swarm_cache`) and exports high-resolution 160 DPI dashboards:
```
+----------------------------------------+---------------------------------------+
| | PANEL 2: STIGMERGY & BAYESIAN BELIEF |
| | 2D RGB Heatmap of Pheromone Fields: |
| PANEL 1: MAIN URBAN MAP | - Blue/Cyan : Exploration Layer (tau_e)|
| - Static grid obstacles & avenue lines | - Red : Avoidance Layer (tau_a) |
| - Frontier task circles (? / ID) | - Gold/Yel : Bayesian Target Belief |
| - Target vehicle markers (Wanted/Found)| |
| - Robot position glyphs & headings +---------------------------------------+
| - Historic breadcrumb trails (:) | PANEL 3: TELEMETRY & PROGRESS |
| - Translucent 90-deg FOV vision wedges | - Exploration Coverage Curve (%) |
| | - Real-time Mission State Badge |
| | - 5-Stage Multi-Stage Confidence HUD |
+----------------------------------------+---------------------------------------+
```
Sources: [visualization/swarm_visualizer.py:98-142](), [visualization/swarm_visualizer.py:285-412]()
### 4. Animated GIF Compilation
When `Pillow` is installed, sequential PNG frames saved every `--render-every` ticks are compiled into `/swarm_mission_animation.gif` at 4 FPS ($250\,\text{ms}$ duration per frame), visualizing the full search trajectory.
Sources: [scripts/run_simulation.py:304-320]()
### 5. Post-Mission Debrief Report
At the end of execution, `scripts/run_simulation.py` logs an operational summary:
```text
============================================================================
MISSION DEBRIEF REPORT
============================================================================
Scenario : easy_target
Total Simulation Time : 24.0 seconds (Wall clock: 0.82s)
Exploration Coverage : 64.2%
Completed Tasks : 14
Reassignment Events : 2
Failed Robots Handled : 0
Recovered Orphan Tasks: 0
Vehicle Detections : 3 visual sighting(s)
False Positives Disc. : 0 non-target vehicle(s) rejected
Target Confirmations : 1
Target Vehicle Status : 🎯 TARGET CONFIRMED & REPORTED
- Target Plate : DXB-88392
- Discovered By : robot_3
- Discovery Time : 22.4 s
- Match Confidence : 94.6%
- Physical Location : (25.0, 25.0)
============================================================================
```
Sources: [scripts/run_simulation.py:336-360]()
---
## 5. Verification & Fault-Tolerant Execution Flow
The simulation execution sequence coordinates perception, allocation, kinematics, and dynamic fault recovery in a tight discrete loop:
```mermaid
sequenceDiagram
autonumber
participant Runner as scripts/run_simulation.py
participant Coord as SwarmCoordinator
participant FailDet as FailureDetector
participant ACO as ACOAllocationStrategy
participant Robot as RobotAgent (Fleet)
participant Viz as SwarmVisualizer
Runner->>Coord: setup_simulation(args)
Coord->>Robot: register_robot(agent_1..4)
loop Every Simulation Tick (dt = 0.2s)
opt Tick == inject_failure_tick
Runner->>Coord: simulate_robot_failure(robot_id, reason)
Coord->>FailDet: mark_failed(robot_id)
Coord->>ACO: recover_orphaned_tasks(robot_id)
ACO-->>Coord: tasks returned to PENDING pool
end
Coord->>Robot: step_coordination(timestamp, dt)
Robot->>Robot: Update SLAM & Raycast FOV
opt Vehicle Detected in FOV
Robot->>Robot: Plate OCR & Fuzzy String Match
opt Match Score >= Threshold
Robot->>Coord: report_target_found(plate, pose, confidence)
Coord->>Coord: Deposit Recruitment Pheromone (tau_s)
end
end
Coord->>ACO: evaluate_frontier_tasks()
opt Tick % render_every == 0
Coord->>Viz: render_svg_map()
opt Matplotlib Available
Coord->>Viz: render_matplotlib_dashboard()
end
end
end
Runner->>Viz: render_ascii_map() & save final summaries
```
### Dynamic Robot Failure Injection Mechanics
When `--inject-failure-tick` is reached (such as tick 20 in Scenario E), `coordinator.simulate_robot_failure()` triggers the following recovery pipeline:
1. **Status Transition**: The target robot state changes to `RobotState.FAILED` with motor actuation arrested ($v=0, \omega=0$).
2. **Obstacle Registration**: The failed robot pose is added to the urban environment grid as a physical obstacle cell to prevent collisions by peer robots.
3. **Task Harvesting**: Any task currently marked as `TaskStatus.ASSIGNED` to the failed robot is immediately revoked, reset to `TaskStatus.PENDING`, and submitted to the ACO bidding queue.
4. **Pheromone Redistribution**: Local avoidance pheromone ($\tau_a$) is cleared to permit surviving agents to service the abandoned sector.
Sources: [scripts/run_simulation.py:219-229](), [simulation/scenarios.py:187-208](), [visualization/swarm_visualizer.py:251-254]()
### Automated Verification & Benchmark Testing
The test suite and benchmark harness validate all components against baseline requirements:
```bash
# Run unit and integration tests (including visualizer and simulation runner tests)
pytest tests/test_visualization.py -v
# Run the full 8-scenario AI benchmark evaluation suite
python3 scripts/evaluate_ai_benchmarks.py
```
`evaluate_ai_benchmarks.py` runs Scenarios A through H in sequence, computing global metrics including precision ($100\%$), recall ($100\%$), average OCR character accuracy ($96.5\%$), average inference latency ($< 1.0\,\text{ms}$), and swarm exploration coverage, exporting results to `output/ai_benchmark_report.json`.
Sources: [tests/test_visualization.py:29-163](), [evaluation/ai_benchmarks.py:182-263](), [scripts/evaluate_ai_benchmarks.py:20-100]()
---
## Summary
SwarmX provides a completely reproducible simulation platform for multi-agent urban target search. Developers can begin in under two minutes using standard Python (`python3 scripts/run_simulation.py --scenario easy_target`) to observe decentralized ACO task allocation, stigmergic pheromone diffusion, and multi-stage vehicle identification through terminal ASCII HUDs and SVG maps. Adding optional graphical dependencies enables rich Matplotlib dashboard telemetry and animated GIF generation, while containerized Docker launchers extend the exact same coordination logic into 3D ROS 2 and Gazebo Fortress worlds.
---
## 03. ACO Multi-Objective Scoring & Tri-Layer Stigmergy
> Detailed mechanics of the decentralized Ant Colony Optimization engine, the 8-factor score formulation, Boltzmann Softmax action selection, tri-layer pheromone evaporation dynamics, and recursive Bayesian target probability updates.
- Page Markdown: https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/pages/03-aco-multi-objective-scoring-tri-layer-stigmergy.md
- Generated: 2026-10-19T09:29:06.361Z
### Source Files
- `swarm/aco/aco_engine.py`
- `swarm/aco/task_scorer.py`
- `swarm/aco/aco_config.py`
- `swarm/pheromone/pheromone_map.py`
- `swarm/pheromone/target_probability_map.py`
- `configs/aco_params.yaml`
Relevant source files
The following files were used as context for generating this wiki page:
- [swarm/aco/aco_config.py](swarm/aco/aco_config.py)
- [swarm/aco/aco_engine.py](swarm/aco/aco_engine.py)
- [swarm/aco/task_scorer.py](swarm/aco/task_scorer.py)
- [swarm/pheromone/evaporation_model.py](swarm/pheromone/evaporation_model.py)
- [swarm/pheromone/pheromone_map.py](swarm/pheromone/pheromone_map.py)
- [swarm/pheromone/pheromone_updater.py](swarm/pheromone/pheromone_updater.py)
- [swarm/pheromone/target_probability_map.py](swarm/pheromone/target_probability_map.py)
- [swarm/task_allocation/search_history.py](swarm/task_allocation/search_history.py)
- [swarm/task_allocation/frontier_task.py](swarm/task_allocation/frontier_task.py)
- [swarm/task_allocation/allocation_strategy.py](swarm/task_allocation/allocation_strategy.py)
- [configs/aco_params.yaml](configs/aco_params.yaml)
# ACO Multi-Objective Scoring & Tri-Layer Stigmergy
The Ant Colony Optimization (ACO) allocation engine in `swarm_car_search` coordinates decentralized frontier selection across autonomous search robots. In GPS-denied or bandwidth-constrained urban search operations, relying on centralized auctioneers or global scheduling causes single points of failure and communication bottlenecks. Instead, this system coordinates agents through **stigmergy**—indirect coordination through environmental modifications—combining multi-layer synthetic pheromone fields, recursive Bayesian belief distributions, and spatio-temporal search memories.
At the core of each agent's decision-making process is an 8-factor multi-objective attractiveness evaluator (`TaskScorer`) coupled with a scale-invariant Boltzmann Softmax action selector (`ACOEngine`). This architecture balances pure geographic exploration, fast vehicle candidate verification, travel distance efficiency, peer congestion dispersion, and vehicle battery health, preventing swarm clustering and redundant searches.
```mermaid
flowchart TD
subgraph StigmergicBelief["Stigmergic & Spatial Belief State"]
direction TB
PM["PheromoneMap
(Tri-Layer Continuous Grid)"]
TPM["TargetProbabilityMap
(Log-Odds Bayesian Belief)"]
SSM["SearchSpatialMemory
(Dwell & Sighting Bins)"]
Peers["SwarmState Peers
(Active Commitments)"]
end
subgraph Inputs["Task & Robot Inputs"]
direction TB
Tasks["Candidate SearchTasks
(Status: PENDING)"]
Pose["Robot Pose2D & Battery"]
end
subgraph ScoringEngine["Multi-Objective Task Scoring (TaskScorer)"]
direction TB
Factors["Compute 8 Normalized Factors:
• Info Gain (+) • Target Prior (+)
• Historical Value (+) • Positive Pheromone (+)
• Distance Cost (-) • Peer Congestion (-)
• Collision Avoidance (-) • Redundancy Penalty (-)"]
NetUtil["Net Utility Synthesis
max(0.05, 1.0 + ΣPos - ΣNeg)"]
ClassRatio["Classical Ratio Integration
(τ^α · Urgency^δ) / Cost^ε"]
BatMod["Battery Modulation
0.4 + 0.6 · BatteryLevel"]
Factors --> NetUtil
NetUtil --> ClassRatio
ClassRatio --> BatMod
end
subgraph SelectionEngine["Action Selection (ACOEngine)"]
direction TB
Gate{"Selection Gate"}
DetChoice["Deterministic Top-1 Choice"]
EpsGreedy["Epsilon-Greedy Random Draw"]
Softmax["Boltzmann Softmax Distribution
Normalized Logits: (Score/Max - 1) / Temp"]
Sample["Cumulative Probability Sampling"]
Gate -->|deterministic=True or Temp ≤ 0.01| DetChoice
Gate -->|rand < epsilon_greedy| EpsGreedy
Gate -->|Stochastic Exploration| Softmax
Softmax --> Sample
end
StigmergicBelief --> Factors
Inputs --> Factors
BatMod --> Gate
Sample --> Selected["Assigned SearchTask"]
DetChoice --> Selected
EpsGreedy --> Selected
```
---
## 8-Factor Multi-Objective Scoring Formulation
The candidate task attractiveness evaluator is implemented in `TaskScorer.explain_score()` and `TaskScorer.compute_score()`. Each candidate frontier task $i$ is scored relative to an evaluating robot pose $\mathbf{x}_{\text{robot}}$ by synthesizing eight distinct positive rewards and negative cost penalties, modulated by classical ACO pheromone-heuristic ratios and robot battery status.
### Mathematical Formulation
The final attractiveness score is computed as:
$$\text{FinalScore}(i, \mathbf{x}_{\text{robot}}) = \text{NetUtility}_i \times \text{ClassicalRatio}_i \times \text{BatteryFactor}$$
Where:
1. **Net Utility Synthesis**:
$$\text{NetUtility}_i = \max\left(0.05, \; 1.0 + \sum_{p \in \text{Pos}} w_p \cdot f_p(i) - \sum_{n \in \text{Neg}} w_n \cdot f_n(i)\right)$$
2. **Classical ACO Pheromone-Distance Ratio**:
$$\text{ClassicalRatio}_i = \frac{(\tau_i)^{\alpha \cdot 0.5} \times (\text{Urgency}_i)^\delta}{(\text{Cost}_i)^{\epsilon \cdot 0.5}}$$
where $\tau_i = \max(0.01, \text{pheromone}_{\text{success}})$, $\text{Cost}_i = \max(d_{\min}, \|\mathbf{x}_{\text{task}} - \mathbf{x}_{\text{robot}}\|)$, and exponents $\alpha, \delta, \epsilon$ are defined in `ACOWeights`.
3. **Battery Factor**:
$$\text{BatteryFactor} = \max(0.2, \; \min(1.0, \; 0.4 + 0.6 \cdot \text{BatteryLevel}))$$
```python
# File: swarm/aco/task_scorer.py:121-132
net_utility = max(0.05, 1.0 + positive_components - negative_components)
tau_val = max(0.01, tau)
cost_val = max(self.config.min_distance_offset, euclidean_dist)
classical_ratio = ((tau_val ** (w.alpha * 0.5)) * (urgency ** w.delta)) / (cost_val ** (w.epsilon * 0.5))
battery_factor = max(0.2, min(1.0, 0.4 + 0.6 * battery_level))
final_score = net_utility * classical_ratio * battery_factor
```
Sources: [swarm/aco/task_scorer.py:106-133](), [swarm/aco/aco_config.py:10-32]()
---
### The 8 Scoring Factors Breakdown
The scoring pipeline decomposes task features into eight normalized factors in $[0.0, 1.0]$:
| # | Factor | Symbol | Weight Param | Type | Source & Normalization Formula |
|---|---|---|---|---|---|
| **1** | **Information Gain** | $f_{\text{ig}}$ | `w_information_gain` (1.0) | Reward | $\min\left(1.0, \max\left(0.05, \frac{\text{task.size}}{\text{max\_frontier\_size}}\right)\right)$ where $\text{max\_frontier\_size} = 30.0$ |
| **2** | **Target Prior** | $f_{\text{tp}}$ | `w_target_prior` (1.4) | Reward | $\text{TargetProbabilityMap.get\_box\_probability}(\text{bbox}, \text{mode}=\text{"max"})$ or task prior $P \in [0.01, 1.0]$ |
| **3** | **Historical Value** | $f_{\text{he}}$ | `w_historical_value` (0.8) | Reward | `SearchSpatialMemory.get_historical_value(x, y)`: returns $1.0$ for fresh cells, decayed by $e^{-0.40 \cdot \text{visits}}$ |
| **4** | **Exploration / Pheromone** | $f_{\text{fs}}$ | `w_exploration_value` (0.9) | Reward | $\min\left(1.0, \max\left(0.05, \frac{\tau_{\text{SUCCESS}}}{20.0}\right)\right)$ aggregated over task bounding box |
| **5** | **Distance Cost** | $f_{\text{dist}}$ | `w_distance_cost` (0.7) | Penalty | $\min\left(1.0, \max\left(0.01, \frac{\|\mathbf{x}_{\text{task}} - \mathbf{x}_{\text{robot}}\|}{\text{max\_norm\_dist}}\right)\right)$ where $\text{max\_dist} = 70.0\text{m}$ |
| **6** | **Congestion Cost** | $f_{\text{cong}}$ | `w_congestion_cost` (1.0) | Penalty | $\min\left(1.0, \frac{\text{congestion\_count}}{3.0}\right)$ based on peer robot task commitments |
| **7** | **Collision / Avoidance Risk** | $f_{\text{col}}$ | `w_collision_risk` (0.6) | Penalty | $\min\left(1.0, \max\left(0.0, \frac{k_{\text{AVOIDANCE}}}{15.0}\right)\right)$ aggregated from `AVOIDANCE` layer |
| **8** | **Redundancy Penalty** | $f_{\text{red}}$ | `w_redundant_penalty` (0.8) | Penalty | `SearchSpatialMemory.get_redundancy_penalty(x, y)`: $1.0 - e^{-0.35 \cdot (\text{visits}-1)}$ for zero-candidate sweeps |
### Dynamic Urgency Multiplier
Task priority is modulated dynamically by task classification (`TaskType`):
- `TaskType.TARGET_CONFIRMATION`: Multiplier **3.5** (immediate verification of high-probability vehicle detections)
- `TaskType.VEHICLE_INVESTIGATION`: Multiplier **2.2** (inspection of visual car candidate detections)
- `TaskType.EXPLORATION`: Multiplier **1.0** (baseline frontier mapping)
$$\text{Urgency} = \max(0.1, \; \text{task.priority} \times \text{UrgencyMultiplier})$$
Sources: [swarm/aco/task_scorer.py:52-105](), [swarm/aco/aco_config.py:22-32](), [swarm/task_allocation/frontier_task.py:21-26]()
---
## Boltzmann Softmax Action Selection
Action selection is performed in `ACOEngine.select_task()`. Rather than purely greedily selecting the task with the highest attractiveness score (which causes the entire swarm to converge on the same cluster), the engine employs a stochastic Boltzmann Softmax distribution with an $\epsilon$-greedy exploratory branch.
```
Candidate PENDING Tasks
│
▼
┌───────────────┐ Yes
│ deterministic?│─────────────► Return Highest Scored Task
└───────┬───────┘
│ No
▼
┌───────────────┐ Yes
│ rand < ε_opt? │─────────────► Uniform Random Task Choice
└───────┬───────┘
│ No
▼
┌───────────────┐ Yes
│ Temp ≤ 0.01? │─────────────► Return Highest Scored Task
└───────┬───────┘
│ No
▼
Compute Scale-Invariant Logits:
logit_i = ( (score_i / max_score) - 1.0 ) / Temperature
│
▼
Probabilities:
P_i = exp(logit_i) / Σ exp(logit_j)
│
▼
Sample Task via Cumulative Distribution
```
### Scale-Invariant Softmax Formulation
Standard Softmax $\frac{e^{S_i / T}}{\sum e^{S_j / T}}$ is vulnerable to numerical overflow when scores are large, and underflow when scores are small. `ACOEngine` implements a normalized relative logit transformation:
$$\text{logit}_i = \frac{\frac{S_i}{S_{\max}} - 1.0}{T}$$
Because $\frac{S_i}{S_{\max}} \in (0.0, 1.0]$, every logit is strictly bounded in:
$$\text{logit}_i \in \left[-\frac{1}{T}, \; 0.0\right]$$
The selection probabilities are calculated as:
$$P(i) = \frac{\exp(\text{logit}_i)}{\sum_{j=1}^{N} \exp(\text{logit}_j)}$$
```python
# File: swarm/aco/aco_engine.py:157-165
# Normalized relative logits: (score / max_score - 1.0) / temp
# Guaranteed to be in range [-1/temp, 0], preventing numerical underflow/overflow
weights = [math.exp(((s / max_score) - 1.0) / temp) for s in scores]
total_weight = sum(weights)
if total_weight <= 0.0:
return scored_tasks[0][0]
probabilities = [w / total_weight for w in weights]
```
### Selection Modes and Edge Cases
1. **Filtering**: Non-pending tasks (`ASSIGNED`, `IN_PROGRESS`, `COMPLETED`, `CANCELLED`) are filtered out prior to scoring.
2. **Deterministic Bypass**: If `config.stochastic.deterministic == True` or $T \le 0.01$, the highest-scoring candidate is chosen directly.
3. **$\epsilon$-Greedy Perturbation**: With probability $\epsilon$ (`epsilon_greedy = 0.05`), a random available pending task is chosen to maintain wide-area frontier dispersion.
4. **Near-Zero Scores**: If $S_{\max} \le 10^{-9}$, Softmax collapses to returning the top-ranked task.
Sources: [swarm/aco/aco_engine.py:103-176](), [swarm/aco/aco_config.py:34-40]()
---
## Tri-Layer Stigmergy & Pheromone Evaporation Dynamics
Synthetic pheromones are maintained on a continuous 2D spatial scalar grid by `PheromoneMap` and updated by `PheromoneUpdater`. Rather than maintaining a single scalar trail, the environment maintains three distinct stigmergic layers running concurrently.
```mermaid
classDiagram
class PheromoneLayerType {
<>
EXPLORATION
SUCCESS
AVOIDANCE
}
class EvaporationConfig {
+dict evaporation_rates
+float min_pheromone = 0.01
+float max_pheromone = 50.0
}
class PheromoneMap {
+float width
+float height
+float resolution
+dict layers
+world_to_grid(x, y) Tuple
+grid_to_world(gx, gy) Tuple
+get_value(layer, x, y) float
+set_value_grid(layer, gx, gy, val)
+get_box_aggregate(layer, min_x, min_y, max_x, max_y, mode) float
}
class EvaporationModel {
+EvaporationConfig config
+get_decay_factor(layer, dt) float
+clamp(value) float
}
class PheromoneUpdater {
+PheromoneMap map
+EvaporationModel evaporation_model
+deposit(layer, x, y, amount, radius)
+deposit_exploration(x, y, amount, radius)
+deposit_success(x, y, amount, radius)
+deposit_avoidance(x, y, amount, radius)
+step_evaporation(dt)
+sync_from_peer_update(layer, x, y, amount, radius)
}
PheromoneMap --> PheromoneLayerType
PheromoneMap --> EvaporationConfig
EvaporationModel --> EvaporationConfig
PheromoneUpdater --> PheromoneMap
PheromoneUpdater --> EvaporationModel
```
### Layer Semantics & Decay Rates
| Pheromone Layer | Enum Identifier | Evaporation Rate $\rho_k$ | Half-Life ($dt=1$) | Functional Purpose |
|---|---|---|---|---|
| **Exploration** | `PheromoneLayerType.EXPLORATION` | **0.01** (1% per step) | $\approx 69$ steps | Long-term memory of visited/mapped territory; prevents redundant long-term re-exploration. |
| **Success** | `PheromoneLayerType.SUCCESS` | **0.05** (5% per step) | $\approx 14$ steps | Recruitment beacon deposited when vehicle candidates or targets are detected; attracts peer agents for verification. |
| **Avoidance** | `PheromoneLayerType.AVOIDANCE` | **0.30** (30% per step) | $\approx 2$ steps | Short-term spatial reservation deposited around active robots to enforce spatial separation and prevent swarm clustering. |
### Discrete Evaporation Model
The evaporation update follows discrete exponential decay bounded within $[\tau_{\min}, \tau_{\max}]$:
$$\tau_k(t + \Delta t) = \text{clamp}\left(\tau_k(t) \cdot (1 - \rho_k)^{\Delta t}, \; \tau_{\min}, \; \tau_{\max}\right)$$
where $\tau_{\min} = 0.01$ and $\tau_{\max} = 50.0$. In `PheromoneUpdater.step_evaporation()`, this decay factor is computed once per layer and applied to all grid cells.
### Radial Gaussian Kernel Deposition
When a robot deposits pheromone with a non-zero radius $r \ge \text{resolution}$, `PheromoneUpdater.deposit()` spreads the deposit across neighbor cells using a 2D Gaussian kernel:
$$\Delta \tau(x, y) = \text{amount} \times \exp\left(-\frac{d^2}{2\sigma^2}\right) \quad \text{for } d \le r$$
where $\sigma = \frac{r}{2.0}$ and $d = \sqrt{(x - x_0)^2 + (y - y_0)^2}$.
```python
# File: swarm/pheromone/pheromone_updater.py:50-68
cell_radius = int(math.ceil(radius / self.map.resolution))
sigma = radius / 2.0
two_sigma_sq = 2.0 * sigma * sigma
for dy in range(-cell_radius, cell_radius + 1):
gy = cy + dy
for dx in range(-cell_radius, cell_radius + 1):
gx = cx + dx
dist = math.hypot(dx * self.map.resolution, dy * self.map.resolution)
if dist <= radius:
weight = math.exp(-(dist * dist) / two_sigma_sq)
curr = self.map.get_value_grid(layer, gx, gy)
self.map.set_value_grid(layer, gx, gy, curr + amount * weight)
```
Sources: [swarm/pheromone/evaporation_model.py:10-56](), [swarm/pheromone/pheromone_map.py:10-114](), [swarm/pheromone/pheromone_updater.py:25-101]()
---
## Recursive Bayesian Target Probability Updates
The `TargetProbabilityMap` maintains a spatial belief grid of target vehicle presence $P(\text{target} \mid \text{location}) \in [0.01, 0.99]$. To avoid catastrophic floating-point underflow or normalization overhead during sequential updates, the map stores beliefs in **log-odds** representation:
$$L(x, y) = \ln \left(\frac{P(x, y)}{1 - P(x, y)}\right)$$
Posterior probabilities are recovered via the logistic sigmoid function clamped to $[-10.0, 10.0]$:
$$P(x, y) = \frac{1}{1 + \exp\left(-\text{clip}(L(x, y), -10.0, 10.0)\right)}$$
```mermaid
sequenceDiagram
autonumber
participant Robot as Search Robot (Perception)
participant TPM as TargetProbabilityMap (Log-Odds)
participant Scorer as TaskScorer (ACO)
Note over Robot,TPM: Phase 1: Environment Prior Initialization
Robot->>TPM: initialize_from_environment(env)
TPM-->>TPM: Set zone priors P ∈ [0.05, 0.95] into log_odds grid
Note over Robot,TPM: Phase 2: Candidate Vehicle Sighted
Robot->>TPM: deposit_positive_evidence(x, y, confidence=0.85, radius=4.0)
TPM-->>TPM: ΔL = ln(conf / (1 - conf)) ∈ [0.2, 3.5]
TPM-->>TPM: Add Gaussian kernel: L = min(8.0, L + ΔL · weight)
Note over Robot,TPM: Phase 3: Empty Area Swept or Decoy Rejected
Robot->>TPM: deposit_negative_evidence(x, y, decay_factor=0.15, radius=2.5)
TPM-->>TPM: Subtract Gaussian kernel: L = max(-6.0, L - decay · weight)
Note over Scorer,TPM: Phase 4: ACO Task Scoring Query
Scorer->>TPM: get_box_probability(min_x, min_y, max_x, max_y, mode="max")
TPM-->>Scorer: Returns posterior target prior p_target ∈ [0.01, 0.99]
```
### Positive Bayesian Evidence Update
When a perception pipeline detects a potential target car (e.g., color match, license plate candidate, or CNN detection), `deposit_positive_evidence()` converts confidence into log-odds innovation $\Delta L$:
$$\Delta L = \ln \left(\frac{\max(0.01, \text{confidence})}{1.0 - \min(0.99, \text{confidence})}\right)$$
This update is clamped to $\Delta L \in [0.2, 3.5]$ and deposited with a 2D Gaussian kernel over radius $r=4.0\text{m}$ ($\sigma = \max(1.0, \frac{r_{\text{cells}}}{2})$):
$$L(r, c) \leftarrow \min\left(8.0, \; L(r, c) + \Delta L \cdot \exp\left(-\frac{d^2}{2\sigma^2}\right)\right)$$
### Negative Bayesian Evidence Update
When a robot surveys an empty region or confirms that a detected vehicle is an uninteresting civilian car or decoy, `deposit_negative_evidence()` decrements log-odds over radius $r=2.5\text{m}$ with decay factor $0.15$:
$$L(r, c) \leftarrow \max\left(-6.0, \; L(r, c) - 0.15 \cdot \exp\left(-\frac{d^2}{2\sigma^2}\right)\right)$$
Sources: [swarm/pheromone/target_probability_map.py:13-145]()
---
## Spatial Search Memory & Redundancy Tracking
Complementing the continuous pheromone fields, `SearchSpatialMemory` provides a discrete 2D spatial bin tracker ($\text{bin\_size} = 2.0\text{m}$) tracking visitation counts, visiting robot identifiers, candidate sightings, and confirmed decoy rejections.
```python
# File: swarm/task_allocation/search_history.py:69-88
def get_redundancy_penalty(self, x: float, y: float) -> float:
b = self._to_bin(x, y)
visits = self.visit_counts.get(b, 0)
candidates = self.candidate_counts.get(b, 0)
if visits <= 1:
return 0.0
# If candidates were sighted here, do not penalize revisiting for confirmation
if candidates > 0:
return 0.05
# Penalty rises smoothly with repeat unpromising visits
penalty = 1.0 - math.exp(-0.35 * (visits - 1))
return round(float(min(0.95, penalty)), 3)
```
1. **Redundancy Penalty ($f_{\text{red}}$)**:
- Zero penalty ($0.0$) if visits $\le 1$.
- If candidate sightings exist in the cell, penalty is capped at $0.05$ so robots are not deterred from re-visiting to confirm targets.
- For unpromising sweeps without sightings, penalty smoothly climbs towards $0.95$ via $1 - e^{-0.35(\text{visits} - 1)}$.
2. **Historical Exploration Value ($f_{\text{he}}$)**:
- Pristine, unvisited cells return $1.0$.
- Confirmed decoy locations with no candidate sightings return $0.15$.
- Cells with candidate sightings are boosted to $\min(1.0, 0.80 + 0.10 \times \text{candidates})$.
- Empty visits decay exponentially via $\max(0.10, e^{-0.40 \times \text{visits}})$.
Sources: [swarm/task_allocation/search_history.py:46-113]()
---
## Configuration & Parameter Reference
Configuration values can be set programmatically via dataclasses in `swarm/aco/aco_config.py` or loaded from `configs/aco_params.yaml`.
```yaml
# File: configs/aco_params.yaml
objective_function_weights:
alpha: 1.2 # Weight of positive pheromone trail (exploitation)
beta: 1.5 # Weight of information gain / unexplored ratio (exploration)
gamma: 1.8 # Weight of target vehicle detection likelihood
delta: 1.0 # Weight of search priority / mission urgency
epsilon: 1.3 # Penalty exponent for distance / travel cost
zeta: 2.0 # Penalty exponent for congestion (other robots targeting area)
theta: 1.5 # Penalty exponent for negative/recent visit pheromone
pheromone_dynamics:
evaporation_rates:
exploration: 0.01 # Slow decay for permanent environmental exploration memory
success: 0.05 # Moderate decay for vehicle sighting evidence
avoidance: 0.30 # Rapid decay for temporary robot reservation / congestion
deposit_values:
exploration_visit: 1.0
vehicle_detected: 5.0
confirmed_target_match: 20.0
active_reservation: 3.0
min_pheromone: 0.01
max_pheromone: 50.0
stochastic_selection:
temperature: 1.0 # Softmax temperature parameter (higher = more exploratory)
epsilon_greedy: 0.05 # Probability of picking random valid frontier for exploration
```
### Parameter Reference Table
| Category | Parameter | Default Value | Description |
|---|---|---|---|
| **Classical Exponents** | `alpha` | `1.2` | Exponent for positive pheromone $\tau$ in classical ratio |
| | `beta` | `1.5` | Exponent for frontier size / information gain |
| | `gamma` | `1.8` | Exponent for vehicle detection prior probability |
| | `delta` | `1.0` | Exponent for mission priority / urgency |
| | `epsilon` | `1.3` | Exponent for Euclidean distance travel cost penalty |
| | `zeta` | `2.0` | Exponent for peer congestion penalty |
| | `theta` | `1.5` | Exponent for avoidance / reservation pheromone |
| **Normalized Weights** | `w_information_gain` | `1.0` | Reward weight for normalized frontier cluster size |
| | `w_target_prior` | `1.4` | Reward weight for Bayesian target probability $P(\text{target})$ |
| | `w_historical_value` | `0.8` | Reward weight for spatial memory exploration history |
| | `w_exploration_value`| `0.9` | Reward weight for `SUCCESS` layer pheromone aggregation |
| | `w_distance_cost` | `0.7` | Penalty weight for Euclidean travel distance |
| | `w_congestion_cost` | `1.0` | Penalty weight for peer robot target commitment overlap |
| | `w_collision_risk` | `0.6` | Penalty weight for `AVOIDANCE` layer pheromone |
| | `w_redundant_penalty`| `0.8` | Penalty weight for repeatedly swept empty zones |
| | `w_battery_health` | `0.5` | Base weight for battery state modulation |
| **Stochastic Selection**| `temperature` | `1.0` | Boltzmann Softmax exploration temperature ($T \to 0$ exploits, $T \to \infty$ uniform) |
| | `epsilon_greedy` | `0.05` | Probability of choosing a random pending task |
| | `deterministic` | `False` | When `True`, bypasses Softmax and selects top score strictly |
| **Pheromone Map** | `min_pheromone` | `0.01` | Lower bound floor for all pheromone layers |
| | `max_pheromone` | `50.0` | Upper saturation ceiling for pheromones |
| | `resolution` | `0.5` m | Spatial cell resolution of continuous 2D pheromone map |
| **Normalization** | `max_normalization_distance` | `70.0` m | Normalization distance denominator |
| | `max_frontier_size` | `30.0` | Normalization frontier cluster size denominator |
Sources: [configs/aco_params.yaml:1-36](), [swarm/aco/aco_config.py:10-80]()
---
## Architectural Integration & Data Flow
Within the multi-robot coordination architecture, the ACO components integrate through `ACOAllocationStrategy` and `TaskManager`. During every task allocation cycle, the robot:
1. Receives candidate frontiers extracted by SLAM and deduplicated by `TaskManager`.
2. Queries active peer commitments from `SwarmState.peers` to compute peer congestion counts.
3. Samples the continuous `PheromoneMap` bounding boxes for `SUCCESS` and `AVOIDANCE` trails.
4. Samples the `TargetProbabilityMap` bounding box for posterior vehicle belief.
5. Ingests dwell counts and decoy rejection records from `SearchSpatialMemory`.
6. Executes `TaskScorer.compute_score()` across all candidates and passes scores to `ACOEngine.select_task()` for Boltzmann Softmax sampling.
7. Deposits `AVOIDANCE` pheromone upon commitment and broadcasts stigmergic updates to peer robots via `sync_from_peer_update()`.
This closed-loop stigmergic feedback cycle allows the multi-robot swarm to rapidly locate target vehicles in large, cluttered environments without centralized communication bottlenecks.
---
## 04. Perception Pipeline & Subsystem Contracts
> The 5-stage target identification pipeline covering raycast FOV occlusion, Levenshtein fuzzy license plate OCR, appearance verification, temporal tracking, and the abstract ISLAMProvider interface contract decoupling Swarm from SLAM.
- Page Markdown: https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/pages/04-perception-pipeline-subsystem-contracts.md
- Generated: 2026-10-19T09:34:56.817Z
### Source Files
- `interfaces/slam_interface.py`
- `interfaces/perception_interface.py`
- `perception/matching_engine.py`
- `perception/vehicle_detector.py`
- `perception/plate_recognizer.py`
- `swarm/communication/local_comm.py`
Relevant source files
The following files were used as context for generating this wiki page:
- [interfaces/slam_interface.py](interfaces/slam_interface.py)
- [interfaces/perception_interface.py](interfaces/perception_interface.py)
- [perception/matching_engine.py](perception/matching_engine.py)
- [perception/vehicle_detector.py](perception/vehicle_detector.py)
- [perception/plate_recognizer.py](perception/plate_recognizer.py)
- [perception/appearance_verifier.py](perception/appearance_verifier.py)
- [perception/temporal_tracker.py](perception/temporal_tracker.py)
- [slam_providers/base_slam_provider.py](slam_providers/base_slam_provider.py)
- [swarm/communication/local_comm.py](swarm/communication/local_comm.py)
- [swarm/communication/swarm_messages.py](swarm/communication/swarm_messages.py)
- [swarm/core/robot_agent.py](swarm/core/robot_agent.py)
- [ros2_ws/src/swarm_gazebo/swarm_gazebo/slam_bridge.py](ros2_ws/src/swarm_gazebo/swarm_gazebo/slam_bridge.py)
- [ros2_ws/src/swarm_gazebo/swarm_gazebo/perception_bridge.py](ros2_ws/src/swarm_gazebo/swarm_gazebo/perception_bridge.py)
# Perception Pipeline & Subsystem Contracts
The target search and identification subsystem in **SwarmX** provides an end-to-end vision, verification, and spatial abstraction architecture designed for multi-robot urban search operations. Autonomous search teams face real-world sensing degradations, including optical occlusions from buildings, motion blur, character ambiguities under optical character recognition (OCR), and transient false alarms caused by non-target civilian vehicles. To resolve these challenges reliably without saturating the swarm with premature alarms, SwarmX implements a rigorous **5-Stage Target Identification Pipeline** that progressively filters sensory inputs and fuses detection confidence across geometry, optical text, vehicle appearance, and multi-frame temporal voting.
Equally critical is the architectural decoupling between spatial awareness and multi-robot swarm decision-making. The swarm intelligence subsystem (task allocation, Ant Colony Optimization, stigmergy, and path planning) never interfaces directly with raw sensor hardware (LiDAR, wheel encoders, IMUs) or concrete SLAM nodes. Instead, all spatial localization, occupancy grids, and exploration frontiers flow through the formal **`ISLAMProvider`** abstract interface contract. This structural boundary establishes zero compile-time coupling between the SLAM and Swarm development teams, allowing seamless runtime substitution between mock simulators, Gazebo bridge nodes, and production ROS 2 navigation stacks.
---
## 1. Subsystem Architecture & Interface Contracts
The SwarmX platform enforces strict interface segregation using Python Abstract Base Classes (`abc.ABC`) and immutable data carriers. The codebase divides engineering ownership cleanly: the SLAM and State Estimation layer is maintained independently from Swarm Coordination and Perception.
```mermaid
classDiagram
class ISLAMProvider {
<>
+get_latest_state(robot_id: str) SLAMState*
+is_healthy(robot_id: str) bool*
}
class BaseSLAMProvider {
#_latest_states: Dict~str, SLAMState~
#_health_status: Dict~str, bool~
+update_cached_state(robot_id: str, state: SLAMState)
+set_healthy(robot_id: str, healthy: bool)
+get_latest_state(robot_id: str) SLAMState
+is_healthy(robot_id: str) bool
}
class GazeboSLAMBridge {
+map_width_m: float
+map_height_m: float
+resolution: float
+_on_scan(robot_id, msg)
+_on_odom(robot_id, msg)
+get_latest_state(robot_id: str) SLAMState
+is_healthy(robot_id: str) bool
}
class IPerceptionProvider {
<>
+detect_vehicles(robot_id: str, camera_frame_id: str) VehicleDetectionEvent*
+recognize_license_plate(detection: VehicleDetectionEvent) PlateDetectionEvent*
+match_watchlist(plate_event: PlateDetectionEvent) MatchEvent*
}
class IVehicleDetector {
<>
+detect_vehicles(robot_id: str, robot_pose: Pose2D, timestamp: float, vehicles: List~TargetVehicle~, env: UrbanEnvironment) List*
}
class SimulatedVehicleDetector {
+max_detection_range: float
+fov_radians: float
+base_confidence: float
+_has_line_of_sight(rx, ry, tx, ty, env) bool
+detect_vehicles(...) List
}
class SimulatedPlateRecognizer {
+base_ocr_confidence: float
+max_ocr_distance: float
+character_noise_rate: float
+recognize_license_plate(detection, vehicle) PlateDetectionEvent
}
class VehicleAppearanceVerifier {
+color_weight: float
+shape_weight: float
+class_weight: float
+compute_color_similarity(observed_color, target_color) float
+compute_shape_similarity(observed_ar, target_ar) float
+verify_appearance(observed_class, observed_color, observed_ar) Tuple
}
class MultiFrameTargetTracker {
+required_confirmations: int
+association_radius: float
+ewma_alpha: float
+tracks: Dict~str, CandidateTrack~
+add_observation(...) Tuple
+prune_stale_tracks(current_time, max_age)
}
class MatchingEngine {
+match_threshold: float
+candidate_threshold: float
+w_vehicle: float
+w_plate: float
+w_ocr: float
+w_appearance: float
+w_temporal: float
+add_target_vehicle(target: TargetVehicle)
+evaluate_multi_stage_match(plate_event, detection_event) Tuple
+classify_plate(plate_event, detection_event) Tuple
}
ISLAMProvider <|-- BaseSLAMProvider
BaseSLAMProvider <|-- GazeboSLAMBridge
IVehicleDetector <|-- SimulatedVehicleDetector
MatchingEngine o-- VehicleAppearanceVerifier
MatchingEngine o-- MultiFrameTargetTracker
```
Sources: [interfaces/slam_interface.py:69-100](), [interfaces/perception_interface.py:86-103](), [perception/vehicle_detector.py:33-47](), [perception/matching_engine.py:51-98](), [slam_providers/base_slam_provider.py:10-40]()
---
### The `ISLAMProvider` Contract
The `ISLAMProvider` interface in `interfaces/slam_interface.py` decouples the Swarm Decision Layer from low-level mapping and localization algorithms. A consumer requires only two operations from any SLAM engine:
```python
# interfaces/slam_interface.py
class ISLAMProvider(ABC):
"""
Abstract Base Class representing the contract that any SLAM provider
(MockSLAMProvider or real ROS2-backed SLAM) must satisfy.
"""
@abstractmethod
def get_latest_state(self, robot_id: str) -> SLAMState:
"""Retrieve the most recent SLAM estimation for the specified robot."""
pass
@abstractmethod
def is_healthy(self, robot_id: str) -> bool:
"""Check if the SLAM tracking and mapping pipeline is healthy."""
pass
```
Sources: [interfaces/slam_interface.py:69-100]()
#### Core Data Structures
The interface defines four strongly-typed, serialization-safe data structures:
| Data Structure | Immutability | Core Fields | Description |
| :--- | :--- | :--- | :--- |
| **`Pose2D`** | `@dataclass(frozen=True)` | `x: float`, `y: float`, `theta: float` | 2D position and heading angle in radians $[-\pi, \pi]$ within the global reference frame. |
| **`OccupancyGrid2D`** | `@dataclass` | `width: int`, `height: int`, `resolution: float`, `origin: Pose2D`, `data: List[int]` | 1D row-major array representing urban occupancy: `-1` (unexplored/unknown), `0` (free/navigable), and `100` (occupied/obstacle). |
| **`FrontierCluster`** | `@dataclass(frozen=True)` | `cluster_id: str`, `centroid_x: float`, `centroid_y: float`, `size: int`, `bounding_box: Tuple[float, float, float, float]` | Clustered boundary cells separating explored free space from unknown territory, representing candidate exploration targets for swarm dispatch. |
| **`SLAMState`** | `@dataclass` | `robot_id: str`, `timestamp: float`, `pose: Pose2D`, `linear_velocity: float`, `angular_velocity: float`, `occupancy_grid: OccupancyGrid2D`, `frontiers: List[FrontierCluster]`, `exploration_ratio: float`, `navigation_cost_map: Optional[List[float]]` | The complete instantaneous perception and spatial snapshot passed to the robot agent and coordinator per control cycle. |
Sources: [interfaces/slam_interface.py:12-68]()
#### Runtime Provider Polymorphism
SwarmX supports dynamic dependency injection across environments without altering swarm logic:
1. **`MockSLAMProvider`**: Generates synthetic expanding circular occupancy grids, deterministic frontier clusters, and kinematic poses for headless unit testing and high-speed Monte Carlo simulations.
2. **`BaseSLAMProvider`**: Implements thread-safe caching via an internal `_latest_states: Dict[str, SLAMState]` store and health tracking (`is_healthy`).
3. **`GazeboSLAMBridge`**: Extends `ISLAMProvider` and ROS 2 `rclpy.node.Node`, subscribing directly to `/{robot_id}/odom` and LiDAR laser scans, populating real-time occupancy grids and detecting frontier clusters via raycasting.
Sources: [slam_providers/base_slam_provider.py:10-40](), [ros2_ws/src/swarm_gazebo/swarm_gazebo/slam_bridge.py:39-75](), [ros2_ws/src/swarm_gazebo/swarm_gazebo/slam_bridge.py:321-351]()
---
## 2. The 5-Stage Target Identification Pipeline
The visual detection pipeline isolates target vehicles from civilian distractors through five sequential, interdependent processing stages. Rather than relying on a monolithic classification probability, each stage produces a dedicated confidence coefficient ($C_i \in [0.0, 1.0]$) that is preserved in the emitted event for explainability.
```mermaid
flowchart TD
subgraph S1["Stage 1: Spatial & Geometric Gating"]
A["Sensor Frame & Robot Pose2D"] --> B{"Distance <= 14.0m?"}
B -- No --> R1["Reject: Out of Range"]
B -- Yes --> C{"Bearing within FOV 90°?"}
C -- No --> R2["Reject: Out of View"]
C -- Yes --> D{"Bresenham Raycast: Grid != 100?"}
D -- No --> R3["Reject: Wall Occlusion"]
D -- Yes --> E["Emit VehicleDetectionEvent (C_veh)"]
end
subgraph S2["Stage 2: Plate Localization"]
E --> F["Regress Plate Sub-BoundingBox2D"]
F --> G["Compute Plate Likelihood (C_plate)"]
end
subgraph S3["Stage 3: OCR & String Distance"]
G --> H["Crop Plate Region & Attenuate with Distance"]
H --> I["Optical Confusion Substitution (8<->B, 0<->O, etc.)"]
I --> J["Levenshtein Distance vs Watchlist (S_plate)"]
J --> K["OCR Confidence (C_ocr x S_plate)"]
end
subgraph S4["Stage 4: Appearance Verification"]
E --> L["Extract Class, Color RGB, & Aspect Ratio"]
L --> M["Class Concordance (1.0 vs 0.2)"]
L --> N["Euclidean RGB Distance Normalization"]
L --> O["Aspect Ratio Geometry Concordance"]
M & N & O --> P["Composite Appearance Score (C_app)"]
end
subgraph S5["Stage 5: Multi-Frame Temporal Tracking"]
K & P --> Q["Associate Spatial Track (< 3.0m Radius)"]
Q --> R["EWMA Filter on Instantaneous Score (alpha=0.40)"]
R --> S["Modal Plate Majority Voting (S_modal)"]
S --> T["Temporal Confidence Calculation (C_temp)"]
end
subgraph S6["Fusion & Decision Gate"]
E & G & K & P & T --> U["Weighted Fusion: C = Sum(w_i * C_i)"]
U --> V{"Classification Check"}
V -- "C >= 0.75 & S_plate >= 0.45 & k >= k_req" --> CONF["CONFIRMED MATCH\nTrigger Alarm, Stop, Broadcast"]
V -- "C >= 0.40 & S_plate >= 0.45" --> CAND["CANDIDATE SIGHTING\nNavigate Closer for High-Res OCR"]
V -- "Otherwise" --> NON["NON-TARGET\nDiscard False Alarm & Resume Search"]
end
```
Sources: [perception/vehicle_detector.py:49-182](), [perception/plate_recognizer.py:24-87](), [perception/appearance_verifier.py:34-136](), [perception/temporal_tracker.py:84-222](), [perception/matching_engine.py:51-208]()
---
### Stage 1: Raycast FOV & Occlusion-Aware Vehicle Detection
Visual candidate detection is handled by `SimulatedVehicleDetector` in `perception/vehicle_detector.py`. To eliminate unphysical detections through walls, candidates undergo three sequential spatial tests:
1. **Range Test**: Target Euclidean distance $d = \sqrt{(x_v - x_r)^2 + (y_v - y_r)^2}$ must not exceed `max_detection_range` (default $14.0\text{ m}$):
$$d \le \min(R_{\max}, R_{\text{detection\_radius}})$$
2. **Angular Field-of-View (FoV) Test**: Heading error $\Delta \theta$ relative to the robot's heading $\theta$ must fall within the camera aperture (default $90^\circ$, or $\pm 45^\circ$ from camera center line):
$$\theta_{\text{bearing}} = \text{atan2}(y_v - y_r, x_v - x_r)$$
$$\Delta \theta = (\theta_{\text{bearing}} - \theta + \pi) \pmod{2\pi} - \pi$$
$$|\Delta \theta| \le \frac{\text{FOV}_{\text{radians}}}{2.0}$$
3. **Obstacle Line-of-Sight (LOS) Raycasting**: An occlusion ray is stepped between $(x_r, y_r)$ and $(x_v, y_v)$ through the `UrbanEnvironment` 2D grid at half-cell resolution intervals ($\Delta s = 0.5 \times \text{resolution}$). If any intermediate grid cell contains value `100` (static obstacle/wall), the vehicle is occluded and rejected:
```python
# perception/vehicle_detector.py
def _has_line_of_sight(self, rx: float, ry: float, tx: float, ty: float, env: Optional[UrbanEnvironment] = None) -> bool:
if env is None:
return True
dx, dy = tx - rx, ty - ry
dist = math.hypot(dx, dy)
if dist < 1e-3:
return True
steps = int(math.ceil(dist / (env.resolution * 0.5)))
step_x, step_y = dx / steps, dy / steps
for i in range(1, steps):
check_x = rx + i * step_x
check_y = ry + i * step_y
g = env.world_to_grid(check_x, check_y)
if g is not None and env.get_cell(g[0], g[1]) == 100:
return False # Obstacle blocks line of sight
return True
```
If the candidate passes all checks, its distance-attenuated detection confidence $C_{\text{veh}}$ is calculated:
$$f_{\text{dist}} = \max\left(0.0, 1.0 - \frac{d}{1.5 \times R_{\max}}\right)$$
$$C_{\text{veh}} = \min\left(0.99, \max(0.50, C_{\text{base}} \times f_{\text{dist}})\right)$$
The detector projects an image bounding box `BoundingBox2D` whose scale shrinks inversely with distance: $\text{scale} = \max(0.15, \min(0.60, 3.0 / \max(1.0, d)))$.
Sources: [perception/vehicle_detector.py:70-162]()
---
### Stage 2: License Plate Localization
Once a vehicle bounding box is confirmed, Stage 2 localizes the rectangular license plate crop within the lower-central segment of the vehicle silhouette (`BoundingBox2D(ymin=0.60, xmin=0.35, ymax=0.80, xmax=0.65)`). The localization probability $C_{\text{plate}}$ is conditioned on distance-dependent optical resolution:
$$C_{\text{plate}} = \begin{cases}
\min\left(0.98, \max\left(0.85, 0.95 - \frac{d}{20.0}\right)\right) & \text{if } d \le 6.0\text{ m} \\
\min\left(0.85, \max\left(0.60, 0.90 - \frac{d}{15.0}\right)\right) & \text{if } 6.0\text{ m} < d \le 10.0\text{ m} \\
\max\left(0.35, 0.70 - \frac{d}{12.0}\right) & \text{if } d > 10.0\text{ m}
\end{cases}$$
Candidate string visibility also exhibits discrete degradation: vehicles beyond $5.0\text{ m}$ only reveal a partial prefix (`plate[:3] + "-???"`), while vehicles beyond $9.0\text{ m}$ yield no legible plate preview, signaling the swarm agent to close range.
Sources: [perception/vehicle_detector.py:146-171](), [perception/plate_recognizer.py:85-87]()
---
### Stage 3: OCR Engine & Levenshtein Fuzzy Alignment
Stage 3 executes character extraction on the cropped license plate image via `SimulatedPlateRecognizer` (`perception/plate_recognizer.py`) and aligns the result against the target watchlist via `MatchingEngine` (`perception/matching_engine.py`).
#### Optical Confusion Simulation
At long distances ($d > 7.0\text{ m}$) or when environmental noise is injected (`character_noise_rate > 0.0`), optical character recognition engines confuse topologically similar glyphs. The recognizer simulates realistic neural network OCR misclassifications using `OCR_CONFUSION_MAP`:
$$\text{Confusion Mappings:}\quad '8' \longleftrightarrow 'B',\quad '0' \longleftrightarrow 'O',\quad '2' \longleftrightarrow 'Z',\quad '5' \longleftrightarrow 'S',\quad '1' \longleftrightarrow 'I'$$
When corruption occurs, the character confidence is discounted: $\text{conf}_{\text{char}} \leftarrow \max(0.35, \text{conf}_{\text{char}} \times 0.70)$.
Sources: [perception/plate_recognizer.py:15-22](), [perception/plate_recognizer.py:67-81]()
#### Levenshtein Fuzzy String Alignment
To prevent typo-induced misidentifications, `MatchingEngine` strips non-alphanumeric punctuation and computes normalized string similarity:
```python
# perception/matching_engine.py
def normalize_plate(plate: str) -> str:
"""Strips hyphens, spaces, and standardizes to uppercase."""
return "".join(ch for ch in plate.upper() if ch.isalnum())
def levenshtein_similarity(s1: str, s2: str) -> float:
"""Computes normalized string similarity in range [0.0, 1.0] using Levenshtein distance."""
if s1 == s2:
return 1.0
if not s1 or not s2:
return 0.0
len1, len2 = len(s1), len(s2)
dp = [[0] * (len2 + 1) for _ in range(len1 + 1)]
for i in range(len1 + 1): dp[i][0] = i
for j in range(len2 + 1): dp[0][j] = j
for i in range(1, len1 + 1):
for j in range(1, len2 + 1):
cost = 0 if s1[i - 1] == s2[j - 1] else 1
dp[i][j] = min(dp[i - 1][j] + 1, dp[i][j - 1] + 1, dp[i - 1][j - 1] + cost)
dist = dp[len1][len2]
return max(0.0, 1.0 - (dist / float(max(len1, len2))))
```
The composite optical score for Stage 3 combines the OCR model confidence and string alignment score:
$$S_3 = C_{\text{ocr}} \times S_{\text{plate}}$$
Sources: [perception/matching_engine.py:17-50](), [perception/matching_engine.py:128-131]()
---
### Stage 4: Vehicle Appearance Verification
The `VehicleAppearanceVerifier` (`perception/appearance_verifier.py`) prevents plate spoofing or OCR false positives by validating the visual signature of the vehicle against the ground-truth `TargetAppearanceProfile`:
$$C_{\text{app}} = w_{\text{color}} S_{\text{color}} + w_{\text{shape}} S_{\text{shape}} + w_{\text{class}} S_{\text{class}}$$
Default feature weights are: $w_{\text{color}} = 0.55$, $w_{\text{shape}} = 0.25$, and $w_{\text{class}} = 0.20$.
```
+---------------------------------------------------------------------------------+
| VehicleAppearanceVerifier Component Breakdown |
+---------------------------------------------------------------------------------+
| 1. Class Match: |
| S_class = 1.0 if observed_class == target_class else 0.20 |
| |
| 2. Calibrated Color Similarity: |
| RGB Euclidean Distance: D_rgb = sqrt(sum((RGB_obs - RGB_tgt)^2)) |
| Normalized Distance: d_norm = D_rgb / 441.673 (where 441.673 = sqrt(3*255^2))|
| Sharpened Similarity: S_color = max(0.0, 1.0 - d_norm)^1.5 |
| |
| 3. Shape & Geometry Concordance: |
| Aspect Ratio (length / width): Sedan=2.2, SUV=1.8, Truck=2.8, Van=1.9 |
| Aspect Ratio Error: e_ar = abs(AR_obs - AR_tgt) / AR_tgt |
| Shape Similarity: S_shape = max(0.0, 1.0 - e_ar) |
+---------------------------------------------------------------------------------+
```
Sources: [perception/appearance_verifier.py:12-31](), [perception/appearance_verifier.py:53-136]()
---
### Stage 5: Multi-Frame Temporal Tracking & Consistency
A single video frame is vulnerable to illumination flashes, glints, or momentary occlusions. The `MultiFrameTargetTracker` (`perception/temporal_tracker.py`) enforces multi-frame spatial and temporal persistence.
1. **Spatial Track Gating**: Observations are assigned to an active `CandidateTrack` if Euclidean separation is within `association_radius = 3.0\text{ m}`. The track centroid is updated incrementally:
$$\bar{x}_{n+1} = \frac{n \cdot \bar{x}_n + x_{\text{obs}}}{n + 1},\quad \bar{y}_{n+1} = \frac{n \cdot \bar{y}_n + y_{\text{obs}}}{n + 1}$$
2. **Instantaneous Frame Scoring**:
$$\text{score}_{\text{inst}} = 0.20 C_{\text{veh}} + 0.35 (C_{\text{ocr}} \times S_{\text{plate}}) + 0.25 C_{\text{app}} + 0.20 S_{\text{plate}}$$
3. **EWMA Confidence Accumulation**: Instantaneous scores are smoothed over time using Exponential Weighted Moving Average with learning rate $\alpha = 0.40$:
$$C_{\text{accum}}^{(t)} = (1.0 - \alpha) C_{\text{accum}}^{(t-1)} + \alpha \cdot \text{score}_{\text{inst}}$$
4. **Modal Plate Majority Voting**: Detections within the track vote on the plate string. The modal plate frequency $R_{\text{modal}} = \frac{\text{count}_{\text{modal}}}{k}$ is weighted by its Levenshtein similarity to target plate $S_{\text{modal}}$:
$$R_{\text{match}} = R_{\text{modal}} \times S_{\text{modal}}$$
5. **Frame Count Factor**: Confidence scales with observation count $k$ up to required confirmations $k_{\text{req}}$:
$$F_{\text{factor}} = \min\left(1.0, \frac{k}{k_{\text{req}}}\right)$$
6. **Composite Temporal Confidence**:
$$C_{\text{temp}} = 0.50 C_{\text{accum}} + 0.30 R_{\text{match}} + 0.20 F_{\text{factor}}$$
A track is verified only when $k \ge k_{\text{req}}$, $C_{\text{temp}} \ge 0.82$, and $S_{\text{modal}} \ge 0.75$. Stale unconfirmed tracks older than $60.0\text{ s}$ are pruned via `prune_stale_tracks()`.
Sources: [perception/temporal_tracker.py:84-235](), [perception/temporal_tracker.py:239-247]()
---
## 3. Multi-Stage Fusion & Swarm Decision Logic
The overall target verification confidence $C$ combines all five stages into a single metric computed by `MatchingEngine.evaluate_multi_stage_match()`:
$$C = w_{\text{veh}} C_{\text{veh}} + w_{\text{plate}} C_{\text{plate}} + w_{\text{ocr}} (C_{\text{ocr}} \times S_{\text{plate}}) + w_{\text{app}} C_{\text{app}} + w_{\text{temp}} C_{\text{temp}}$$
```python
# perception/matching_engine.py
composite_confidence = (
self.w_vehicle * c_veh # w_vehicle = 0.15
+ self.w_plate * c_plate # w_plate = 0.15
+ self.w_ocr * ocr_composite # w_ocr = 0.30, ocr_composite = c_ocr * best_similarity
+ self.w_appearance * c_app # w_appearance = 0.20
+ self.w_temporal * c_temp # w_temporal = 0.20
)
composite_confidence = max(0.0, min(0.99, round(composite_confidence, 3)))
```
Sources: [perception/matching_engine.py:71-85](), [perception/matching_engine.py:156-164]()
### Fusion Weights & Stage Diagnostics
| Stage | Metric Symbol | Default Weight | Key Diagnostic Output | Failure Consequence |
| :--- | :---: | :---: | :--- | :--- |
| **1. Vehicle Detector** | $C_{\text{veh}}$ | `0.15` | `stage_breakdown["vehicle_confidence"]` | Wall occlusion or out-of-range leads to immediate frame drop. |
| **2. Plate Localizer** | $C_{\text{plate}}$ | `0.15` | `stage_breakdown["plate_confidence"]` | Low resolution attenuates overall score, forcing closer navigation. |
| **3. Plate OCR & Similarity** | $C_{\text{ocr}} \times S_{\text{plate}}$ | `0.30` | `stage_breakdown["plate_similarity"]` | Character typos mitigated by Levenshtein matching; decoy plates rejected. |
| **4. Appearance Verifier** | $C_{\text{app}}$ | `0.20` | `stage_breakdown["appearance_match"]` | Wrong vehicle color or class penalizes score, stopping spoofed plates. |
| **5. Temporal Tracker** | $C_{\text{temp}}$ | `0.20` | `stage_breakdown["temporal_consistency"]` | Transient single-frame spikes cannot cross confirmation threshold. |
Sources: [perception/matching_engine.py:71-77](), [perception/matching_engine.py:165-175]()
---
### Classification Tiers & FSM State Transitions
The robot agent's Finite State Machine (`swarm/states/state_machine.py`) responds dynamically to the classification returned by `classify_plate()`:
```
+---------------------------------------+
| Perception Scan Result |
+---------------------------------------+
|
+------------------------+------------------------+
| |
[C >= 0.75 & S_plate >= 0.45 & k >= k_req] [C >= 0.40 & S_plate >= 0.45]
| |
v v
+--------------+ +--------------+
| CONFIRMED | | CANDIDATE |
+--------------+ +--------------+
| 1. Stop Robot Kinematics | 1. Lock Waypoint to (cx, cy)
| 2. FSM -> REPORTING | 2. FSM -> NAVIGATING
| 3. Broadcast Alert to Swarm Bus | 3. Close range (< 5.0m)
| 4. Escalate Mission Completion | 4. Re-evaluate with high-res OCR
+--------------+ +--------------+
|
| (All other outcomes: C < 0.40 or S_plate < 0.45)
v
+--------------+
| NON_TARGET |
+--------------+
| 1. Discard Observation / Increment False Positives
| 2. FSM -> SEARCHING (continue frontier patrol)
+--------------+
```
```python
# swarm/core/robot_agent.py
if classification == "CONFIRMED" and match_event is not None:
self.sim_robot.stop()
self._reporting_elapsed = 0.0
self.fsm.transition_to(RobotState.REPORTING, timestamp, reason="CONFIRMED_TARGET_MATCH")
self.last_confirmed_match = match_event
self.total_target_confirmations += 1
if self.comm_channel is not None:
alert_msg = create_vehicle_alert_message(self.robot_id, timestamp, match_event)
self.comm_channel.broadcast(alert_msg)
return match_event
elif classification == "CANDIDATE":
cx, cy = det_event.estimated_global_position
self.assigned_waypoint = (cx, cy)
if self.current_state != RobotState.NAVIGATING:
self.fsm.transition_to(RobotState.NAVIGATING, timestamp, reason="APPROACHING_CANDIDATE_FOR_OCR")
return None
else:
self.total_false_positives += 1
if self.current_state != RobotState.SEARCHING:
self.fsm.transition_to(RobotState.SEARCHING, timestamp, reason="NON_TARGET_OR_UNCONFIRMED")
return None
```
Sources: [swarm/core/robot_agent.py:215-248]()
---
## 4. Swarm Communication & Stigmergic Feedback
When a robot confirms or suspects a target, perception events propagate across the multi-robot collective through direct messaging and stigmergic pheromone fields.
```mermaid
sequenceDiagram
autonumber
participant Cam as Camera / Detector
participant Agent as RobotAgent (robot_1)
participant FSM as StateMachine
participant Comm as LocalMessageBus
participant Peer as RobotAgent (robot_2)
participant Coord as SwarmCoordinator
Note over Cam,Agent: Step 1-4 Perception Sweep
Cam->>Agent: VehicleDetectionEvent + PlateDetectionEvent
Agent->>Agent: evaluate_multi_stage_match()
alt Classification == CANDIDATE
Agent->>FSM: transition_to(NAVIGATING, "APPROACHING_CANDIDATE")
Agent->>Agent: Move closer to candidate (< 5m) for crisp OCR
else Classification == CONFIRMED
Agent->>Agent: sim_robot.stop()
Agent->>FSM: transition_to(REPORTING, "CONFIRMED_TARGET_MATCH")
Agent->>Comm: broadcast(create_vehicle_alert_message)
Comm->>Peer: deliver(SwarmMessage: TARGET_FOUND)
Comm->>Coord: deliver(SwarmMessage: TARGET_FOUND)
Coord->>Coord: Escalate mission status & halt exploration
Peer->>Peer: Abort search, transition to rendezvous / assist
end
```
Sources: [swarm/core/robot_agent.py:215-231](), [swarm/communication/swarm_messages.py:134-150](), [swarm/communication/local_comm.py:28-33]()
### Inter-Robot Alert Message Schema
Direct communication is implemented via `LocalMessageBus` (`swarm/communication/local_comm.py`), which delivers zero-latency in-memory messages. A confirmed sighting triggers `create_vehicle_alert_message()` in `swarm/communication/swarm_messages.py`:
```python
# swarm/communication/swarm_messages.py
def create_vehicle_alert_message(
sender_id: str,
timestamp: float,
match_event: MatchEvent,
) -> SwarmMessage:
"""Constructs a high-priority vehicle detection broadcast message."""
payload = {
"target_id": match_event.target_id,
"plate_number": match_event.plate_number,
"match_confidence": match_event.match_confidence,
"confirmed": match_event.confirmed,
"location": list(match_event.location),
"evidence_reference": match_event.evidence_reference,
}
return SwarmMessage(
message_id=str(uuid.uuid4()),
sender_id=sender_id,
timestamp=timestamp,
msg_type=MessageType.TARGET_FOUND,
payload=payload,
)
```
Sources: [swarm/communication/swarm_messages.py:134-155]()
### Stigmergic Recruitment
In addition to direct network broadcasts, sightings interact with the swarm's spatial pheromone fields:
* **Recruitment Pheromone ($\tau_s$, rate $\rho = 0.05$)**: When a candidate vehicle is detected ($C \ge 0.40$), recruitment pheromone is deposited at the candidate's coordinates $(x_v, y_v)$. Surrounding robots sensing the local gradient redirect their exploration paths toward the candidate to provide multi-angle camera verification.
* **Avoidance & Reservation Pheromone ($\tau_a$, rate $\rho = 0.30$)**: Once verified or during approach, robots deposit reservation pheromone to prevent multiple agents from colliding or redundantly searching the same vehicle spot.
Sources: [swarm/core/robot_agent.py:215-231](), [ros2_ws/src/swarm_gazebo/swarm_gazebo/perception_bridge.py:180-200]()
---
## 5. Summary
The SwarmX perception architecture integrates a rigorous 5-stage target identification pipeline with an abstracted SLAM boundary. By sequentially cascading **raycast field-of-view occlusion gating**, **license plate sub-region localization**, **Levenshtein fuzzy OCR alignment**, **RGB Euclidean appearance verification**, and **multi-frame temporal tracking**, the platform suppresses optical noise and rejects decoy vehicles before escalating mission-critical alerts. Furthermore, decoupling state estimation behind the abstract **`ISLAMProvider`** contract guarantees that swarm navigation and coordination algorithms remain modular and portable across unit testing mocks, 3D Gazebo physics simulations, and physical ROS 2 hardware deployments.
---
## 05. Autonomous Fault Detection & Dynamic Reallocation
> Telemetry monitoring for silent heartbeat timeouts, battery exhaustion, and kinematic stalls, coupled with dynamic task revocation, avoidance pheromone deposition over disabled units, and immediate peer reallocation.
- Page Markdown: https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/pages/05-autonomous-fault-detection-dynamic-reallocation.md
- Generated: 2026-10-19T09:39:11.199Z
### Source Files
- `swarm/failure_handling/failure_detector.py`
- `swarm/failure_handling/recovery_manager.py`
- `swarm/core/robot_agent.py`
- `swarm/task_allocation/task_manager.py`
- `tests/test_failure_handling.py`
Relevant source files
The following files were used as context for generating this wiki page:
- [swarm/failure_handling/failure_detector.py](swarm/failure_handling/failure_detector.py)
- [swarm/failure_handling/recovery_manager.py](swarm/failure_handling/recovery_manager.py)
- [swarm/core/robot_agent.py](swarm/core/robot_agent.py)
- [swarm/task_allocation/task_manager.py](swarm/task_allocation/task_manager.py)
- [swarm/core/swarm_coordinator.py](swarm/core/swarm_coordinator.py)
- [swarm/core/swarm_state.py](swarm/core/swarm_state.py)
- [swarm/pheromone/pheromone_updater.py](swarm/pheromone/pheromone_updater.py)
- [tests/test_failure_handling.py](tests/test_failure_handling.py)
# Autonomous Fault Detection & Dynamic Reallocation
In distributed multi-robot search missions, individual agents navigate dense, unpredictable urban terrain where physical obstacles, radio frequency (RF) shadowing, mechanical drivetrain failures, and power depletion are persistent operational risks. Without autonomous fault handling, a single immobilized or disconnected robot can create persistent blind spots, hold uncompleted frontier tasks indefinitely, and induce catastrophic bottlenecks across the fleet.
The fault tolerance subsystem in `swarm_car_search` resolves these challenges through a closed-loop resilience architecture. The system pairs continuous multi-modal telemetry monitoring in [`FailureDetector`](swarm/failure_handling/failure_detector.py) with rapid remediation in [`RecoveryManager`](swarm/failure_handling/recovery_manager.py). When an agent ceases transmission, exhausts its battery reserve, gets entrapped, or throws an unrecoverable internal exception, the framework revokes task commitments, marks the disabled unit as a physical obstacle using high-potency stigmergic avoidance pheromones, and immediately dispatches orphaned objectives to the most suitable active peers.
---
## System Architecture & Resilience Pipeline
The fault detection and recovery pipeline operates across both centralized coordination loops and decentralized peer-to-peer state tracking. It monitors high-frequency telemetry, identifies failure patterns against configurable thresholds, and executes sequential remediation without human intervention.
```mermaid
flowchart TD
subgraph Fleet ["Robot Fleet Layer (swarm/core/robot_agent.py)"]
R1["Robot Agent 1\n(Active / Stalled)"]
R2["Robot Agent 2\n(Idle Candidate)"]
R3["Robot Agent 3\n(Navigating Candidate)"]
HB["Heartbeat & Pose Telemetry\n(create_heartbeat)"]
R1 --> HB
end
subgraph Monitoring ["Telemetry Ingestion & Anomaly Detection (swarm/failure_handling/failure_detector.py)"]
FD["FailureDetector\n(record_telemetry / check_failures)"]
HB --> FD
FD --> T1{"Heartbeat Silent?\n(> 6.0s)"}
FD --> T2{"Battery Critical?\n(< 10%)"}
FD --> T3{"Kinematic Stall?\n(Delta pos < 0.3m for 8.0s)"}
FD --> T4{"State == FAILED?\n(Explicit Fault)"}
end
subgraph Recovery ["Recovery & Dynamic Reallocation (swarm/failure_handling/recovery_manager.py)"]
RM["RecoveryManager\n(handle_failure)"]
T1 -- Yes --> FE["RobotFailureEvent"]
T2 -- Yes --> FE
T3 -- Yes --> FE
T4 -- Yes --> FE
FE --> RM
A1["1. Deactivate Peer\n(SwarmState.mark_peer_inactive)"]
A2["2. Revoke Orphaned Task\n(TaskManager.release_task -> PENDING)"]
A3["3. Deposit Avoidance Pheromone\n(tau_a kernel at wreck coords)"]
A4["4. Dynamic Suitability Reallocation\n(Dist * Battery * Workload)"]
RM --> A1
RM --> A2
RM --> A3
RM --> A4
end
subgraph StateAndStigmergy ["State & Environment Updates"]
TM["TaskManager\n(Pending Task Queue)"]
PM["PheromoneMap\n(Layer: AVOIDANCE)"]
SS["Decentralized SwarmState\n(Peer Registry)"]
A1 --> SS
A2 --> TM
A3 --> PM
A4 --> R2
end
```
The system operates continuously within the primary coordination loop in [`SwarmCoordinator.step_coordination()`](swarm/core/swarm_coordinator.py#L226-L281), while individual agents maintain local peer watchdogs via [`SwarmState.check_peer_timeouts()`](swarm/core/swarm_state.py#L96-L109).
Sources: [swarm/core/swarm_coordinator.py:226-281](), [swarm/failure_handling/recovery_manager.py:45-72]()
---
## Failure Detection Engine: Multi-Mode Anomaly Identification
The [`FailureDetector`](swarm/failure_handling/failure_detector.py#L58-L69) ingests state telemetry (`Pose2D`, `battery_level`, `RobotState`, `assigned_task_id`, and `timestamp`) on each tick and checks against four distinct failure modalities.
```
+----------------------------------------------------------------------------------------------------+
| FAILURE DETECTION MODES |
+----------------------+-----------------------------+-----------------------------------------------+
| Mode (FailureType) | Default Threshold Trigger | Root Cause & System Response |
+----------------------+-----------------------------+-----------------------------------------------+
| HEARTBEAT_TIMEOUT | silence > 6.0s | RF blackout, transceiver destruction, crash. |
| | | Declares robot unreachable; frees its task. |
+----------------------+-----------------------------+-----------------------------------------------+
| BATTERY_CRITICAL | battery_level < 0.10 (10%) | Cell exhaustion below operational reserve. |
| | | Preemptively halts agent before complete death.|
+----------------------+-----------------------------+-----------------------------------------------+
| STALLED | displacement < 0.3m | Physical wheel trap, urban obstacle lockup. |
| | over duration > 8.0s | Triggered only in NAVIGATING or SEARCHING. |
+----------------------+-----------------------------+-----------------------------------------------+
| EXPLICIT_FAULT | state == RobotState.FAILED | Internal diagnostic failure, motor burnout, |
| | | or unrecoverable sensor/SLAM fault. |
+----------------------+-----------------------------+-----------------------------------------------+
```
### Telemetry Tracking & Kinematic Stall Anchoring
To prevent false-positive stall alerts when a robot is legitimately waiting in `IDLE`, pausing during `VERIFYING`, or broadcasting in `REPORTING`, stall evaluation is strictly restricted to active transit states: `RobotState.NAVIGATING` and `RobotState.SEARCHING`.
The detector establishes a spatial anchor (`stall_anchor_pose`) and tracks the duration an agent remains within `stall_distance_threshold`:
$$\Delta d = \sqrt{(x_{\text{current}} - x_{\text{anchor}})^2 + (y_{\text{current}} - y_{\text{anchor}})^2}$$
If $\Delta d \ge 0.3\,\text{m}$, the anchor updates to the current pose and the timer resets. If $\Delta d < 0.3\,\text{m}$ persists beyond `stall_timeout_sec` ($8.0\,\text{s}$), a `FailureType.STALLED` event is triggered.
```python
# swarm/failure_handling/failure_detector.py:101-118
# Evaluate movement for stall tracking
if state in (RobotState.NAVIGATING, RobotState.SEARCHING):
if rec.stall_anchor_pose is None:
rec.stall_anchor_pose = pose
rec.stall_start_time = timestamp
else:
dist = math.hypot(pose.x - rec.stall_anchor_pose.x, pose.y - rec.stall_anchor_pose.y)
if dist >= self.config.stall_distance_threshold:
# Made sufficient displacement; reset stall anchor
rec.stall_anchor_pose = pose
rec.stall_start_time = timestamp
else:
# Idle, returning, or reporting: not considered stalled
rec.stall_anchor_pose = None
rec.stall_start_time = None
```
### Failure Event Lifecycle & FSM State Representation
```mermaid
stateDiagram-v2
[*] --> Monitored: record_telemetry()
state Monitored {
Healthy: Active Heartbeats & Valid Kinematics
}
Healthy --> HeartbeatTimeout: silence > heartbeat_timeout_sec
Healthy --> BatteryCritical: battery < battery_critical_threshold
Healthy --> KinematicStall: dist < 0.3m for > stall_timeout_sec
Healthy --> ExplicitFault: state == RobotState.FAILED
state DetectedFailure {
HeartbeatTimeout
BatteryCritical
KinematicStall
ExplicitFault
}
DetectedFailure --> FailureEventGenerated: RobotFailureEvent Instantiated
FailureEventGenerated --> RecoveryExecuted: RecoveryManager.handle_failure()
RecoveryExecuted --> FailedRegistry: Registered in _failed_robots
FailedRegistry --> Monitored: reset_robot() (Reboot / Reset)
```
Sources: [swarm/failure_handling/failure_detector.py:16-56](), [swarm/failure_handling/failure_detector.py:119-198]()
---
## The Recovery Pipeline: Revocation, Stigmergic Marking, and Peer Dispatch
When [`FailureDetector.check_failures()`](swarm/failure_handling/failure_detector.py#L119-L198) emits a [`RobotFailureEvent`](swarm/failure_handling/failure_detector.py#L25-L34), the [`RecoveryManager`](swarm/failure_handling/recovery_manager.py#L45-L72) executes a 4-step remediation protocol via `handle_failure()`.
```mermaid
sequenceDiagram
autonumber
participant SC as SwarmCoordinator
participant FD as FailureDetector
participant RM as RecoveryManager
participant TM as TaskManager
participant PU as PheromoneUpdater / Map
participant SS as SwarmStates (Peers)
participant Peer as Best Surviving Peer
SC->>FD: check_failures(timestamp)
FD-->>SC: [RobotFailureEvent(r_fail, last_pose, task_id)]
SC->>RM: handle_failure(event, task_manager, swarm_states, ...)
Note over RM,SS: Step 1: Decentralized Peer Deactivation
RM->>SS: mark_peer_inactive(r_fail)
Note over RM,TM: Step 2: Task Revocation & Release
RM->>TM: release_task(task_id, timestamp)
TM-->>TM: Status -> PENDING, assigned_robot = None
RM->>SS: active_tasks[task_id].mark_released(timestamp)
Note over RM,PU: Step 3: Stigmergic Wreck Avoidance Deposit
RM->>PU: deposit_avoidance(x, y, amount=12.0, radius=2.5)
PU-->>PU: Inject radial Gaussian tau_a obstacle kernel
Note over RM,Peer: Step 4: Multi-Criteria Dynamic Reassignment
RM->>RM: Score surviving peers (Suitability = dist * battery * workload)
RM->>Peer: assign_task(chosen_task, timestamp)
RM->>TM: assign_task(task_id, peer_id, timestamp)
RM-->>SC: RecoveryResult(actions=[DEACTIVATED, RELEASED, AVOIDANCE, REASSIGNED])
```
### 1. Peer Inactivation Across Swarm States
The failed agent is marked inactive in every surviving robot's local [`SwarmState`](swarm/core/swarm_state.py#L86-L95). This purges the agent from peer availability lists, halts stale coordinate updates, and prevents peer-to-peer consensus deadlocks.
### 2. Orphaned Task Revocation
If the disabled unit was navigating toward or executing an active [`SearchTask`](swarm/task_allocation/frontier_task.py), ownership is immediately revoked. The task's status is reset from `ASSIGNED` or `IN_PROGRESS` back to `PENDING`, clearing its assigned robot field:
```python
# swarm/failure_handling/recovery_manager.py:90-99
if orphaned_task is not None and orphaned_task.status in (TaskStatus.ASSIGNED, TaskStatus.IN_PROGRESS):
result.orphaned_task_id = orphaned_task.task_id
task_manager.release_task(orphaned_task.task_id, event.timestamp)
result.actions_taken.append(RecoveryAction.TASK_RELEASED)
# Synchronize task release in decentralized swarm states
for sstate in swarm_states.values():
if orphaned_task.task_id in sstate.active_tasks:
sstate.active_tasks[orphaned_task.task_id].mark_released(event.timestamp)
```
### 3. Wreck Avoidance Pheromone Deposition
An immobilized robot in an urban corridor represents a physical obstacle. To prevent healthy robots from colliding with or queuing behind the disabled unit, the [`RecoveryManager`](swarm/failure_handling/recovery_manager.py#L101-L108) instructs the [`PheromoneUpdater`](swarm/pheromone/pheromone_updater.py#L78-L80) to deposit high-intensity avoidance pheromone ($\tau_a$) directly onto the obstacle grid:
$$\tau_a(\mathbf{x}) \leftarrow \tau_a(\mathbf{x}) + A_{\text{wreck}} \cdot \exp\left(-\frac{\|\mathbf{x} - \mathbf{x}_{\text{wreck}}\|^2}{2\sigma^2}\right)$$
Default configuration parameters:
- `wreck_avoidance_amount` ($A_{\text{wreck}}$): `12.0`
- `wreck_avoidance_radius` ($r_{\text{avoid}}$): `2.5` meters ($\sigma = r_{\text{avoid}} / 2 = 1.25\,\text{m}$)
Because the ACO path planner and task scoring engine penalize paths crossing high $\tau_a$ concentrations, surviving robots naturally navigate around the disabled vehicle.
### 4. Immediate Dynamic Reallocation with Multi-Criteria Scoring
Rather than waiting for normal periodic task allocation ticks, [`RecoveryManager`](swarm/failure_handling/recovery_manager.py#L110-L165) calculates suitability scores for all active peers:
$$S(r_i) = F_{\text{dist}}(r_i) \times F_{\text{batt}}(r_i) \times F_{\text{workload}}(r_i)$$
Where:
- **Distance Factor**: $F_{\text{dist}}(r_i) = \frac{1.0}{1.0 + 0.05 \cdot \|\mathbf{p}_i - \mathbf{c}_{\text{task}}\|}$
- **Battery Factor**: $F_{\text{batt}}(r_i) = \max(0.1, E_{\text{batt}, i})$
- **Workload Factor**:
- $1.0$ if the candidate is in `RobotState.IDLE` or `RobotState.REASSIGNING`.
- $0.65$ if the candidate is in `RobotState.NAVIGATING` on background exploration (preemptible). Robots handling urgent `TaskType.TARGET_CONFIRMATION` are non-preemptible.
The candidate with the highest suitability score is chosen, and the task is immediately assigned and committed in [`TaskManager`](swarm/task_allocation/task_manager.py#L120-L125).
Sources: [swarm/failure_handling/recovery_manager.py:73-165](), [swarm/pheromone/pheromone_updater.py:49-69]()
---
## Decentralized State Synchronization & Local Peer Watchdogs
While the [`SwarmCoordinator`](swarm/core/swarm_coordinator.py) executes centralized safety monitoring during simulation runs, the architecture supports decentralized swarm resilience. Each robot instance maintains an independent [`SwarmState`](swarm/core/swarm_state.py#L50-L63) that tracks peer heartbeats:
```python
# swarm/core/swarm_state.py:96-109
def check_peer_timeouts(self, current_time: float, timeout_sec: float) -> List[str]:
"""
Evaluates peer heartbeats against timeout threshold.
Marks silent peers as inactive, releases any assigned tasks,
and returns list of newly timed-out robot IDs.
"""
timed_out: List[str] = []
for peer_id, peer in self.peers.items():
if peer.is_active:
if (current_time - peer.last_heartbeat) > timeout_sec:
self.mark_peer_inactive(peer_id)
timed_out.append(peer_id)
return timed_out
```
```
+----------------------------------------------------------------------------------------------------+
| SUBSYSTEM ROLES & FAILURE RESPONSES |
+--------------------------+-------------------------------------------------------------------------+
| Component | Role in Failure Handling |
+--------------------------+-------------------------------------------------------------------------+
| FailureDetector | Ingests telemetry, tracks stall anchors, validates thresholds, |
| | and emits RobotFailureEvent. |
+--------------------------+-------------------------------------------------------------------------+
| RecoveryManager | Coordinates peer deactivation, task revocation, avoidance deposition, |
| | and immediate peer suitability scoring. |
+--------------------------+-------------------------------------------------------------------------+
| RobotAgent | Executes onboard FSM, emits periodic heartbeats, switches to |
| | COMMUNICATION_LOST fallback if isolated, and handles task preemption. |
+--------------------------+-------------------------------------------------------------------------+
| SwarmState | Maintains peer active/inactive maps and releases local task locks |
| | upon detecting peer heartbeat timeouts. |
+--------------------------+-------------------------------------------------------------------------+
| TaskManager | Manages global task lifecycle (PENDING, ASSIGNED, IN_PROGRESS), |
| | resetting status and clearing robot bindings upon task release. |
+--------------------------+-------------------------------------------------------------------------+
| PheromoneUpdater | Injects radial Gaussian avoidance kernels (tau_a) at failure coordinates|
| | to divert peer traffic. |
+--------------------------+-------------------------------------------------------------------------+
```
Sources: [swarm/core/swarm_state.py:86-109](), [swarm/core/robot_agent.py:275-288]()
---
## Verification & Fault Injection Scenarios
The test suite in [`tests/test_failure_handling.py`](tests/test_failure_handling.py) verifies failure detection and recovery mechanics across unit, integration, and decentralized boundary conditions.
```mermaid
flowchart LR
subgraph TestSuites ["Test Suites in tests/test_failure_handling.py"]
T1["TestFailureDetector\n- Heartbeat timeout\n- Low battery (<10%)\n- Kinematic stall\n- Explicit FSM fault\n- Robot reset"]
T2["TestRecoveryManager\n- Task release to PENDING\n- Avoidance pheromone > 10.0\n- RecoveryAction audit trail"]
T3["TestSwarmCoordinatorResilience\n- Dynamic reassignment\n- RF communication loss\n- Telemetry metrics tracking"]
T4["TestDecentralizedSwarmStateTimeout\n- Peer timeout evaluation\n- Local task status release"]
end
```
### Verified Test Cases
1. **Heartbeat Silence Detection (`test_heartbeat_timeout`)**:
- Asserts an agent transmitting at $t = 0.0\,\text{s}$ remains healthy at $t = 3.0\,\text{s}$ under a $5.0\,\text{s}$ threshold, and transitions to `FailureType.HEARTBEAT_TIMEOUT` at $t = 6.0\,\text{s}$.
2. **Critical Battery Cutoff (`test_battery_critical`)**:
- Confirms that a telemetry record with $8\%$ battery against a $10\%$ threshold immediately generates a `FailureType.BATTERY_CRITICAL` event.
3. **Kinematic Wheel Stall (`test_stalled_robot`)**:
- Confirms that an agent moving only $0.05\,\text{m}$ over $7.0\,\text{s}$ while in `RobotState.NAVIGATING` exceeds the $6.0\,\text{s}$ stall timeout and triggers `FailureType.STALLED`.
4. **Wreck Marking & Task Reclamation (`test_task_release_and_avoidance_deposit`)**:
- Asserts that upon failure of `robot_fail`, its assigned task reverts to `TaskStatus.PENDING`, and the avoidance layer in `PheromoneMap` at the crash coordinates records $\tau_a > 10.0$.
5. **Dynamic Swarm Peer Takeover (`test_dynamic_reassignment_on_simulated_failure`)**:
- Injects a simulated motor driver failure (`MOTOR_DRIVER_BURNOUT`) on `robot_1` via `SwarmCoordinator.simulate_robot_failure()`.
- Verifies that `robot_2` assumes ownership of the orphaned task within a single coordination tick ($0.1\,\text{s}$ interval) and transitions to `RobotState.NAVIGATING`.
Sources: [tests/test_failure_handling.py:35-139](), [tests/test_failure_handling.py:140-195](), [tests/test_failure_handling.py:196-297](), [tests/test_failure_handling.py:298-338]()
---
## Summary
The autonomous fault detection and dynamic reallocation architecture provides continuous operational resilience for multi-robot search missions. By decoupling telemetry monitoring in [`FailureDetector`](swarm/failure_handling/failure_detector.py) from remedial action in [`RecoveryManager`](swarm/failure_handling/recovery_manager.py), the system isolates broken units, prevents urban bottlenecks via stigmergic avoidance pheromone deposition, and redistributes orphaned tasks based on proximity, battery reserves, and current workload. This design guarantees mission continuity without requiring centralized operator intervention.
---
## 06. Beyond the First 30 Minutes: 3D Gazebo & Extension Points
> Synthesis of initial learnings, bridging from 2D discrete simulation to containerized ROS 2 / Gazebo Fortress 3D worlds, running Monte Carlo benchmark evaluations against baselines, and architectural extension hooks.
- Page Markdown: https://grok-wiki.com/public/wiki/ivaziz-swarm-car-search-66a3d00cb601/pages/06-beyond-the-first-30-minutes-3d-gazebo-extension-points.md
- Generated: 2026-10-19T09:43:06.490Z
### Source Files
- `scripts/run_evaluation.py`
- `evaluation/ai_benchmarks.py`
- `run_gazebo_gui.sh`
- `docker/Dockerfile`
- `docs/thesis_chapter_swarm.md`
Relevant source files
The following files were used as context for generating this wiki page:
- [scripts/run_evaluation.py](scripts/run_evaluation.py)
- [evaluation/ai_benchmarks.py](evaluation/ai_benchmarks.py)
- [run_gazebo_gui.sh](run_gazebo_gui.sh)
- [docker/Dockerfile](docker/Dockerfile)
- [docs/thesis_chapter_swarm.md](docs/thesis_chapter_swarm.md)
- [ros2_ws/src/swarm_gazebo/launch/swarm_simulation.launch.py](ros2_ws/src/swarm_gazebo/launch/swarm_simulation.launch.py)
- [ros2_ws/src/swarm_gazebo/swarm_gazebo/slam_bridge.py](ros2_ws/src/swarm_gazebo/swarm_gazebo/slam_bridge.py)
- [ros2_ws/src/swarm_gazebo/swarm_gazebo/perception_bridge.py](ros2_ws/src/swarm_gazebo/swarm_gazebo/perception_bridge.py)
- [ros2_ws/src/swarm_gazebo/swarm_gazebo/robot_controller.py](ros2_ws/src/swarm_gazebo/swarm_gazebo/robot_controller.py)
- [ros2_ws/src/swarm_gazebo/swarm_gazebo/coordinator_node.py](ros2_ws/src/swarm_gazebo/swarm_gazebo/coordinator_node.py)
- [interfaces/slam_interface.py](interfaces/slam_interface.py)
- [interfaces/perception_interface.py](interfaces/perception_interface.py)
- [interfaces/server_interface.py](interfaces/server_interface.py)
- [swarm/task_allocation/allocation_strategy.py](swarm/task_allocation/allocation_strategy.py)
- [swarm/aco/aco_config.py](swarm/aco/aco_config.py)
- [evaluation/baselines.py](evaluation/baselines.py)
- [evaluation/metrics_collector.py](evaluation/metrics_collector.py)
# Beyond the First 30 Minutes: 3D Gazebo & Extension Points
Once developers have verified the core 2D discrete simulation and understood the bio-inspired Ant Colony Optimization (ACO) coordination loop, the next engineering phase involves elevating the system into high-fidelity environments and rigorous scientific evaluation. The SwarmX architecture provides two primary pathways beyond the introductory setup: deploying into a containerized 3D physics environment powered by ROS 2 Humble and Gazebo Fortress, and executing statistical Monte Carlo comparative benchmarks against classical exploration baselines.
This guide bridges the gap between discrete algorithmic prototyping and full-stack robotics engineering. It walks through containerized headless and graphical Gazebo execution on host hardware, maps the inter-process ROS 2 parameter bridge communication channels, analyzes the 15-metric comparative evaluation pipeline and 4-variant ACO ablation studies, and defines explicit architectural extension points across SLAM, perception, task allocation, and external telemetry.
---
## 1. Bridging from 2D Discrete Simulation to 3D Gazebo Fortress
The fast 2D simulation uses discretized grid cells ($1.0\,\text{m}$ resolution) and forward Euler differential kinematics for rapid algorithm iteration. To test real sensor noise, continuous physics, wheel slip, and geometric line-of-sight occlusions, SwarmX provides a fully containerized ROS 2 Humble and Gazebo Fortress (Ignition) 3D simulation stack.
### 1.1 Containerized Runtime & Display Architecture
To ensure cross-platform reproducibility across Linux and macOS (specifically Apple Silicon ARM64), the 3D stack is isolated inside a multi-stage Docker container. Rather than demanding complex native X11 forwarding from the host, the container incorporates a self-contained virtual display server (`Xvfb`), window manager (`fluxbox`), VNC server (`x11vnc`), and HTML5 WebSocket gateway (`websockify` + `novnc`).
```
+---------------------------------------------------------------------------------------+
| HOST MACHINE |
| - Web Browser: http://localhost:6080/vnc.html (noVNC HTML5 3D View) |
| - Native Screen Sharing: vnc://localhost:5900 (Apple Silicon / Desktop VNC) |
+-------------------------------------------+-------------------------------------------+
| Port 6080 (WS) / Port 5900 (RFB)
+-------------------------------------------v-------------------------------------------+
| DOCKER CONTAINER (swarm_gazebo:humble) |
| |
| +-----------------------+ +-------------------+ +-----------------------+ |
| | Virtual Display | --> | fluxbox (WM) | --> | x11vnc (:1 -> 5900) | |
| | Xvfb :1 1600x1000x24 | | | | websockify (:6080) | |
| +-----------+-----------+ +-------------------+ +-----------------------+ |
| | DISPLAY=:1 |
| +-----------v-------------------------------------------------------------------+ |
| | Gazebo Sim (Ignition Fortress) | |
| | - 50m x 50m Urban Search World (urban_search_world.world) | |
| | - Physics Engine, Differential AGV Spawners, Target Sedan (DXB-88392) | |
| | - 2D LiDAR Sensors, RGB Cameras, Wheel Odometry Encoders | |
| +---------------------------------------+---------------------------------------+ |
| | Ignition Transport Topics |
| +---------------------------------------v---------------------------------------+ |
| | ROS 2 <-> Gazebo Parameter Bridge (ros_gz_bridge) | |
| | - Clock, Odometry, LaserScan, RGB Images, cmd_vel Twists | |
| +---------------------------------------+---------------------------------------+ |
| | ROS 2 Topics (/robot_*/...) |
| +---------------------------------------v---------------------------------------+ |
| | SwarmX ROS 2 Orchestrator & Node Stack (MultiThreadedExecutor) | |
| | - GazeboSLAMBridge (ISLAMProvider) | |
| | - GazeboPerceptionBridge (Camera FOV + Fuzzy OCR Matcher) | |
| | - GazeboSwarmCoordinatorNode (ACO Task Scorer + Failure Recovery) | |
| | - GazeboRobotController (Closed-loop proportional waypoint navigation) | |
| +-------------------------------------------------------------------------------+ |
+---------------------------------------------------------------------------------------+
```
Sources: [docker/Dockerfile:1-56](), [run_gazebo_gui.sh:15-125]()
### 1.2 Topic Plumbing & The ROS–Gazebo Parameter Bridge
Communication between Gazebo Fortress physics and ROS 2 nodes is managed by `ros_gz_bridge parameter_bridge`. The bridge translates Ignition transport messages into standard ROS 2 message types across all four robot namespaces (`robot_1` through `robot_4`).
| Topic Pattern | Direction | ROS 2 Message Type | Gazebo Transport Type | Functional Role |
| :--- | :---: | :--- | :--- | :--- |
| `/clock` | `Gz -> ROS` | `rosgraph_msgs/msg/Clock` | `ignition.msgs.Clock` | Synchronizes ROS simulation time with Gazebo physics. |
| `/{rid}/cmd_vel` | `ROS -> Gz` | `geometry_msgs/msg/Twist` | `ignition.msgs.Twist` | Forward and angular velocity commands to differential drive joints. |
| `/{rid}/odom` | `Gz -> ROS` | `nav_msgs/msg/Odometry` | `ignition.msgs.Odometry` | Wheel odometry state (pose estimate, linear and angular velocity). |
| `/{rid}/scan` | `Gz -> ROS` | `sensor_msgs/msg/LaserScan` | `ignition.msgs.LaserScan` | Planar LiDAR range returns for obstacle mapping and frontier discovery. |
| `/{rid}/camera/image_raw` | `Gz -> ROS` | `sensor_msgs/msg/Image` | `ignition.msgs.Image` | Forward-facing camera feed for vehicle detection and plate OCR. |
Sources: [run_gazebo_gui.sh:127-147](), [ros2_ws/src/swarm_gazebo/launch/swarm_simulation.launch.py:29-80]()
### 1.3 Master Execution Launchers
The repository supplies two entry points for 3D simulation:
1. **Host-Level Master Bash Script (`run_gazebo_gui.sh`)**:
Verifies Docker daemon health, builds the container image `swarm_gazebo:humble` if missing, compiles the ROS 2 workspace using `colcon build --symlink-install`, starts Xvfb/Fluxbox/VNC inside Docker, launches `ign gazebo`, opens `http://localhost:6080/vnc.html` automatically in the host browser, and boots the Python orchestrator.
```bash
./run_gazebo_gui.sh
```
2. **Native ROS 2 Launch Description (`swarm_simulation.launch.py`)**:
Used when running directly within the sourced ROS 2 environment. Coordinates `gazebo_world.launch.py`, `slam_bridge_node`, `perception_bridge_node`, `coordinator_node`, `mission_monitor_node`, and four individual `robot_controller_node` instances.
Sources: [run_gazebo_gui.sh:48-159](), [ros2_ws/src/swarm_gazebo/launch/swarm_simulation.launch.py:16-81]()
---
## 2. ROS 2 Bridge Nodes & Controller Implementation
The integration cleanly wraps the core SwarmX Python library within ROS 2 nodes, ensuring zero coupling between algorithm logic and the underlying middleware.
```mermaid
flowchart TD
subgraph GazeboSimulation["Gazebo Fortress 3D Physics"]
GZ_Clock["/clock"]
GZ_Odom["/{rid}/odom"]
GZ_Scan["/{rid}/scan"]
GZ_Cam["/{rid}/camera/image_raw"]
GZ_Cmd["/{rid}/cmd_vel"]
end
subgraph BridgeLayer["ROS 2 Bridge Layer"]
Bridge["ros_gz_bridge parameter_bridge"]
SLAMBridge["GazeboSLAMBridge (ISLAMProvider)"]
PercepBridge["GazeboPerceptionBridge"]
end
subgraph SwarmCore["Swarm Intelligence Core"]
CoordNode["GazeboSwarmCoordinatorNode"]
ACO["ACOEngine & Boltzmann Softmax"]
PheroMap["Tri-Layer Pheromone Field"]
FaultMgr["Dynamic Failure Recovery"]
end
subgraph MotionControl["Motion Control Layer"]
Controllers["GazeboRobotController (/robot_*/cmd_vel)"]
end
GZ_Clock --> Bridge
GZ_Odom --> Bridge
GZ_Scan --> Bridge
GZ_Cam --> Bridge
Bridge --> GZ_Cmd
Bridge --> SLAMBridge
Bridge --> PercepBridge
SLAMBridge -- "SLAMState (Pose + Occupancy + Frontiers)" --> CoordNode
PercepBridge -- "/swarm/perception_events" --> CoordNode
CoordNode --> ACO
CoordNode --> PheroMap
CoordNode --> FaultMgr
CoordNode -- "Assigned Waypoints" --> Controllers
Controllers --> Bridge
```
### 2.1 SLAM Bridge (`GazeboSLAMBridge`)
Subscribes to `/robot_*/odom` and `/robot_*/scan`. It maintains an internal `UrbanEnvironment` representation and transforms local robot odometry into the global coordinate frame using the pre-configured spawn poses:
- `robot_1`: $(6.0, 6.0, 0.0)$
- `robot_2`: $(6.0, 44.0, -\pi/2)$
- `robot_3`: $(44.0, 6.0, \pi/2)$
- `robot_4`: $(44.0, 44.0, \pi)$
It implements the `ISLAMProvider` contract, allowing `GazeboSwarmCoordinatorNode` to poll `get_latest_state(robot_id)` to extract `FrontierCluster` instances and `OccupancyGrid2D` state.
```python
# ros2_ws/src/swarm_gazebo/swarm_gazebo/slam_bridge.py
class GazeboSLAMBridge(ISLAMProvider, Node):
def __init__(self, robot_ids: List[str], map_width_m: float = 50.0, ...):
Node.__init__(self, "gazebo_slam_bridge")
self.spawn_poses = spawn_poses or DEFAULT_SPAWN_POSES
# Tracks local occupancy grids and frontier clusters
```
Sources: [ros2_ws/src/swarm_gazebo/swarm_gazebo/slam_bridge.py:31-110]()
### 2.2 Closed-Loop Proportional Controller (`GazeboRobotController`)
Translates discrete task centroids into continuous physical motion. The controller runs at $20\,\text{Hz}$ ($\Delta t = 0.05\,\text{s}$), transforming local odometry to world coordinates, computing heading error $\Delta \theta = \text{atan2}(y_{\text{goal}} - y, x_{\text{goal}} - x) - \theta$, and applying proportional gains ($K_p^{\text{linear}} = 0.8, K_p^{\text{angular}} = 2.0$):
$$v = \min(v_{\max}, K_p^{\text{linear}} \cdot d) \cdot \max(0.0, \cos(\Delta\theta))$$
$$\omega = \text{clamp}(K_p^{\text{angular}} \cdot \Delta\theta, -\omega_{\max}, \omega_{\max})$$
When the Euclidean distance to the waypoint falls below the goal tolerance ($0.8\,\text{m}$), the controller publishes task completion on `/{robot_id}/task_status` and idles until the next ACO assignment.
Sources: [ros2_ws/src/swarm_gazebo/swarm_gazebo/robot_controller.py:41-120]()
---
## 3. Monte Carlo Benchmarking & Baseline Comparisons
To scientifically substantiate claims regarding swarm coordination, SwarmX includes a Monte Carlo benchmarking framework in `scripts/run_evaluation.py` and scenario suites in `evaluation/ai_benchmarks.py`.
### 3.1 Comparative Strategies
The benchmark compares three distinct task allocation paradigms:
1. **Proposed ACO (`ACOAllocationStrategy`)**: Evaluates frontier clusters using multi-objective stigmergic scoring (information gain, land-use detection prior, distance cost, congestion penalties, avoidance fields) and assigns tasks via scale-invariant Boltzmann Softmax ($T = 0.5$).
2. **Nearest Frontier (`NearestFrontierStrategy`)**: The classical Yamauchi greedy baseline. Selects the closest unassigned frontier purely by Euclidean distance, ignoring peer congestion, search history, and vehicle priors.
3. **Random Walk (`RandomAllocationStrategy`)**: Stochastic uncoordinated baseline that picks randomly among pending frontiers.
Sources: [evaluation/baselines.py:20-46](), [swarm/task_allocation/allocation_strategy.py:38-100]()
### 3.2 The 15 Comparative Performance Metrics
Telemetry is accumulated per tick across all robots by `MetricsCollector` and synthesized into a 15-metric performance matrix:
| # | Metric Name | Unit / Type | Description & Calculation |
| :---: | :--- | :---: | :--- |
| **1** | **Target Discovery Rate** | `%` | Percentage of trials where target `DXB-88392` was verified ($C_{\text{match}} \ge 0.80$). |
| **2** | **Mean Time-to-Detection (TTD)** | `s` | Average elapsed simulation time until confirmed target acquisition. |
| **3** | **Median Time-to-Detection** | `s` | Median TTD across successful runs, robust to outlier runs. |
| **4** | **Final Area Coverage** | `%` | Explored free grid cells as a fraction of total navigable urban area. |
| **5** | **Exploration Rate** | `m²/s` | Area mapped per unit time ($\Delta \text{Area} / \Delta t$). |
| **6** | **Trajectory Overlap Ratio** | `%` | Proportion of cell visits beyond the initial visit: $\frac{\sum \max(0, N_v - 1)}{\sum N_v}$. |
| **7** | **Total Distance Traveled** | `m` | Cumulative odometric distance traveled by all fleet agents. |
| **8** | **Battery Energy Consumed** | Units | Normalized propulsion and sensor battery consumption. |
| **9** | **Tasks Completed** | Count | Total frontier exploration tasks navigated and closed. |
| **10** | **Dynamic Reassignments** | Count | Task reallocations triggered by proximity shifts or failure events. |
| **11** | **Failure Recovery Rate** | `%` | Percentage of injected hardware faults successfully recovered. |
| **12** | **Failure Recovery Latency** | `s` | Elapsed time between robot failure injection and peer orphan task reassignment. |
| **13** | **Candidate Detections** | Count | Vehicles sighted within camera FOV requiring plate inspection. |
| **14** | **False Positives Rejected** | Count | Civilian decoy vehicles evaluated by OCR and discarded. |
| **15** | **Target Confirmations** | Count | Verified target sightings passing temporal and appearance thresholds. |
Sources: [scripts/run_evaluation.py:229-309](), [evaluation/metrics_collector.py:21-54]()
### 3.3 4-Variant ACO Ablation Studies
To isolate the exact contribution of each stigmergic term in the multi-objective utility function:
$$\text{Score}(t_i, r_j) = \frac{[\tau_i]^\alpha \cdot [\eta_i]^\beta \cdot [P_i]^\gamma \cdot [G_i]^\delta}{[C(r_j, t_i)]^\varepsilon \cdot [\Omega_i]^\zeta \cdot [K_i]^\theta}$$
The test runner supports an `--ablation` mode executing four configurations:
| Ablation Variant | Positive Pheromone ($\alpha$) | Target Prior ($\gamma$) | Scientific Objective |
| :--- | :---: | :---: | :--- |
| `ACO_BASELINE` | $\alpha = 0.0$ | $\gamma = 0.0$ | Strips all domain heuristics and target recruitment; acts as pure spatial dispersion. |
| `ACO_PROB_ONLY` | $\alpha = 0.0$ | $\gamma = 1.8$ | Uses land-use priors (e.g. parking zones) without dynamic stigmergic recruitment. |
| `ACO_PHERO_ONLY` | $\alpha = 1.2$ | $\gamma = 0.0$ | Relies on dynamic visual cue pheromone deposits without land-use priors. |
| `ACO_FULL` | $\alpha = 1.2$ | $\gamma = 1.8$ | Full proposed system combining prior maps with stigmergic peer recruitment. |
Sources: [scripts/run_evaluation.py:583-608](), [docs/thesis_chapter_swarm.md:120-134]()
### 3.4 AI Benchmark Scenarios A through H
Deterministic perception and coordination evaluations are encoded in `evaluation/ai_benchmarks.py` across eight scenarios:
| Scenario | Code Name | Primary Challenge Tested |
| :--- | :--- | :--- |
| **A** | `scenario_a` | Reachable target in open central plaza. |
| **B** | `scenario_b` | Partially occluded target behind building corner. |
| **C** | `scenario_c` | Civilian decoy vehicles present (`AD-10294` white SUV, `SHJ-44910` red sedan). |
| **D** | `scenario_d` | Poor/noisy OCR with simulated character substitutions ($B \leftrightarrow 8, O \leftrightarrow 0, Z \leftrightarrow 2$). |
| **E** | `scenario_e` | Catastrophic robot failure injected on `robot_1` at tick 20. |
| **F** | `scenario_f` | Cooperative multi-robot encounter requiring two-vehicle convergence. |
| **G** | `scenario_g` | Distant target located in the far corner of the search boundary. |
| **H** | `scenario_h` | Low-prior zone target requiring Bayesian exploration propagation. |
Sources: [evaluation/ai_benchmarks.py:184-194](), [docs/thesis_chapter_swarm.md:250-280]()
---
## 4. Architectural Extension Hooks & Developer Contracts
SwarmX was engineered with explicit abstract base classes (ABCs) and dataclass schemas to enable independent development across university cohorts or modular production upgrades.
```mermaid
classDiagram
class ISLAMProvider {
<>
+get_latest_state(robot_id: str) SLAMState*
+is_healthy(robot_id: str) bool*
}
class IPerceptionProvider {
<>
+detect_vehicles(robot_id: str, camera_frame_id: str) VehicleDetectionEvent*
+recognize_license_plate(detection: VehicleDetectionEvent) PlateDetectionEvent*
+match_watchlist(plate_event: PlateDetectionEvent) MatchEvent*
}
class ITaskAllocationStrategy {
<>
+allocate_task(robot_id, robot_pose, available_tasks, swarm_state, pheromone_map, target_probability_map, search_memory, battery_level) SearchTask*
}
class IServerReporter {
<>
+report_match(report: DetectionReport) bool*
+send_telemetry(telemetry: MissionReport) bool*
}
class GazeboSLAMBridge {
+get_latest_state(robot_id: str) SLAMState
+is_healthy(robot_id: str) bool
}
class ACOAllocationStrategy {
+allocate_task(...) SearchTask
}
ISLAMProvider <|.. GazeboSLAMBridge : implements
ITaskAllocationStrategy <|.. ACOAllocationStrategy : implements
```
### 4.1 Extending SLAM: The `ISLAMProvider` Contract
Developers integrating custom cartographer, RTAB-Map, or hardware LiDAR systems only need to implement two methods in `interfaces/slam_interface.py`:
- `get_latest_state(robot_id: str) -> SLAMState`: Returns current `Pose2D`, linear/angular velocities, `OccupancyGrid2D`, and extracted `List[FrontierCluster]`.
- `is_healthy(robot_id: str) -> bool`: Flags whether tracking is valid or if the robot is drifting/lost.
Sources: [interfaces/slam_interface.py:69-100]()
### 4.2 Extending Perception: The `IPerceptionProvider` Contract
Located in `interfaces/perception_interface.py`, this contract decouples computer vision from swarm decision making:
- `detect_vehicles(robot_id, camera_frame_id) -> Optional[VehicleDetectionEvent]`: YOLO/SSD bounding box detector.
- `recognize_license_plate(detection) -> Optional[PlateDetectionEvent]`: Crop OCR model returning recognized text and character confidence scores.
- `match_watchlist(plate_event) -> Optional[MatchEvent]`: Fuzzy matching engine (Levenshtein distance) comparing extracted text against stolen vehicle registries.
Sources: [interfaces/perception_interface.py:86-103]()
### 4.3 Implementing Custom MRTA Strategies
To benchmark a new task allocation algorithm (e.g. Market-based Auctioning, Consensus-Based Bundle Algorithm / CBBA, or Deep Reinforcement Learning), subclass `ITaskAllocationStrategy` in `swarm/task_allocation/allocation_strategy.py` and register it in `evaluation/baselines.py`:
```python
# swarm/task_allocation/allocation_strategy.py
class ITaskAllocationStrategy(ABC):
@abstractmethod
def allocate_task(
self,
robot_id: str,
robot_pose: Pose2D,
available_tasks: List[SearchTask],
swarm_state: Optional[SwarmState] = None,
pheromone_map: Optional[PheromoneMap] = None,
target_probability_map: Optional[Any] = None,
search_memory: Optional[Any] = None,
battery_level: float = 1.0,
) -> Optional[SearchTask]:
"""Selects a candidate task for the specified robot."""
pass
```
Sources: [swarm/task_allocation/allocation_strategy.py:19-36](), [evaluation/baselines.py:27-46]()
### 4.4 External Dashboard Reporting: `IServerReporter`
To stream telemetry, live occupancy maps, or incident alerts to cloud dashboards, implement `IServerReporter` in `interfaces/server_interface.py`:
- `report_match(report: DetectionReport) -> bool`: Dispatches target alerts containing full multi-stage confidence breakdowns and image evidence URLs.
- `send_telemetry(telemetry: MissionReport) -> bool`: Emits periodic swarm health, active agent counts, and cumulative area coverage metrics.
Sources: [interfaces/server_interface.py:31-43]()
### 4.5 Tuning ACO Weights via `ACOConfig`
Hyperparameters can be updated programmatically or loaded dynamically via YAML using `ACOConfig.from_yaml(path)`:
- Exponents: $\alpha$ (pheromone exploitation), $\beta$ (information gain), $\gamma$ (target prior), $\delta$ (urgency), $\varepsilon$ (distance penalty), $\zeta$ (congestion penalty), $\theta$ (avoidance repulsion).
- Stochastic parameters: `temperature` (Boltzmann Softmax dispersion), `epsilon_greedy`, and `deterministic`.
Sources: [swarm/aco/aco_config.py:10-84]()
---
## 5. Practical Operational Workflows & CLI Recipes
```
+---------------------------------------------------------------------------------------+
| OPERATIONAL RECIPES |
| |
| 1. Interactive 3D Gazebo GUI (Docker + noVNC Web Display): |
| $ ./run_gazebo_gui.sh |
| Browser opens: http://localhost:6080/vnc.html?autoconnect=true&resize=remote |
| |
| 2. Headless Monte Carlo Benchmark (3 Trials, Easy Target): |
| $ python3 scripts/run_evaluation.py --scenario easy_target --trials 3 |
| |
| 3. ACO 4-Variant Ablation Study: |
| $ python3 scripts/run_evaluation.py --scenario hidden_target --ablation |
| |
| 4. Deterministic AI Benchmark Suite (Scenarios A through H): |
| $ python3 scripts/evaluate_ai_benchmarks.py |
+---------------------------------------------------------------------------------------+
```
### Running Monte Carlo Evaluations
To execute comparative trials between ACO, Nearest Frontier, and Random Walk:
```bash
python3 scripts/run_evaluation.py \
--scenario easy_target \
--trials 5 \
--steps 120 \
--robots 4 \
--output-dir output/evaluation_run
```
The runner generates:
- `output/evaluation_run/benchmark_results.json`: Complete raw JSON telemetry for every trial.
- `output/evaluation_run/benchmark_report.md`: Markdown summary table with means, standard deviations, and recovery latencies.
- `output/evaluation_run/benchmark_coverage_comparison.svg`: Pure-Python vector graphic showing exploration curves over time.
- `output/evaluation_run/coverage_comparison.png` & `metrics_comparison.png`: Publication-grade PNG charts when Matplotlib is available.
Sources: [scripts/run_evaluation.py:62-83](), [scripts/run_evaluation.py:634-658]()
### Running ACO Ablation Experiments
To test the individual effects of pheromone recruitment and prior probability weighting:
```bash
python3 scripts/run_evaluation.py --ablation --trials 5 --scenario hidden_target
```
Sources: [scripts/run_evaluation.py:583-608]()
---
## 6. Compound Knowledge Profile & Architecture Governance
This documentation and architectural review synthesizes learnings using Every's Compound Engineering methodology (adapted for provider-neutral Blueprint, Page Shape, and QA Review workflows):
- **Provider Neutrality & BYOC/BYOK**: All interfaces, bridges, and benchmark runners operate locally and independently of hosted cloud vendors or proprietary model APIs. Perception interfaces consume local vision models (or synthetic mocks), and ROS 2 middleware communicates over standard DDS/Ignition transport protocols.
- **Portability Across Runtimes**: Whether evaluated in discrete Python 2D simulation (`simulation/environment.py`), continuous 3D Gazebo Fortress (`ros2_ws/src/swarm_gazebo`), or physical embedded robotic platforms (such as TurtleBot3 or custom differential drive AGVs), the coordination layer depends solely on abstract interface definitions (`ISLAMProvider`, `IPerceptionProvider`, `ITaskAllocationStrategy`).
---
## Summary
Moving beyond the initial 30 minutes in the SwarmX codebase transitions developers from discrete 2D simulation to containerized 3D physical verification and rigorous statistical benchmarking. By leveraging `run_gazebo_gui.sh` and Docker, engineers can inspect continuous robot navigation, raycasted line-of-sight occlusion, and camera perception in Gazebo Fortress without host dependency pollution. Concurrently, the Monte Carlo benchmarking harness (`run_evaluation.py`) and clean interface abstractions (`ISLAMProvider`, `IPerceptionProvider`, `ITaskAllocationStrategy`) empower researchers to extend perception, swap SLAM algorithms, test novel MRTA schedulers, and publish empirical evaluations across standardized urban scenarios.
---