Spaces:
Sleeping
Sleeping
| 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() | |
| 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, | |
| } | |
| ) | |
| 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 | |