freeciv_env/runtime.py

from __future__ import annotations

import asyncio
import json
import threading
import time
from collections import deque
from typing import Protocol
from urllib.parse import urlencode, urlparse
from urllib.request import Request, urlopen

from freeciv_env.adapter import ActionRef, RawSnapshot


DEBUG_EVENTS: deque[str] = deque(maxlen=400)


def debug_event(message: str) -> None:
    DEBUG_EVENTS.append(f"{time.strftime('%H:%M:%S')} {message}")


class FreecivSession(Protocol):
    def reset(self, seed: int | None = None) -> RawSnapshot: ...

    def apply_action(self, action_ref: ActionRef) -> RawSnapshot: ...

    def end_turn(self) -> RawSnapshot: ...

    def close(self) -> None: ...


class _InteractiveBot:
    def __init__(self, session: "LiveFreecivSession"):
        from freecivbot.bot.base_bot import BaseBot

        class InteractiveBotImpl(BaseBot):
            def __init__(self, owner: "LiveFreecivSession"):
                super().__init__()
                self._owner = owner

            def conduct_turn(self, pplayer, info_controls, end_turn_hook):
                super().conduct_turn(pplayer, info_controls, end_turn_hook)
                self._publish_snapshot()

            def calculate_next_move(self):
                if self._turn_active:
                    self._publish_snapshot()

            def _publish_snapshot(self):
                self._acquire_state()
                self._owner._publish_snapshot(
                    RawSnapshot(
                        turn=self.turn,
                        state=self._turn_state,
                        actions=self._turn_opts,
                    )
                )

        self.impl = InteractiveBotImpl(session)


class _ConfiguredCivClient:
    def __init__(self, bot, user_name: str, *, client_port: int, visual_monitor: bool = False):
        from freecivbot.civclient import CivClient

        class ConfiguredCivClientImpl(CivClient):
            def init_control(self, ws_client):
                self.ws_client = ws_client
                self.init_controller()
                if self.visual_monitor:
                    self.monitor.start_monitor()
                login_message = {
                    "pid": 4,
                    "username": self.user_name,
                    "capability": "+Freeciv.Web.Devel-3.3",
                    "version_label": "-dev",
                    "major_version": 2,
                    "minor_version": 5,
                    "patch_version": 99,
                    "port": self.client_port,
                    "password": None,
                    "subject": None,
                }
                debug_event(f"sending login username={self.user_name} port={self.client_port}")
                self.ws_client.send(login_message)

            def handle_chat_msg(self, packet):
                from freecivbot.utils.fc_events import E_UNDEFINED

                message = packet["message"]
                conn_id = packet["conn_id"]
                event = packet["event"]

                if message is None:
                    return
                if event is None or event < 0 or event >= E_UNDEFINED:
                    print("Undefined message event type")
                    print(packet)
                    print("\r\n")
                    packet["event"] = event = E_UNDEFINED

                if conn_id in self.clstate.connections:
                    message = "<b>" + self.clstate.connections[conn_id]["username"] + ":</b>" + message
                else:
                    if "/metamessage" in message:
                        return
                    if "Metaserver message string" in message:
                        return

                packet["message"] = message
                debug_event(f"chat message: {message}")
                print(packet)
                print("\r\n")

                if "You are logged in as" in message:
                    debug_event("logged in; sending /set minplayers 1 and prepare_game")
                    self.ws_client.send_message("/set minplayers 1")
                    self.prepare_game()

            def handle_conn_info(self, packet):
                from freecivbot.connectivity.client_state import C_S_PREPARING
                from freecivbot.utils.freecivlog import freelog

                pconn = self.clstate.find_conn_by_id(packet["id"])

                if not packet["used"]:
                    if pconn is None:
                        freelog(f"Server removed unknown connection {packet['id']}")
                        return
                    self.clstate.client_remove_cli_conn(pconn)
                    pconn = None
                else:
                    pplayer = self.player_ctrl.valid_player_by_number(packet["player_num"])
                    if pplayer is None:
                        return
                    packet["playing"] = pplayer

                    if self.clstate.has_id(packet["id"]):
                        self.clstate.init_state(packet)

                    self.clstate.conn_list_append(packet)

                if self.clstate.has_id(packet["id"]) and self.clstate.cur_player() != packet["playing"]:
                    self.clstate.set_client_state(C_S_PREPARING)

        self.impl = ConfiguredCivClientImpl(
            bot,
            user_name,
            client_port=client_port,
            visual_monitor=visual_monitor,
        )


class _ConfiguredCivConnection:
    def __init__(self, civ_client, base_url: str, *, owner: "LiveFreecivSession", wait_for_server: int = 120, retry_interval: int = 5):
        from math import ceil

        import websocket

        self._websocket = websocket
        self.client = civ_client
        self.base_url = base_url
        self._owner = owner
        self._loop = None
        self._owner._connection = self
        self.civserverport = self._reserve_client_port(base_url, civ_client.client_port)
        self.client.client_port = self.civserverport
        self.proxyport = 1000 + self.civserverport
        debug_event(f"reserved civ port={self.civserverport} proxyport={self.proxyport}")
        self._retry_interval = retry_interval
        self._num_retries = int(ceil(wait_for_server / retry_interval))
        self._cur_retry = 0
        self._ws_url = self._build_ws_url(base_url)
        self.network_init()

    def _build_ws_url(self, base_url: str) -> str:
        parsed = urlparse(base_url)
        scheme = "wss" if parsed.scheme == "https" else "ws"
        host = parsed.hostname or "localhost"
        return f"{scheme}://{host}:{self.proxyport}/civsocket/{self.proxyport}"

    def _reserve_client_port(self, base_url: str, requested_port: int) -> int:
        parsed = urlparse(base_url)
        scheme = parsed.scheme or "http"
        host = parsed.hostname or "localhost"
        port = parsed.port
        if port is None:
            port = 443 if scheme == "https" else 80
        query = urlencode({"civserverport": requested_port})
        launcher_url = f"{scheme}://{host}:{port}/civclientlauncher?{query}"
        origin = f"{scheme}://{host}:{port}"
        request = Request(
            launcher_url,
            method="POST",
            headers={
                "User-Agent": "Mozilla/5.0",
                "Origin": origin,
                "Referer": origin + "/",
            },
        )
        with urlopen(request, timeout=10) as response:
            result = response.headers.get("result")
            reserved_port = response.headers.get("port")
            debug_event(
                f"launcher response status={response.status} result={result} port={reserved_port} body={response.read(200).decode('utf-8', 'ignore')}"
            )
        if result != "success" or reserved_port is None:
            raise RuntimeError(f"failed to reserve freeciv client port via {launcher_url}")
        return int(reserved_port)

    def _retry(self):
        self._cur_retry += 1
        time.sleep(self._retry_interval)
        return self._detect_server_up()

    def _detect_server_up(self):
        ws = self._websocket.WebSocket()
        try:
            debug_event(f"probing websocket {self._ws_url}")
            ws.connect(self._ws_url, timeout=10)
            debug_event("websocket probe succeeded")
            return True
        except Exception as err:
            debug_event(f"websocket probe failed: {err!r}")
            print("Connect not successful:", err, " retrying in %s seconds." % self._retry_interval)
            if self._cur_retry < self._num_retries:
                return self._retry()
            return False
        finally:
            try:
                ws.close()
            except Exception:
                pass

    def network_init(self):
        self._cur_retry = 0
        print("Connecting to server at %s ..." % self.base_url)
        if self._detect_server_up():
            self.websocket_init()
        else:
            print("Connection could not be established!")

    def websocket_init(self):
        from tornado import ioloop

        from freecivbot.connectivity.clinet import CivWSClient

        asyncio.set_event_loop(asyncio.new_event_loop())
        ioloop.IOLoop.clear_current()
        self._loop = ioloop.IOLoop.current()

        debug_event(f"starting tornado websocket client for {self._ws_url}")
        client = CivWSClient(self.client)

        def send_json(data):
            if not client._ws_connection:
                raise RuntimeError("Web socket connection is closed.")
            msg = json.dumps(data, separators=(",", ":"))
            client._ws_connection.write_message(msg)

        client.send = send_json
        client.connect(self._ws_url)

        try:
            self._loop.start()
        except KeyboardInterrupt:
            client.close()

    def submit(self, fn) -> None:
        if self._loop is None:
            raise RuntimeError("freeciv connection loop is not ready")
        done = threading.Event()
        error: BaseException | None = None

        def run():
            nonlocal error
            try:
                fn()
            except BaseException as exc:
                error = exc
            finally:
                done.set()

        self._loop.add_callback(run)
        if not done.wait(timeout=10):
            raise TimeoutError("timed out dispatching action to freeciv loop")
        if error is not None:
            raise error

    def close(self) -> None:
        if self._loop is None:
            return
        self.submit(self.client.close)


class LiveFreecivSession:
    def __init__(
        self,
        *,
        username: str = "openenvbot",
        client_port: int = 6000,
        base_url: str = "http://localhost",
        turn_timeout_s: float = 60.0,
    ):
        self.username = username
        self.client_port = client_port
        self.base_url = base_url
        self.turn_timeout_s = turn_timeout_s

        self._bot_wrapper: _InteractiveBot | None = None
        self._client = None
        self._connection: _ConfiguredCivConnection | None = None
        self._thread: threading.Thread | None = None
        self._ready = threading.Event()
        self._snapshot_lock = threading.Lock()
        self._snapshot: RawSnapshot | None = None
        self._thread_error: BaseException | None = None
        self._reset_counter = 0
        self._session_seed = time.monotonic_ns() % 1_000_000

    def reset(self, seed: int | None = None) -> RawSnapshot:
        del seed
        self.close()
        self._reset_counter += 1
        username = self._next_username()
        client_port = self.client_port + ((self._session_seed + self._reset_counter - 1) % 3)

        self._ready.clear()
        self._thread_error = None
        self._snapshot = None

        self._bot_wrapper = _InteractiveBot(self)
        self._client = _ConfiguredCivClient(
            self._bot_wrapper.impl,
            username,
            client_port=client_port,
            visual_monitor=False,
        ).impl

        def run() -> None:
            try:
                debug_event(f"session thread starting username={username} base_url={self.base_url} client_port={client_port}")
                _ConfiguredCivConnection(self._client, self.base_url, owner=self)
            except BaseException as exc:  # pragma: no cover - surfaced in waiters
                debug_event(f"session thread error: {exc!r}")
                self._thread_error = exc
                self._ready.set()

        self._thread = threading.Thread(target=run, name="freeciv-live-session", daemon=True)
        self._thread.start()
        return self._wait_for_snapshot("reset")

    def apply_action(self, action_ref: ActionRef) -> RawSnapshot:
        snapshot = self._require_snapshot()
        action_list = snapshot.actions[action_ref.controller]
        valid_actions = action_list.get_actions(action_ref.actor_id, valid_only=True)
        action = None if valid_actions is None else valid_actions.get(action_ref.raw_action_key)
        if action is None:
            raise ValueError(
                f"action {action_ref.raw_action_key} is no longer valid for {action_ref.controller}:{action_ref.actor_id}"
            )
        self._ready.clear()
        connection = self._require_connection()
        connection.submit(lambda: action_list.trigger_validated_action(action))
        return self._wait_for_snapshot(action_ref.raw_action_key)

    def end_turn(self) -> RawSnapshot:
        if self._bot_wrapper is None:
            raise RuntimeError("session has not been reset")
        self._ready.clear()
        connection = self._require_connection()
        connection.submit(self._bot_wrapper.impl.end_turn)
        return self._wait_for_snapshot("end_turn")

    def close(self) -> None:
        debug_event("closing live session")
        if self._connection is not None:
            try:
                self._connection.close()
            except Exception:
                pass
        elif self._client is not None:
            try:
                self._client.close()
            except Exception:
                pass
        if self._thread is not None and self._thread.is_alive():
            self._thread.join(timeout=5)
        self._bot_wrapper = None
        self._client = None
        self._connection = None
        self._thread = None
        self._snapshot = None
        self._thread_error = None
        self._ready.clear()

    def _publish_snapshot(self, snapshot: RawSnapshot) -> None:
        debug_event(f"snapshot published turn={snapshot.turn}")
        with self._snapshot_lock:
            self._snapshot = snapshot
        self._ready.set()

    def _next_username(self) -> str:
        suffix = str(self._session_seed + self._reset_counter)
        prefix_len = max(1, 31 - len(suffix))
        return f"{self.username[:prefix_len]}{suffix}"

    def _require_connection(self) -> _ConfiguredCivConnection:
        if self._connection is None:
            raise RuntimeError("freeciv connection is not ready")
        return self._connection

    def _require_snapshot(self) -> RawSnapshot:
        with self._snapshot_lock:
            if self._snapshot is None:
                raise RuntimeError("no live snapshot is available")
            return self._snapshot

    def _wait_for_snapshot(self, reason: str) -> RawSnapshot:
        deadline = time.monotonic() + self.turn_timeout_s
        debug_event(f"waiting for snapshot reason={reason} timeout={self.turn_timeout_s}")
        while time.monotonic() < deadline:
            if self._thread_error is not None:
                raise RuntimeError(f"freeciv session failed during {reason}") from self._thread_error
            if self._ready.wait(timeout=0.1):
                if self._thread_error is not None:
                    raise RuntimeError(f"freeciv session failed during {reason}") from self._thread_error
                snapshot = self._require_snapshot()
                if snapshot is not None:
                    return snapshot
        debug_event(f"snapshot wait timed out reason={reason}")
        raise TimeoutError(f"timed out waiting for freeciv snapshot during {reason}")