# Project export: OpenSwarm This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: TreeHacks 2026 - Tagline: A Framework for Centralized Swarm Orchestration - Devpost: https://devpost.com/software/openswarm - GitHub: https://github.com/k-kochhar/OpenHive - Video: https://www.youtube.com/embed/t7FWFP3fg4s?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 4 GitHub contributor(s) — Jay Katyan (15 commits), NSP909 (5 commits), Kshitij Kochhar (1 commits), aditya (1 commits) ## Devpost submission (written by the team) ### Inspiration As physical AI systems scale, coordinating multiple agents in dynamic environments remains brittle and hardware dependent. Most multi-robot systems rely on distributed intelligence, where each robot must reason independently. This increases cost, complexity, and limits scalability. We wanted to explore a different model: what if intelligence lived in one centralized "queen" that observes the environment from above, reasons globally, and orchestrates simple, stateless worker robots? OpenSwarm was built to be that orchestration layer. ### What it does OpenSwarm is a reusable, extensible framework for centralized multi-agent physical coordination. A central intelligence system, the queen, maintains a global world model from an overhead perspective. It receives high-level user commands (natural language), decomposes them into structured subtasks using large language models, assigns roles optimally, and continuously replans when the environment changes. Worker robots are intentionally simple and expendable. They do not reason. They execute movement and positioning commands from the queen via a standardized actions interface. Extensible World Architecture OpenSwarm uses a modular "world" system where each environment is self-contained with its own: init.md - defines the world structure, agents, and capabilities actions.py - implements available actions (move_to, collect, extinguish, etc.) The queen automatically adapts to each world by reading its init document and available actions, and creates it's own interpretation of the world, the state and the goals, making it trivially easy to add new environments without modifying core orchestration logic. Because intelligence is centralized and hardware is decoupled via the actions interface, this framework extends far beyond ground robots. For example, in natural disaster scenarios, overhead drones can act as the queen's perception layer, mapping debris fields, fire spread, or structural damage in real time. The queen can then coordinate fleets of low-cost, expendable ground machines to clear paths, deliver supplies, or stabilize hazardous zones. If an obstruction arises, the queen detects it and reallocates instantly without compromising the mission. ### How we built it Architecture Layers 1. Perception Layer An overhead view (camera or simulation) tracks robot and object coordinates and maintains world state. Each world implements _get_state() to provide current positions, obstacles, and task-relevant information. 2. Orchestration Layer (Queen) The queen runs a continuous poll loop that: Reads the world initialization document and generates a structured world model using an LLM Monitors a task queue (populated via user input or autonomous triggers) For each task, sends the current world state + available actions to an LLM Receives structured function calls with optimal bot assignments Executes calls in parallel when possible (different bots) or sequentially (same bot) Continuously updates world state and triggers replanning when needed The queen supports multiple LLM backends (Claude, GPT, Gemini) via a modular interface, with vision support for screenshot-based reasoning. 3. Execution Layer Worker robots receive structured commands via a standardized actions interface. Each action function: Takes explicit parameters (positions, bot IDs, etc.) Writes commands to an IPC file or directly to hardware Returns immediately (non-blocking) For physical robots, commands flow through: WebSocket server (multi-device coordination) ESP32 microcontrollers with PWM motor control ArUco marker tracking for real-time localization 4. Neural Pathfinding with Modal OpenSwarm uses a neural A* pathfinder with learned heuristics for efficient collision-free navigation: Training: Trained a neural network (HeuristicNetV2) on thousands of pathfinding scenarios to predict optimal distance-to-goal heuristics Inference: Model deployed to Modal for large-scale parallel inference Scaling: Supports up to 50 concurrent pathfinding requests via Modal's parallel execution The Modal deployment enables: Zero cold starts (min_containers=1) Auto-scaling under load (max_containers=10) Batched pathfinding for large swarms (50 bots in mimic_world) Collision Avoidance For large swarms, paths are grouped into collision-free waves: All bot paths are computed in parallel via Modal Paths sharing grid cells are separated into sequential waves Each wave moves simultaneously Next wave starts after longest path in previous wave completes This enables smooth, collision-free movement of 50 bots without complex multi-agent planning. ### Challenges we ran into 1. Power Management One of the key challenges we faced was balancing power efficiency with performance while keeping robots compact. We initially planned to use a single 18650 battery with a voltage booster. However, this setup could not provide enough current to reliably drive both motors simultaneously. To overcome this, we simplified the power architecture by removing the booster and implementing PWM-based motor control. This allowed us to efficiently manage power delivery while maintaining reliable performance and meeting size constraints. 2. Large-Scale Pathfinding Computing collision-free paths for 50 bots in real-time was initially too slow with traditional A*. We solved this by: Training a neural heuristic to accelerate A* (reduces search space by ~60%) Deploying to Modal for GPU acceleration Implementing parallel path computation with wave-based collision avoidance 3. Hardware Abstraction Making the same actions interface work for both simulation and physical robots required careful design: Simulated bots use file-based IPC with instant state updates Physical bots use WebSocket communication with real-time camera tracking Both implement the same move_to(target, bot_id) interface Main orchestration code remains identical across worlds ### Accomplishments we're proud of Extensible architecture - Adding a new world requires only 3 files (init.md, actions.py, simulation.py). The queen adapts automatically. Extensible architecture - Adding a new world requires only 3 files (init.md, actions.py, simulation.py). The queen adapts automatically. Large-scale coordination - Demonstrated real-time coordination of 50 bots with collision-free movement using Modal for GPU-accelerated pathfinding. Large-scale coordination - Demonstrated real-time coordination of 50 bots with collision-free movement using Modal for GPU-accelerated pathfinding. Hardware abstraction - Same orchestration code controls both simulated and physical robots via a standardized actions interface. Hardware abstraction - Same orchestration code controls both simulated and physical robots via a standardized actions interface. Multi-modal reasoning - Queen processes natural language commands, visual input (screenshots), and structured world state to make decisions. Multi-modal reasoning - Queen processes natural language commands, visual input (screenshots), and structured world state to make decisions. Neural pathfinding - Trained and deployed a learned heuristic that accelerates A* by 3-5× over traditional manhattan distance heuristics. Neural pathfinding - Trained and deployed a learned heuristic that accelerates A* by 3-5× over traditional manhattan distance heuristics. ### What we learned Separating intelligence from execution dramatically simplifies scaling. When the queen maintains a unified world model, coordination becomes a systems problem rather than a robotics problem. Adding more bots doesn't increase complexity, it just increases parallelism. LLMs are surprisingly good at spatial reasoning when given visual context (screenshots) and structured state information. The queen can reason about optimal bot assignments, formation control, and priority allocation without hand-coded heuristics. Parallel/scalable pathfinding changes what's possible. Modal's deployment infrastructure let us scale from 2-bot demos to 50-bot swarms without rewriting pathfinding logic. The same neural model runs locally during development and on GPU in production. ### What's next In the long term, we see applications for OpenSwarm in warehouse logistics, disaster response, construction automation, satellite coordination, and more. We'd love to expand on OpenSwarm's feature set and continue to make its hardware more practical and accessible. ## README (from the GitHub repository) No README available. ## Detected evidence (automated analysis) Indexed codebase: 52 recognized source files, 308 KB. - Python (language) — detected in the code - Google Gemini (technology) — claimed on Devpost, not found in the code - PyTorch (technology) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (81 of 81) ``` .gitignore concept/main.py concept/queen.py concept/README.md concept/requirements.txt concept/robot.py concept/state_machine.py concept/utils.py concept/world.py hardware/design_stl/2nd_level_robot_chassis.stl hardware/design_stl/bare_chassis.stl hardware/design_stl/innerWheel.stl hardware/design_stl/robot_base_chassis_with_mg90s.stl hardware/design_stl/SG90-Servo-Arms.stl hardware/design_stl/tire.stl hardware/hardware_code/arucomakers.py hardware/hardware_code/camera_test.py hardware/hardware_code/main_bot_1/main_bot_1.ino hardware/hardware_code/main_bot_2/main_bot_2.ino hardware/hardware_code/main_bot_3/main_bot_3.ino hardware/hardware_code/main_bot/main_bot.ino hardware/hardware_code/main_control_bot_multi.py hardware/hardware_code/shape_detector.py hardware/hardware_code/utils/servo_serial_calib/servo_serial_calib.ino hardware/hardware_code/utils/servo_ws_calib_multi/servo_ws_calib_multi.ino hardware/hardware_code/utils/test_ws_servo_calib_multi.py hardware/robot_motor_config.drawio hive/files/mimic_commands.json hive/files/mimic_state.json hive/files/sim_commands.json hive/files/sim_state.json hive/files/state.json hive/files/state.md hive/files/tasks.json hive/files/world.md hive/fire_world/actions.py hive/fire_world/checkpoints/best_model.pt hive/fire_world/init.md hive/fire_world/model.py hive/fire_world/pathfinder.py hive/fire_world/simulation.py hive/llms/__init__.py hive/llms/cla.py hive/llms/gog.py hive/llms/oai.py hive/main.py hive/mimic_world/actions.py hive/mimic_world/hand_debug.py hive/mimic_world/hand_landmarker.task hive/mimic_world/init.md hive/mimic_world/modal_app.py hive/mimic_world/simulation.py hive/move_world/actions.py hive/move_world/checkpoints/best_model.pt hive/move_world/init.md hive/move_world/model.py hive/move_world/pathfinder.py hive/move_world/simulation.py hive/ohm.py hive/prompts.py hive/robot_world/actions.py hive/robot_world/init.md hive/robot_world/overlay.py requirements.txt robot/playground/ArUco_test/detect_markers.py robot/playground/communication/communication.py robot/playground/motion/motion_controller.py robot/playground/motion/path.json robot/playground/motion/robot.py robot/playground/motion/sim.py robot/src/active_bots.json robot/src/controller.py robot/src/markers.json robot/src/multi_robot.py robot/src/path.json robot/src/server.py robot/src/utils/__init__.py robot/src/utils/camera.py robot/src/utils/path_follower.py robot/src/utils/robot_client.py robot/src/utils/screen.py ``` ### Dependencies - concept/requirements.txt: pygame@>=2.5.0 - requirements.txt: numpy, opencv-contrib-python@==4.13.0.92, opencv-python@==4.13.0.92, pygame@==2.6.1, pyrealsense2-macosx, websockets ### Recent commits (newest first) - feat: final stuff - Merge branch 'main' of https://github.com/k-kochhar/OpenHive - Merge branch 'main' of https://github.com/k-kochhar/OpenHive - Hardware files - fieat: stuff - Fire simulation - [feat] concept simulation - feat: stiff - Updated requirements.txt - Multi-robot code - Some updates - Time-based rotations - Basic path following implemented - Restructured repository - Added example code for ArUco markers - Command parsing - Multi-robot simulation - Simple motion controller - VS Code ignores - Refactored simulation ## Key source files (fetched from GitHub, selected and truncated for size) ### hive/move_world/init.md ```markdown A 64x64 grid world. Each cell is either free or an obstacle. There are two bots (bot 0 and bot 1) that start at random positions on the grid. Scattered around the grid are gold coins on free cells. Each bot can do two things: - move_to(target_pos, bot_id) — move the specified bot to a target [row, col] position. Pathfinding through obstacles is handled automatically. - collect(bot_id) — pick up a coin at the specified bot's current position. The bot must be standing on a coin for this to work. To collect a coin: first move_to the coin's position with the right bot_id, then call collect(bot_id). Coin positions are included in the state (list of [row, col] positions). Bot positions are in the state as a list — bot_id is the index (0 or 1). Use both bots in parallel to collect coins faster. ``` ### hive/mimic_world/init.md ```markdown A simulated world with 50 bots on a 64x64 grid that mirror your hand's shape. The system captures the MacBook webcam every 5 seconds, detects hand landmarks via MediaPipe, maps the hand skeleton onto the grid, and dispatches bots to form the hand's outline. No LLM is involved in the hand tracking loop. Hand tracking starts automatically when the module loads. Bots are identified by index (0-49). Coordinates are (row, col) on a 64x64 grid. Available actions: - start_hand_tracking() — start webcam hand tracking (auto-started on load). - stop_hand_tracking() — stop webcam hand tracking. - form_shape(shape_name) — manually arrange bots into a predefined shape: "circle", "square", "triangle", "star", "grid". - move_bot(target_pos, bot_id) — move a single bot to [row, col]. - get_positions(bot_id) — get current position of a bot, or all bots if bot_id=None. How hand tracking works: 1. Webcam captures a frame every 5 seconds. 2. MediaPipe detects 21 hand landmarks. 3. Landmarks are mapped onto the 64x64 grid (mirrored, scaled, centered). 4. Points are interpolated along the hand skeleton to produce ~50 target positions. 5. Bots are assigned to targets (Hungarian algorithm) and dispatched in collision-free waves. 6. If no hand is visible, bots hold their current position. 7. If the hand hasn't moved significantly, no update is sent. ``` ### requirements.txt ``` numpy opencv-contrib-python==4.13.0.92 opencv-python==4.13.0.92 pygame==2.6.1 websockets pyrealsense2-macosx ``` ### concept/requirements.txt ``` pygame>=2.5.0 ``` ### concept/main.py ```python """ Main simulation loop with Pygame rendering and video export. """ import pygame import math import os from datetime import datetime from world import World from robot import Robot from queen import Queen from state_machine import SimulationStateMachine from utils import Vec2 # Constants WINDOW_WIDTH = 1200 WINDOW_HEIGHT = 800 MAP_WIDTH = 900 MAP_HEIGHT = 800 PANEL_WIDTH = 300 FPS = 60 # Colors COLOR_BG = (245, 245, 250) COLOR_GRID = (220, 220, 230) COLOR_BORDER = (80, 80, 80) COLOR_BLUE = (50, 120, 200) COLOR_GREEN = (80, 180, 100) COLOR_ORANGE = (230, 120, 50) COLOR_ITEM = (120, 120, 130) COLOR_BRAVO = (100, 200, 100) COLOR_FIRE = (255, 80, 20) COLOR_FIRE_GLOW = (255, 160, 40) COLOR_PANEL_BG = (25, 25, 30) COLOR_PANEL_TEXT = (200, 200, 200) COLOR_PANEL_HEADER = (100, 200, 255) class Simulation: """Main simulation class.""" def __init__(self, export_frames=False): pygame.init() self.screen = pygame.display.set_mode((WINDOW_WIDTH, WINDOW_HEIGHT)) pygame.display.set_caption("Queen Central Intelligence - Multi-Robot Coordination") self.clock = pygame.time.Clock() self.running = True self.export_frames = export_frames self.frame_count = 0 # Simulation time self.sim_time = 0.0 # Initialize world self.world = World(MAP_WIDTH, MAP_HEIGHT) # Initialize robots at different spawn points self.robots = [ Robot("Alpha", "A", COLOR_BLUE, 200, 100), # Top of map Robot("Beta", "B", COLOR_GREEN, 700, 150), # Top right of map Robot("Charlie", "C", COLOR_ORANGE, 200, 700) # Bottom of map ] # Initialize Queen self.queen = Queen() # Initialize state machine self.state_machine = SimulationStateMachine(self.world, self.robots, self.queen) # Fonts self.font_large = pygame.font.Font(None, 32) self.font_medium = pygame.font.Font(None, 24) self.font_small = pygame.font.Font(None, 18) def run(self): """Main simulation loop.""" while self.running: dt = self.clock.tick(FPS) / 1000.0 self.sim_time += dt # Event handling self._handle_events() # Update simulation self._update(dt) # Render self._render() # Export frame if needed if self.export_frames: self._export_frame() # Auto-quit after mission complete + 2 seconds if self.state_machine.state == self.state_machine.STATE_COMPLETE: if self.state_machine.state_timer > 2.0: self.running = False pygame.quit() def _handle_events(self): """Handle pygame events.""" for event in pygame.event.get(): if event.type == pygame.QUIT: self.running = False elif event.type == pygame.KEYDOWN: if event.key == pygame.K_ESCAPE: self.running = False elif event.key == pygame.K_r: self._reset() def _reset(self): """Reset simulation.""" self.sim_time = 0.0 self.frame_count = 0 self.world = World(MAP_WIDTH, MAP_HEIGHT) self.robots = [ Robot("Alpha", "A", COLOR_BLUE, 200, 100), # Top of map Robot("Beta", "B", COLOR_GREEN, 700, 150), # Top right of map Robot("Charlie", "C", COLOR_ORANGE, 200, 700) # Bottom of map ] self.queen = Queen() self.state_machine = SimulationStateMachine(self.world, self.robots, self.queen) def _update(self, dt): """Update simulation state.""" # Update state machine self.state_machine.update(dt, self.sim_time) # Update robots for robot in self.robots: robot.update(dt, self.world, self.robots) def _render(self): """Render the simulation.""" self.screen.fill(COLOR_BG) # Draw map area self._draw_map() # Draw control panel self._draw_panel() pygame.display.flip() def _draw_map(self): """Draw the overhead map.""" # Grid grid_size = 50 for x in range(0, MAP_WIDTH, grid_size): pygame.draw.line(self.screen, COLOR_GRID, (x, 0), (x, MAP_HEIGHT), 1) for y in range(0, MAP_HEIGHT, grid_size): pygame.draw.line(self.screen, COLOR_GRID, (0, y), (MAP_WIDTH, y), 1) # Destination (Location Bravo) bravo_rect = pygame.Rect( int(self.world.bravo_pos.x - self.world.bravo_size / 2), int(self.world.bravo_pos.y - self.world.bravo_size / 2), self.world.bravo_size, self.world.bravo_size ) pygame.draw.rect(self.screen, COLOR_BRAVO, bravo_rect, 4) bravo_label = self.font_medium.render("BRAVO", True, (50, 150, 50)) label_rect = bravo_label.get_rect(center=(int(self.world.bravo_pos.x), int(self.world.bravo_pos.y - self.world.bravo_size / 2 - 20))) self.screen.blit(bravo_label, label_rect) # Item (Item A) item_rect = pygame.Rect( int(self.world.item_pos.x - self.world.item_size / 2), int(self.world.item_pos.y - self.world.item_size / 2), self.world.item_size, self.world.item_size ) pygame.draw.rect(self.screen, COLOR_ITEM, item_rect) pygame.draw.rect(self.screen, (0, 0, 0), item_rect, 3) item_label = self.font_medium.render("ITEM A", True, (255, 255, 255)) label_rect = item_label.get_rect(center=(int(self.world.item_pos.x), int(self.world.item_pos.y))) self.screen.blit(item_label, label_rect) # Fire if self.world.fire_active or self.world.fire_intensity > 0: self._draw_fire() # Robots for ro [truncated — 6101 more characters] ``` ### hive/main.py ```python """ OpenHive main loop. Usage: python main.py e.g. python main.py move_world Run the simulation first in a separate terminal: cd move_world && python simulation.py Then run main.py — it communicates with the simulation via files. Flow: 1. Reads init.md, runs the init prompt to generate a world document 2. Starts a poll loop (every 3s): a. Refreshes world state via detect_world_state (screenshot + bots → matrix) b. Saves state to files/state.json c. Checks files/tasks.json — if tasks exist, sends to LLM which returns function calls. We execute those calls on the world's actions module. 3. User can type commands at any time — they get added as tasks """ import json import time import sys import base64 import threading import importlib import inspect from pathlib import Path from ohm import chat from prompts import init_prompt, action_prompt, verify_prompt HIVE_DIR = Path(__file__).parent TASKS_FILE = HIVE_DIR / "files" / "tasks.json" WORLD_FILE = HIVE_DIR / "files" / "world.md" STATE_FILE = HIVE_DIR / "files" / "state.json" POLL_INTERVAL = 3 DEFAULT_MODEL = "claude-sonnet-4-5-20250929" # Will be set after loading the world's modules _actions_module = None def load_tasks(): if not TASKS_FILE.exists(): return [] text = TASKS_FILE.read_text().strip() if not text: return [] return json.loads(text) def save_tasks(tasks): TASKS_FILE.write_text(json.dumps(tasks, indent=2)) def add_task(task_text): tasks = load_tasks() tasks.append(task_text) save_tasks(tasks) def get_available_actions(): """Inspect the actions module and return a description of callable functions.""" actions = {} for name, fn in inspect.getmembers(_actions_module, inspect.isfunction): if name.startswith("_"): continue sig = inspect.signature(fn) doc = fn.__doc__ or "" params = [] for pname, param in sig.parameters.items(): p = {"name": pname, "type": str(param.annotation) if param.annotation != inspect.Parameter.empty else "any"} if param.default != inspect.Parameter.empty: p["default"] = repr(param.default) params.append(p) actions[name] = {"params": params, "doc": doc.strip()} return actions def refresh_state(): """Read full state from the world (fast, no LLM call).""" if not hasattr(_actions_module, "_get_state"): return None try: state = _actions_module._get_state() except Exception as e: print(f"[loop] State refresh skipped: {e}") return None if not state: return None STATE_FILE.write_text(json.dumps(state)) return state def run_init(world_dir: Path): """Read init.md, send through the init prompt, save world document.""" init_md = (world_dir / "init.md").read_text() actions_src = (world_dir / "actions.py").read_text() message = ( f"{init_prompt}\n\n" f"--- USER INIT DOCUMENT ---\n{init_md}\n\n" f"--- AVAILABLE ACTIONS (code) ---\n{actions_src}" ) print("[init] Generating world document...") world_doc = chat(DEFAULT_MODEL, message) WORLD_FILE.write_text(world_doc) print(f"[init] World document saved to {WORLD_FILE}") return world_doc def _wait_for_bot_idle(bot_id, timeout=30, poll=0.5): """Wait until the bot finishes its current movement or timeout.""" start = time.time() print(f"[exec] Bot {bot_id}: waiting for movement to finish...") time.sleep(1.0) while time.time() - start < timeout: try: state = _actions_module._get_state() if bot_id not in state.get("active_bots", []): print(f"[exec] Bot {bot_id}: movement finished") return True except Exception: pass time.sleep(poll) print(f"[exec] Bot {bot_id}: timeout waiting for movement") return False def _run_bot_sequence(bot_id, calls): """Execute a sequence of calls for a single bot, waiting between moves.""" for call in calls: fn_name = call.get("function") params = call.get("params", {}) fn = getattr(_actions_module, fn_name, None) if fn is None: print(f"[exec] Bot {bot_id}: unknown action {fn_name} — skipping") continue # Convert list params that should be tuples (positions) for k, v in params.items(): if isinstance(v, list) and len(v) == 2 and all(isinstance(x, (int, float)) for x in v): params[k] = tuple(v) # Replace "bot" with "bot_id" for the action function if "bot" in params: params["bot_id"] = params.pop("bot") # Only pass bot_id if the function accepts it sig = inspect.signature(fn) if "bot_id" not in sig.parameters and "bot_id" in params: params.pop("bot_id") print(f"[exec] Bot {bot_id}: {fn_name}({params})") try: result = fn(**params) print(f"[exec] Bot {bot_id}: {fn_name} → {str(result)[:200]}") except Exception as e: print(f"[exec] Bot {bot_id}: {fn_name} failed: {e}") continue # Wait for movement commands to finish before next call if fn_name in ("move_to", "push_and_exit"): _wait_for_bot_idle(bot_id) def execute_task(task, world_doc, state, available_actions): """ Send a task + state to the LLM. It returns function calls to execute. The LLM responds with JSON: { "calls": [ {"function": "move_to", "params": {"target_pos": [50, 60], "bot": 0}}, ... ], "new_tasks": ["optional follow-up tasks"] } Calls are grouped by bot. Each bot's calls run sequentially (waiting for moves to complete), but different bots run in parallel. """ actions_desc = json.dumps(available_actions, indent=2) state [truncated — 6251 more characters] ``` ### robot/src/server.py ```python import asyncio import websockets HOST = "0.0.0.0" PORT = 8765 devices = {} controllers = {} async def handler(ws): client_id = None is_robot = False try: async for msg in ws: msg = msg.strip() if msg.startswith("ID:"): client_id = msg[3:].strip() if not client_id: await ws.send("ERR:EMPTY_ID") continue # Determine if this is a robot (ESP) or controller if client_id.startswith("ESP"): devices[client_id] = ws is_robot = True print(f"Robot registered: {client_id}") else: controllers[client_id] = ws print(f"Controller registered: {client_id}") await ws.send("REGISTERED") if is_robot: await ws.send("S") continue # Handle targeted commands: TARGET:ESP1:F if not is_robot and msg.startswith("TARGET:"): parts = msg.split(":", 2) if len(parts) == 3: _, target_device, command = parts if target_device in devices: try: await devices[target_device].send(command) print(f"Forwarded {command} -> {target_device}") except: devices.pop(target_device, None) continue # If message is a command (F, B, L, R, S), forward to all robots if not is_robot and msg in ["F", "B", "L", "R", "S"]: dead = [] for device_id, device_ws in list(devices.items()): try: await device_ws.send(msg) print(f"Forwarded {msg} -> {device_id}") except: dead.append(device_id) for device_id in dead: devices.pop(device_id, None) else: # Log other messages if client_id: print(f"From {client_id}: {msg}") except websockets.ConnectionClosed: pass finally: if client_id: if is_robot and devices.get(client_id) is ws: devices.pop(client_id, None) print(f"Robot disconnected: {client_id}") elif not is_robot and controllers.get(client_id) is ws: controllers.pop(client_id, None) print(f"Controller disconnected: {client_id}") async def main(): print(f"WebSocket server listening on ws://{HOST}:{PORT}") async with websockets.serve(handler, HOST, PORT, ping_interval=20, ping_timeout=20): await asyncio.Future() if __name__ == "__main__": asyncio.run(main()) ``` ### hive/ohm.py ```python from llms import oai, gog, cla MODEL_MAP = { "gpt": oai, "gemini": gog, "claude": cla, } def chat(model: str, message: str, image_b64: str = None) -> str: for key, module in MODEL_MAP.items(): if key in model.lower(): kwargs = {"message": message, "model": model} if image_b64: kwargs["image_b64"] = image_b64 return module.chat(**kwargs) raise ValueError(f"Unknown model: {model}. Must contain one of: {', '.join(MODEL_MAP)}") ``` ### concept/queen.py ```python """ Queen central intelligence - logging and metrics. """ class Queen: """Central intelligence logging and metrics system.""" def __init__(self): self.logs = [] self.max_logs = 12 self.current_mode = "INITIALIZING" self.pushing_robots = 0 self.push_speed_percent = 0 self.fire_status = "NONE" def log(self, message, sim_time): """Add a timestamped log message.""" timestamp = f"T+{sim_time:06.2f}s" full_message = f"[{timestamp}] {message}" self.logs.append(full_message) if len(self.logs) > self.max_logs: self.logs.pop(0) def update_metrics(self, mode, pushing_count, speed_percent, fire_status): """Update display metrics.""" self.current_mode = mode self.pushing_robots = pushing_count self.push_speed_percent = speed_percent self.fire_status = fire_status ``` ### concept/world.py ```python """ World state management. """ from utils import Vec2 class World: """Contains all world entities and state.""" def __init__(self, map_width=900, map_height=800): self.map_width = map_width self.map_height = map_height # Item (70x70 square) self.item_size = 70 self.item_pos = Vec2(300, map_height // 2) # Destination (Location Bravo) - 110x110 region self.bravo_size = 110 self.bravo_pos = Vec2(map_width - 150, map_height // 2) # Fire (bottom of map) self.fire_pos = Vec2(500, 650) self.fire_active = False self.fire_intensity = 0.0 # 0 to 1 self.fire_radius = 40 # Suppression radius def is_item_at_bravo(self): """Check if item center has reached destination center.""" # Calculate distance between item center and bravo center dist = (self.item_pos - self.bravo_pos).length() # Item must reach within 10px of the destination center return dist < 10.0 ``` [41 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]