149 lines
4.8 KiB
Python
149 lines
4.8 KiB
Python
"""GABS TCP JSON-RPC 2.0 client.
|
|
|
|
Low-level transport layer for communicating with the Bannerlord.GABS mod.
|
|
GABS runs inside the Windows VM and listens on port 4825. Messages are
|
|
newline-delimited JSON-RPC 2.0.
|
|
|
|
Wire format::
|
|
|
|
-> {"jsonrpc":"2.0","method":"core/get_game_state","id":1}\\n
|
|
<- {"jsonrpc":"2.0","result":{...},"id":1}\\n
|
|
|
|
All public methods raise :class:`GabsError` on failure so callers can
|
|
degrade gracefully without inspecting raw socket errors.
|
|
|
|
Refs: #1093 (M1 Observer), #1091 (Epic)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import socket
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_DEFAULT_HOST = "127.0.0.1"
|
|
_DEFAULT_PORT = 4825
|
|
_DEFAULT_TIMEOUT = 5.0
|
|
_RECV_BUFSIZE = 4096
|
|
|
|
|
|
class GabsError(Exception):
|
|
"""Raised when a GABS call fails (connection, protocol, or RPC error)."""
|
|
|
|
|
|
class GabsClient:
|
|
"""Synchronous TCP JSON-RPC 2.0 client for Bannerlord.GABS.
|
|
|
|
Each public call opens a fresh TCP connection, sends the request, reads
|
|
the response, and closes the socket. This avoids persistent-connection
|
|
complexity and is fast enough for poll intervals of ≥1 s.
|
|
|
|
Args:
|
|
host: VM IP or hostname (default ``127.0.0.1``).
|
|
port: GABS TCP port (default ``4825``).
|
|
timeout: Socket timeout in seconds (default ``5.0``).
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
host: str = _DEFAULT_HOST,
|
|
port: int = _DEFAULT_PORT,
|
|
timeout: float = _DEFAULT_TIMEOUT,
|
|
) -> None:
|
|
self.host = host
|
|
self.port = port
|
|
self.timeout = timeout
|
|
self._req_id = 0
|
|
|
|
# ── Public API ──────────────────────────────────────────────────────────
|
|
|
|
def call(self, method: str, params: dict[str, Any] | None = None) -> Any:
|
|
"""Send a JSON-RPC request and return the ``result`` value.
|
|
|
|
Args:
|
|
method: RPC method name (e.g. ``"core/get_game_state"``).
|
|
params: Optional parameters dict.
|
|
|
|
Returns:
|
|
The ``result`` field from the JSON-RPC response.
|
|
|
|
Raises:
|
|
GabsError: On any connection, protocol, or application-level error.
|
|
"""
|
|
self._req_id += 1
|
|
payload: dict[str, Any] = {
|
|
"jsonrpc": "2.0",
|
|
"method": method,
|
|
"id": self._req_id,
|
|
}
|
|
if params:
|
|
payload["params"] = params
|
|
|
|
try:
|
|
sock = socket.create_connection((self.host, self.port), timeout=self.timeout)
|
|
except OSError as exc:
|
|
raise GabsError(f"TCP connect to {self.host}:{self.port} failed: {exc}") from exc
|
|
|
|
try:
|
|
sock.settimeout(self.timeout)
|
|
raw = json.dumps(payload) + "\n"
|
|
sock.sendall(raw.encode())
|
|
|
|
buf = b""
|
|
while b"\n" not in buf:
|
|
chunk = sock.recv(_RECV_BUFSIZE)
|
|
if not chunk:
|
|
raise GabsError("Connection closed before response received")
|
|
buf += chunk
|
|
|
|
line = buf.split(b"\n", 1)[0]
|
|
resp: dict[str, Any] = json.loads(line.decode())
|
|
except GabsError:
|
|
raise
|
|
except json.JSONDecodeError as exc:
|
|
raise GabsError(f"Malformed JSON from GABS: {exc}") from exc
|
|
except OSError as exc:
|
|
raise GabsError(f"Socket error reading from GABS: {exc}") from exc
|
|
finally:
|
|
sock.close()
|
|
|
|
if "error" in resp:
|
|
err = resp["error"]
|
|
code = err.get("code", "?")
|
|
msg = err.get("message", "unknown error")
|
|
raise GabsError(f"GABS RPC error [{code}]: {msg}")
|
|
|
|
return resp.get("result")
|
|
|
|
def ping(self) -> bool:
|
|
"""Return True if GABS responds to a ping, False otherwise."""
|
|
try:
|
|
self.call("ping")
|
|
return True
|
|
except GabsError as exc:
|
|
logger.debug("GABS ping failed: %s", exc)
|
|
return False
|
|
|
|
def get_game_state(self) -> dict[str, Any]:
|
|
"""Return the current Bannerlord campaign game state."""
|
|
result = self.call("core/get_game_state")
|
|
return result if isinstance(result, dict) else {}
|
|
|
|
def get_player(self) -> dict[str, Any]:
|
|
"""Return the player hero's stats and status."""
|
|
result = self.call("hero/get_player")
|
|
return result if isinstance(result, dict) else {}
|
|
|
|
def get_player_party(self) -> dict[str, Any]:
|
|
"""Return the player's party composition and stats."""
|
|
result = self.call("party/get_player_party")
|
|
return result if isinstance(result, dict) else {}
|
|
|
|
def list_kingdoms(self) -> list[dict[str, Any]]:
|
|
"""Return the list of all active kingdoms in the campaign."""
|
|
result = self.call("kingdom/list_kingdoms")
|
|
return result if isinstance(result, list) else []
|