63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
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] = {}
|
|
|
|
async def start_race(self, room: Room) -> asyncio.Task:
|
|
"""Start race progression loop."""
|
|
task = asyncio.create_task(self._race_loop(room))
|
|
self.active_tasks[room.scope] = task
|
|
return task
|
|
|
|
async def _race_loop(self, room: Room):
|
|
"""Main race progression loop."""
|
|
room.state = RoomState.RUNNING
|
|
|
|
while room.state == RoomState.RUNNING:
|
|
await asyncio.sleep(self.config.RACE_TICK_INTERVAL)
|
|
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
|
|
]
|
|
|
|
if finished_horses:
|
|
champion = self._determine_champion(finished_horses)
|
|
room.champion_name = champion.name
|
|
room.state = RoomState.FINISHED
|
|
break
|
|
|
|
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 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]
|