import asyncio import random from typing import Optional from .models import Room, RoomState, Horse, HorseState from .config import Config class RaceEngine: def __init__(self, config: Config): self.config = config self.active_tasks: dict[str, asyncio.Task] = {} def tick(self, room: Room): """Advance race by one tick. Returns list of finished horses.""" room.tick_count += 1 for horse in room.horses.values(): if horse.state == HorseState.RACING: distance = max(0, random.gauss(10, 3)) horse.position += distance finished_horses = [ h for h in room.horses.values() if h.position >= self.config.RACE_DISTANCE ] return finished_horses def determine_champion(self, horses: list[Horse]) -> Horse: """Determine champion from tied horses.""" if len(horses) == 1: return horses[0] while len(horses) > 1: distances = [max(0, random.gauss(10, 3)) for _ in horses] max_distance = max(distances) horses = [h for h, d in zip(horses, distances) if d == max_distance] return horses[0] def register_task(self, scope: str, task: asyncio.Task): """Register an active race task.""" self.active_tasks[scope] = task def stop_race(self, scope: str): """Stop race and cancel task.""" if scope in self.active_tasks: task = self.active_tasks[scope] if not task.done(): task.cancel() del self.active_tasks[scope] def format_progress(self, room: Room) -> str: """Format race progress as visual bar for each horse.""" distance = self.config.RACE_DISTANCE bar_width = 20 lines = [f"【第{room.tick_count}回合】"] sorted_horses = sorted(room.horses.values(), key=lambda h: h.index) for horse in sorted_horses: progress = min(horse.position / distance, 1.0) filled = int(progress * bar_width) bar = "█" * filled + "░" * (bar_width - filled) lines.append(f" {horse.index:02d}号 {horse.name} |{bar}| {horse.position:.1f}m") return "\n".join(lines)