botli_dataclasses.py

from asyncio import Task
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field, replace
from datetime import UTC, datetime, timedelta
from typing import Any, Literal

import chess
import chess.engine
from chess.polyglot import MemoryMappedReader

from enums import ChallengeColor, PerfType, Variant
from utils import find_variant, parse_time_control


@dataclass(kw_only=True)
class ApiChallengeResponse:
    challenge_id: str | None = None
    was_accepted: bool = False
    error: str | None = None
    was_declined: bool = False
    invalid_initial: bool = False
    invalid_increment: bool = False
    has_reached_rate_limit: bool = False
    wait_seconds: int | None = None
    has_timed_out: bool = False


@dataclass
class BookSettings:
    selection: Literal["weighted_random", "uniform_random", "best_move"] = "best_move"
    max_depth: int | None = None
    max_moves: int | None = None
    allow_repetitions: bool | None = None
    readers: dict[str, MemoryMappedReader] = field(default_factory=dict)


@dataclass
class Bot:
    username: str
    rating_diffs: dict[PerfType, int]

    def __eq__(self, value: object) -> bool:
        if isinstance(value, Bot):
            return value.username == self.username

        return NotImplemented

    def __hash__(self) -> int:
        return hash(self.username)


@dataclass
class Challenge:
    challenge_id: str
    opponent_username: str

    def __eq__(self, value: object) -> bool:
        if isinstance(value, Challenge):
            return value.challenge_id == self.challenge_id

        return NotImplemented

    def __hash__(self) -> int:
        return hash(self.challenge_id)


@dataclass
class ChallengeRequest:
    opponent_username: str
    initial_time: int
    increment: int
    rated: bool
    color: ChallengeColor
    variant: Variant
    timeout: int

    @classmethod
    def parse_from_command(cls, args: list[str], timeout: int) -> "ChallengeRequest":
        opponent_username = None
        initial_time = 60
        increment = 1
        color = ChallengeColor.RANDOM
        rated = True
        variant = Variant.STANDARD

        for arg in args:
            if "+" in arg:
                initial_time, increment = parse_time_control(arg)
            elif arg.lower() in {"true", "yes", "rated"}:
                rated = True
            elif arg.lower() in {"false", "no", "unrated", "casual"}:
                rated = False
            elif arg.lower() in {"white", "black", "random"}:
                color = ChallengeColor(arg.lower())
            elif found_variant := find_variant(arg):
                variant = found_variant
            elif opponent_username is None:
                opponent_username = arg
            else:
                print(f"Unknown argument: {arg}")

        if opponent_username is None:
            raise ValueError("Username is required.")

        return ChallengeRequest(opponent_username, initial_time, increment, rated, color, variant, timeout)

    def replaced(self, **changes: Any) -> "ChallengeRequest":
        return replace(self, **changes)

    def __eq__(self, value: object) -> bool:
        if isinstance(value, ChallengeRequest):
            return value.opponent_username == self.opponent_username

        return NotImplemented

    def __hash__(self) -> int:
        return hash(self.opponent_username)


@dataclass(kw_only=True)
class ChallengeResponse:
    challenge_id: str | None = None
    success: bool = False
    no_opponent: bool = False
    has_reached_rate_limit: bool = False
    wait_seconds: int | None = None
    is_misconfigured: bool = False


@dataclass
class ChatMessage:
    username: str
    text: str
    room: Literal["player", "spectator"]

    @classmethod
    def from_chat_line_event(cls, chat_line_event: dict[str, Any]) -> "ChatMessage":
        username = chat_line_event["username"]
        text = chat_line_event["text"]
        room = chat_line_event["room"]

        return cls(username, text, room)


@dataclass(frozen=True)
class GameInformation:
    id_: str
    white_title: str | None
    white_name: str
    white_rating: int | None
    white_ai_level: int | None
    white_provisional: bool
    black_title: str | None
    black_name: str
    black_rating: int | None
    black_ai_level: int | None
    black_provisional: bool
    initial_time_ms: int
    increment_ms: int
    speed: str
    rated: bool
    variant: Variant
    variant_name: str
    initial_fen: str
    state: dict[str, Any]
    tournament_id: str | None

    @classmethod
    def from_game_full_event(cls, game_full_event: dict[str, Any]) -> "GameInformation":
        assert game_full_event["type"] == "gameFull"

        id_ = game_full_event["id"]
        white_title = game_full_event["white"].get("title")
        white_name = game_full_event["white"].get("name", "AI")
        white_rating = game_full_event["white"].get("rating")
        white_ai_level = game_full_event["white"].get("aiLevel")
        white_provisional = game_full_event["white"].get("provisional", False)
        black_title = game_full_event["black"].get("title")
        black_name = game_full_event["black"].get("name", "AI")
        black_rating = game_full_event["black"].get("rating")
        black_ai_level = game_full_event["black"].get("aiLevel")
        black_provisional = game_full_event["black"].get("provisional", False)
        initial_time_ms = game_full_event["clock"]["initial"]
        increment_ms = game_full_event["clock"]["increment"]
        speed = game_full_event["speed"]
        rated = game_full_event["rated"]
        variant = Variant(game_full_event["variant"]["key"])
        variant_name = game_full_event["variant"]["name"]
        initial_fen = game_full_event["initialFen"]
        state = game_full_event["state"]
        tournament_id = game_full_event.get("tournamentId")

        return cls(
            id_,
            white_title,
            white_name,
            white_rating,
            white_ai_level,
            white_provisional,
            black_title,
            black_name,
            black_rating,
            black_ai_level,
            black_provisional,
            initial_time_ms,
            increment_ms,
            speed,
            rated,
            variant,
            variant_name,
            initial_fen,
            state,
            tournament_id,
        )

    @property
    def id_str(self) -> str:
        return f"ID: {self.id_}"

    @property
    def white_name_str(self) -> str:
        title_str = f"{self.white_title} " if self.white_title else ""
        return f"{title_str}{self.white_name}"

    @property
    def white_str(self) -> str:
        provisional_str = "?" if self.white_provisional else ""
        rating_str = f"{self.white_rating}{provisional_str}" if self.white_rating else f"Level {self.white_ai_level}"
        return f"{self.white_name_str} ({rating_str})"

    @property
    def black_name_str(self) -> str:
        title_str = f"{self.black_title} " if self.black_title else ""
        return f"{title_str}{self.black_name}"

    @property
    def black_str(self) -> str:
        provisional_str = "?" if self.black_provisional else ""
        rating_str = f"{self.black_rating}{provisional_str}" if self.black_rating else f"Level {self.black_ai_level}"
        return f"{self.black_name_str} ({rating_str})"

    @property
    def tc_str(self) -> str:
        initial_time_min = self.initial_time_ms / 60_000
        if initial_time_min.is_integer():
            initial_time_min = int(initial_time_min)
        return f"{initial_time_min}+{self.increment_ms // 1000}"

    @property
    def tc_format(self) -> str:
        initial_time_min = self.initial_time_ms / 60_000
        if initial_time_min.is_integer():
            initial_time_str = str(int(initial_time_min))
        elif initial_time_min == 0.25:
            initial_time_str = "¼"
        elif initial_time_min == 0.5:
            initial_time_str = "½"
        elif initial_time_min == 0.75:
            initial_time_str = "¾"
        else:
            initial_time_str = str(initial_time_min)
        increment_sec = self.increment_ms // 1000
        return f"TC: {initial_time_str}+{increment_sec}"

    @property
    def rated_str(self) -> str:
        return "Rated" if self.rated else "Casual"

    @property
    def variant_str(self) -> str:
        return f"Variant: {self.variant_name}"

    @property
    def white_opponent(self) -> chess.engine.Opponent:
        return chess.engine.Opponent(self.white_name, self.white_title, self.white_rating, self.white_title == "BOT")

    @property
    def black_opponent(self) -> chess.engine.Opponent:
        return chess.engine.Opponent(self.black_name, self.black_title, self.black_rating, self.black_title == "BOT")

    @property
    def opponent_is_bot(self) -> bool:
        return self.white_title == "BOT" and self.black_title == "BOT"

    @property
    def opponent_is_human(self) -> bool:
        return self.white_title != "BOT" or self.black_title != "BOT"


@dataclass
class GaviotaResult:
    move: chess.Move
    wdl: Literal[-2, -1, 0, 1, 2]
    dtm: int


@dataclass
class LichessMove:
    uci_move: str
    offer_draw: bool
    resign: bool


@dataclass
class MatchmakingData:
    release_time: datetime = datetime.min
    multiplier: int = 1
    color: ChallengeColor = ChallengeColor.WHITE

    @classmethod
    def from_dict(cls, dict_: dict[str, Any]) -> "MatchmakingData":
        release_time = datetime.fromisoformat(dict_["release_time"]) if "release_time" in dict_ else datetime.now()
        multiplier = dict_.get("multiplier", 1)
        color = ChallengeColor(dict_["color"]) if "color" in dict_ else ChallengeColor.WHITE

        return MatchmakingData(release_time, multiplier, color)

    def to_dict(self) -> dict[str, Any]:
        dict_ = {}
        if self.release_time > datetime.now():
            dict_["release_time"] = self.release_time.isoformat(timespec="seconds")

        if self.multiplier == -1 and "release_time" in dict_:
            dict_["multiplier"] = -1

        if self.multiplier > 1:
            dict_["multiplier"] = self.multiplier

        if self.color == ChallengeColor.BLACK:
            dict_["color"] = ChallengeColor.BLACK

        return dict_


@dataclass
class MatchmakingType:
    name: str
    initial_time: int
    increment: int
    rated: bool
    variant: Variant
    perf_type: PerfType
    config_multiplier: int | None
    multiplier: int
    weight: float
    min_rating_diff: int | None
    max_rating_diff: int | None

    def __post_init__(self) -> None:
        self.estimated_game_duration = timedelta(seconds=max(self.initial_time, 3) * 1.34 + self.increment * 91.76)

    def __str__(self) -> str:
        initial_time_min = self.initial_time / 60
        if initial_time_min.is_integer():
            initial_time_str = str(int(initial_time_min))
        elif initial_time_min == 0.25:
            initial_time_str = "¼"
        elif initial_time_min == 0.5:
            initial_time_str = "½"
        elif initial_time_min == 0.75:
            initial_time_str = "¾"
        else:
            initial_time_str = str(initial_time_min)
        tc_str = f"TC: {initial_time_str}+{self.increment}"
        rated_str = "Rated" if self.rated else "Casual"
        variant_str = f"Variant: {self.variant}"
        delimiter = 4 * " "

        return delimiter.join([self.name, tc_str, rated_str, variant_str])

    def __eq__(self, value: object) -> bool:
        if isinstance(value, MatchmakingType):
            return value.name == self.name

        return NotImplemented

    def __hash__(self) -> int:
        return hash(self.name)


@dataclass
class MoveResponse:
    move: chess.Move
    public_message: str
    private_message: str = field(default="", kw_only=True)
    pv: list[chess.Move] = field(default_factory=list, kw_only=True)
    is_draw: bool | None = field(default=None, kw_only=True)
    is_lost: bool | None = field(default=None, kw_only=True)
    trusted_eval: bool = field(default=False, kw_only=True)


@dataclass
class MoveSource:
    method: Callable[[], Awaitable[MoveResponse | None]]
    priority: int
    conditions: list[bool] = field(default_factory=list)

    @property
    def is_available(self) -> bool:
        return all(self.conditions)


@dataclass
class SyzygyResult:
    move: chess.Move
    wdl: Literal[-2, -1, 0, 1, 2]
    dtz: int


@dataclass
class TournamentRequest:
    id_: str
    team: str | None
    password: str | None


@dataclass
class Tournament:
    id_: str
    start_time: datetime
    end_time: datetime
    name: str
    bots_allowed: bool
    team: str | None = None
    password: str | None = None
    start_task: Task[None] | None = None
    end_task: Task[None] | None = None

    @classmethod
    def from_tournament_info(cls, tournament_info: dict[str, Any]) -> "Tournament":
        return cls(
            tournament_info["id"],
            start_time := datetime.fromisoformat(tournament_info["startsAt"]),
            start_time + timedelta(minutes=tournament_info["minutes"]),
            tournament_info.get("fullName", ""),
            tournament_info.get("botsAllowed", False),
        )

    @property
    def seconds_to_start(self) -> float:
        return (self.start_time - datetime.now(UTC)).total_seconds() - 60.0

    @property
    def seconds_to_finish(self) -> float:
        return (self.end_time - datetime.now(UTC)).total_seconds()

    def cancel(self) -> None:
        if self.start_task:
            self.start_task.cancel()

        if self.end_task:
            self.end_task.cancel()