solution.py

from __future__ import annotations

from typing import Any, Optional, Protocol, TypedDict
import socketio
import asyncio
from dataclasses import dataclass

import regex

url = "http://portal.hackazon.org:17001"


class Player(TypedDict):
    uid: int
    ready: bool
    host: bool


class GameInfo(TypedDict):
    name: str
    konami: str
    game_id: str
    players: int
    max_players: int
    public: bool
    slots: list[Optional[Player]]

class GridEntry(TypedDict):
    name: str
    type: str
    max: Optional[int]
    min: Optional[int]

class CommandRequest(TypedDict):
    text: str
    time: float

GridMap = dict[str, tuple[GridEntry, Any]]

class Emittable(Protocol):
    @classmethod
    def parse(cls, cmd: str, grid: GridMap) -> Optional[Emittable]:
        ...

    def emit(self) -> dict:
        ...

@dataclass
class Activate(Emittable):
    name: str
    value: bool

    @classmethod
    def parse(cls, cmd: str, grid: GridMap) -> Optional[Activate]:
        regexes = [r"Activate (?<name>[a-zA-Z ]+)",
                   r"Turn on (?<name>[a-zA-Z ]+)",
                   r"Switch on (?<name>[a-zA-Z ]+)",
                   r"Engage (?<name>[a-zA-Z ]+)",
                   ]
        for pat in regexes:
            if (match := regex.match(pat, cmd)) is not None:
                return Activate(name=match["name"], value=True)
        regexes = [r"Deactivate (?<name>[a-zA-Z ]+)",
                   r"Turn off (?<name>[a-zA-Z ]+)",
                   r"Switch off (?<name>[a-zA-Z ]+)",
                   r"Disengage (?<name>[a-zA-Z ]+)",
                   ]
        for pat in regexes:
            if (match := regex.match(pat, cmd)) is not None:
                return Activate(name=match["name"], value=False)

    def emit(self):
        return {"name": self.name, "value": self.value}

@dataclass
class Change(Emittable):
    name: str
    to: int

    @classmethod
    def parse(cls, cmd: str, grid: GridMap) -> Optional[Change]:
        regexes = [r"Set (?<name>[a-zA-Z ]+) to (?<to>[0-9]+)",
                   r"Change (?<name>[a-zA-Z ]+) to (?<to>[0-9]+)",
                   r"Increase (?<name>[a-zA-Z ]+) to (?<to>[0-9]+)",
                   r"Position (?<name>[a-zA-Z ]+) at (?<to>[0-9]+)",
                   r"Reduce (?<name>[a-zA-Z ]+) to (?<to>[0-9]+)",
                   r"Diminish (?<name>[a-zA-Z ]+) to (?<to>[0-9]+)",
                   ]
        for pat in regexes:
            if (match := regex.match(pat, cmd)) is not None:
                return Change(name=match["name"], to=int(match["to"]))
        regexes = [r"Set (?<name>[a-zA-Z ]+) to maximum",
                   r"Increase (?<name>[a-zA-Z ]+) to the max",
                   ]
        for pat in regexes:
            if (match := regex.match(pat, cmd)) is not None:
                val = grid[match["name"]][0]["max"]
                assert val is not None
                return Change(name=match["name"], to=val)
        regexes = [r"Set (?<name>[a-zA-Z ]+) to minimum",
                   r"Reduce (?<name>[a-zA-Z ]+) to the minimum",
                   ]
        for pat in regexes:
            if (match := regex.match(pat, cmd)) is not None:
                val = grid[match["name"]][0]["min"]
                assert val is not None
                return Change(name=match["name"], to=val)

    def emit(self):
        return {"name": self.name, "value": self.to}


def parse_command(command: str, grid: GridMap) -> Optional[tuple[dict, Any]]:
    if "Prepare" in command:
        return None
    for c in [Activate, Change]:
        if (r := c.parse(command, grid)) is not None:
            x = r.emit()
            print(f"sending command {x}")
            try:
                s = grid[r.name][1]
            except:
                print("what?? grid:",  grid)
                raise
            return x, s
    print(f"!!!!!!!Unknown command!!!! {command=}")

def parse_grid(grid: list[GridEntry], owner) -> dict[str, tuple[GridEntry, Any]]:
    return {e["name"]: (e, owner) for e in grid}

async def other(code, shared_grid):
    sio = socketio.AsyncClient(logger=True)

    uid = None
    connected = False

    @sio.event
    async def connect():
        print("Connected!")
        nonlocal connected
        connected = True
        await join_game()

    @sio.event
    async def welcome(data):
        nonlocal uid
        uid = data["uid"]
        print(f"received welcome {uid=}")
        await join_game()

    @sio.event
    async def game_join_success(data):
        await sio.emit("ready")

    async def join_game():
        if uid is None or not connected:
            return
        await sio.emit("join_game", {"konami": code})

    @sio.event
    async def game_started():
        await sio.emit("intro_done")

    @sio.event
    async def command(data: CommandRequest):
        print("Got command (other)", data)
        if (cmd := parse_command(data["text"], shared_grid)) is not None:
            await cmd[1].emit("command", cmd[0])

    @sio.event
    async def grid(data: list[GridEntry]):
        print("got grid (other):", data)
        shared_grid.update(parse_grid(data, sio))

    @sio.on("*")
    async def log(event, data=None):
        print("Fall through event (other) !", event, data)

    @sio.event
    async def next_level(data):
        await sio.emit("intro_done")

    await sio.connect(url)
    await sio.wait()


async def host():
    sio = socketio.AsyncClient(logger=True)

    uid = None
    connected = False
    game_id = None
    other_task = None
    grid_map = {}

    @sio.event
    async def connect():
        print("Connected!")
        nonlocal connected
        connected = True
        await start_game()

    @sio.event
    async def welcome(data):
        nonlocal uid
        uid = data["uid"]
        print(f"received welcome {uid=}")
        await start_game()

    @sio.event
    async def game_join_success(data):
        nonlocal game_id
        game_id = data["game_id"]
        await sio.emit("ready")

    @sio.event
    async def game_info(data: GameInfo):
        print("Game info:", data)
        nonlocal other_task
        if other_task is None:
            other_task = asyncio.create_task(other(data["konami"], grid_map))

        if (p0 := data["slots"][0]) is not None and p0["ready"]:
            if (p1 := data["slots"][1]) is not None and p1["ready"]:
                await sio.emit("start_game")

    async def start_game():
        if uid is None or not connected:
            return
        await sio.emit("create_game", {"name": "test", "public": True})

    @sio.event
    async def game_started():
        await sio.emit("intro_done")

    @sio.event
    async def command(data: CommandRequest):
        print("Got command (main)", data)
        assert grid_map is not None
        if (cmd := parse_command(data["text"], grid_map)) is not None:
            await cmd[1].emit("command", cmd[0])

    @sio.event
    async def grid(data: list[GridEntry]):
        print("got grid (main):", data)
        grid_map.update(parse_grid(data, sio))

    @sio.event
    async def next_level(data):
        print("level", data)
        await sio.emit("intro_done")
        # grid_map.clear()

    @sio.on("*")
    async def log(event, data=None):
        print("Fall through event (host) !", event, data)

    await sio.connect(url)
    await sio.wait()


asyncio.run(host())