freeciv_env/server/app.py

from __future__ import annotations

import glob
import os
from urllib.request import Request, urlopen

from fastapi import Query
from fastapi.responses import RedirectResponse
from openenv.core.env_server import create_app

from freeciv_env.adapter import prepare_observation
from freeciv_env.models import FreecivAction, FreecivObservation
from freeciv_env.runtime import DEBUG_EVENTS, LiveFreecivSession
from freeciv_env.server.freeciv_environment import FreecivEnvironment


def create_live_session() -> LiveFreecivSession:
    return LiveFreecivSession(
        username=os.getenv("FREECIV_USERNAME", "openenvbot"),
        client_port=int(os.getenv("FREECIV_CLIENT_PORT", "6000")),
        base_url=os.getenv("FREECIV_SERVER_URL", "http://127.0.0.1"),
        turn_timeout_s=float(os.getenv("FREECIV_TURN_TIMEOUT_S", "60")),
    )


def create_freeciv_app(*, session_factory=create_live_session, max_turns: int | None = None):
    if max_turns is None:
        max_turns = int(os.getenv("FREECIV_MAX_TURNS", "50"))
    return create_app(
        lambda: FreecivEnvironment(session_factory=session_factory, max_turns=max_turns),
        FreecivAction,
        FreecivObservation,
        env_name="freeciv_env",
    )


app = create_freeciv_app()


@app.get("/", include_in_schema=False)
def root() -> RedirectResponse:
    return RedirectResponse(url="/web")


@app.get("/debug/internal-status")
def debug_internal_status() -> dict:
    checks = []
    for name, method, url in [
        ("nginx", "GET", "http://127.0.0.1/"),
        ("publite2", "GET", "http://127.0.0.1/pubstatus"),
        ("tomcat", "GET", "http://127.0.0.1:8080/freeciv-web/"),
        ("proxy7000", "GET", "http://127.0.0.1/civsocket/7000/status"),
        ("proxy7001", "GET", "http://127.0.0.1/civsocket/7001/status"),
        ("proxy7002", "GET", "http://127.0.0.1/civsocket/7002/status"),
        ("launcher", "POST", "http://127.0.0.1/civclientlauncher?civserverport=6000"),
    ]:
        try:
            request = Request(url, method=method)
            with urlopen(request, timeout=10) as response:
                body = response.read(200).decode("utf-8", "ignore")
                checks.append(
                    {
                        "name": name,
                        "ok": True,
                        "status": response.status,
                        "body": body,
                    }
                )
        except Exception as exc:
            checks.append({"name": name, "ok": False, "error": repr(exc)})
    return {"checks": checks}


@app.get("/debug/live-log")
def debug_live_log() -> dict:
    return {"events": list(DEBUG_EVENTS)}


@app.get("/debug/freeciv-logs")
def debug_freeciv_logs() -> dict:
    logs = {}
    for path in sorted(glob.glob("/docker/logs/*.log"))[-12:]:
        try:
            with open(path, "r", encoding="utf-8", errors="ignore") as handle:
                lines = handle.readlines()[-80:]
            logs[path] = "".join(lines)
        except Exception as exc:
            logs[path] = repr(exc)
    return {"logs": logs}


@app.get("/debug/startup-log")
def debug_startup_log() -> dict:
    path = "/tmp/start_space.log"
    try:
        with open(path, "r", encoding="utf-8", errors="ignore") as handle:
            lines = handle.readlines()[-120:]
        return {"path": path, "log": "".join(lines)}
    except Exception as exc:
        return {"path": path, "error": repr(exc)}


@app.post("/debug/live-reset")
def debug_live_reset(timeout_s: float = Query(default=120.0, ge=10.0, le=300.0)) -> dict:
    session = create_live_session()
    session.turn_timeout_s = timeout_s
    try:
        reset_snapshot = session.reset()
        reset_observation = prepare_observation(
            reset_snapshot,
            reward=0.0,
            done=False,
            status="ready",
            metadata={},
        ).observation
        next_snapshot = session.end_turn()
        next_observation = prepare_observation(
            next_snapshot,
            reward=0.0,
            done=False,
            status="ok",
            metadata={},
        ).observation
        return {
            "ok": True,
            "reset": {
                "turn": reset_observation.turn,
                "legal_actions": len(reset_observation.legal_actions),
                "summary": reset_observation.summary,
            },
            "step": {
                "turn": next_observation.turn,
                "legal_actions": len(next_observation.legal_actions),
                "summary": next_observation.summary,
            },
        }
    except Exception as exc:
        return {"ok": False, "error": repr(exc)}
    finally:
        session.close()


def main() -> None:
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000, ws_ping_interval=300, ws_ping_timeout=300)


if __name__ == "__main__":
    main()