from collections import deque from enum import Enum, auto from dataclasses import dataclass, field from typing import Optional class ActorState(Enum): IDLE = auto() MOVING = auto() EXECUTING = auto() BLOCKED = auto() COMPLETE = auto() @dataclass class Order: action: str params: dict step_issued: int class BaseActor: def __init__(self, actor_id: str, start_node: int): self.id = actor_id self.position = start_node self.state = ActorState.IDLE self.current_order: Optional[Order] = None self.ticks_remaining = 0 self.last_result = None self.report_queue = [] self.order_queue = deque() self.max_fuel = 50 self.max_supplies = 5 self.fuel = self.max_fuel self.supplies = self.max_supplies self.COMMON_ACTIONS = {"RESUPPLY", "TRANSFER"} def get_snapshot(self): import copy return { "position": self.position, "state": self.state, "current_order": copy.deepcopy(self.current_order), "ticks_remaining": self.ticks_remaining, "last_result": self.last_result, "report_queue": copy.deepcopy(self.report_queue), "order_queue": list(copy.deepcopy(self.order_queue)), "fuel": self.fuel, "supplies": self.supplies } def restore_snapshot(self, snapshot): import copy from collections import deque self.position = snapshot["position"] self.state = snapshot["state"] self.current_order = copy.deepcopy(snapshot["current_order"]) self.ticks_remaining = snapshot["ticks_remaining"] self.last_result = snapshot["last_result"] self.report_queue = copy.deepcopy(snapshot["report_queue"]) self.order_queue = deque(copy.deepcopy(snapshot["order_queue"])) self.fuel = snapshot["fuel"] self.supplies = snapshot["supplies"] # ── FIXED: Added queue clearing to prevent Command Bloat ──────────────── def receive_orders(self, orders: list[dict], step: int): # IMPORTANT: Clearing the queue ensures the unit follows the LATEST # strategy from the ICS and doesn't get stuck doing old, redundant tasks. self.order_queue.clear() for o in orders: # Validate action against subclass VALID_ACTIONS (prevents ENG_1 scouting) valid_set = getattr(self, "VALID_ACTIONS", set()) if o["action"] not in valid_set and o["action"] not in self.COMMON_ACTIONS: self._post_report("REJECTED", f"Unknown action {o['action']}") continue # Optimization: Don't queue a MOVE if we are already at the target if o["action"] == "MOVE" and o.get("target_node") == self.position: continue self.order_queue.append(Order(o["action"], o, step)) def tick(self, world): # If IDLE and we have orders, start the next one immediately if self.state == ActorState.IDLE: if self.order_queue: self._next_order(world) else: return if self.state == ActorState.MOVING: # Check fuel before moving if self.fuel > 0: # Store position to see if we actually moved (Bug 6 fix support) prev_pos = self.position self._tick_move(world) # Only deduct fuel if we actually changed nodes if self.position != prev_pos: self.fuel -= 1 else: self.state = ActorState.IDLE self.current_order = None self._post_report("FAILED", "Out of fuel. Need RESUPPLY or TRANSFER.") self.order_queue.clear() elif self.state == ActorState.EXECUTING: if not self._tick_execute_common(world): self._tick_execute(world) def _needs_travel(self) -> bool: if not self.current_order or self.current_order.action != "MOVE": return False target = self.current_order.params.get("target_node") return target is not None and target != self.position def _post_report(self, result: str, detail=""): self.report_queue.append( { "actor": self.id, "position": self.position, "result": result, "action": self.current_order.action if self.current_order else None, "resources": f"F:{self.fuel} S:{self.supplies}", "detail": detail, } ) @staticmethod def _hazard_mult(node) -> float: m = 1.0 if node.temperature > 42: m += (node.temperature - 42) * 0.02 if node.water_level > 0: m += node.water_level * 0.15 if node.is_collapsed: m += 0.5 return round(m, 2) def _next_order(self, world): if self.order_queue: self.current_order = self.order_queue.popleft() # Logic to skip MOVE orders if we are already there (Instant finish) if self.current_order.action == "MOVE" and not self._needs_travel(): self._finish_order(world) return dur = self._compute_duration_common(world) if dur is None: dur = self._compute_duration(world) self.ticks_remaining = dur # Transition to correct state if self._needs_travel(): self.state = ActorState.MOVING else: self.state = ActorState.EXECUTING else: self.state = ActorState.IDLE def _compute_duration_common(self, world) -> Optional[int]: if self.current_order.action == "RESUPPLY": return 1 if self.current_order.action == "TRANSFER": return 1 return None def _tick_execute_common(self, world) -> bool: o = self.current_order if o.action == "RESUPPLY": self.ticks_remaining -= 1 if self.ticks_remaining <= 0: if self.position == 0: self.fuel = 50 self.supplies = 5 self._post_report("SUCCESS", "Resupplied to max capacity") else: self._post_report("FAILED", "Must be at Node 0 to resupply") self._finish_order(world) return True elif o.action == "TRANSFER": self.ticks_remaining -= 1 if self.ticks_remaining <= 0: target_actor_id = o.params.get("target_actor") fuel_amt = o.params.get("fuel", 0) supplies_amt = o.params.get("supplies", 0) if not hasattr(world, "actors") or target_actor_id not in world.actors: self._post_report( "FAILED", f"Target actor {target_actor_id} not found" ) else: target_actor = world.actors[target_actor_id] if target_actor.position != self.position: self._post_report( "FAILED", "Target actor not at same node. CLEARING QUEUE." ) self.order_queue.clear() elif self.fuel < fuel_amt or self.supplies < supplies_amt: self._post_report( "FAILED", "Insufficient resources to transfer" ) else: self.fuel -= fuel_amt self.supplies -= supplies_amt # Cap at max capacity target_actor.fuel = min( target_actor.max_fuel, target_actor.fuel + fuel_amt ) target_actor.supplies = min( target_actor.max_supplies, target_actor.supplies + supplies_amt, ) self._post_report( "SUCCESS", f"Transferred F:{fuel_amt} S:{supplies_amt} to {target_actor_id}", ) self._finish_order(world) return True return False def _finish_order(self, world): self.state = ActorState.IDLE self.current_order = None # Recursively check for next order to allow "instant" skips of redundant moves if self.order_queue: self._next_order(world) def _compute_duration(self, world) -> int: raise NotImplementedError def _tick_move(self, world): raise NotImplementedError def _tick_execute(self, world): raise NotImplementedError