Add GMCP and MSDP support for rich clients
Implements Phase 7 foundation: - gmcp.py module with package builders for Char.Vitals, Char.Status, Room.Info, Room.Map, and MSDP vitals - Player helper methods send_gmcp() and send_msdp() for convenience - Full test coverage for all GMCP/MSDP functions and edge cases
This commit is contained in:
parent
058ba1b7de
commit
d253012122
3 changed files with 371 additions and 0 deletions
113
src/mudlib/gmcp.py
Normal file
113
src/mudlib/gmcp.py
Normal file
|
|
@ -0,0 +1,113 @@
|
||||||
|
"""GMCP and MSDP package support for rich MUD clients."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from mudlib.player import Player
|
||||||
|
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def send_char_vitals(player: Player) -> None:
|
||||||
|
"""Send Char.Vitals — pl, stamina, max_stamina."""
|
||||||
|
player.writer.send_gmcp(
|
||||||
|
"Char.Vitals",
|
||||||
|
{
|
||||||
|
"pl": round(player.pl, 1),
|
||||||
|
"stamina": round(player.stamina, 1),
|
||||||
|
"max_stamina": round(player.max_stamina, 1),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def send_char_status(player: Player) -> None:
|
||||||
|
"""Send Char.Status — flying, resting, mode, in_combat."""
|
||||||
|
player.writer.send_gmcp(
|
||||||
|
"Char.Status",
|
||||||
|
{
|
||||||
|
"flying": player.flying,
|
||||||
|
"resting": player.resting,
|
||||||
|
"mode": player.mode,
|
||||||
|
"in_combat": player.mode == "combat",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def send_room_info(player: Player) -> None:
|
||||||
|
"""Send Room.Info — zone, coordinates, terrain, exits."""
|
||||||
|
from mudlib.zone import Zone
|
||||||
|
|
||||||
|
zone = player.location
|
||||||
|
if not isinstance(zone, Zone):
|
||||||
|
return
|
||||||
|
|
||||||
|
terrain = zone.get_tile(player.x, player.y)
|
||||||
|
|
||||||
|
# Build exits list from passable adjacent tiles
|
||||||
|
exits = []
|
||||||
|
directions = [
|
||||||
|
("north", 0, -1),
|
||||||
|
("south", 0, 1),
|
||||||
|
("east", 1, 0),
|
||||||
|
("west", -1, 0),
|
||||||
|
]
|
||||||
|
for name, dx, dy in directions:
|
||||||
|
nx, ny = player.x + dx, player.y + dy
|
||||||
|
if zone.is_passable(nx, ny):
|
||||||
|
exits.append(name)
|
||||||
|
|
||||||
|
player.writer.send_gmcp(
|
||||||
|
"Room.Info",
|
||||||
|
{
|
||||||
|
"zone": zone.name,
|
||||||
|
"x": player.x,
|
||||||
|
"y": player.y,
|
||||||
|
"terrain": terrain,
|
||||||
|
"exits": exits,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def send_map_data(player: Player) -> None:
|
||||||
|
"""Send Room.Map — terrain viewport around player for client rendering."""
|
||||||
|
from mudlib.zone import Zone
|
||||||
|
|
||||||
|
zone = player.location
|
||||||
|
if not isinstance(zone, Zone):
|
||||||
|
return
|
||||||
|
|
||||||
|
radius = 10
|
||||||
|
rows = []
|
||||||
|
for dy in range(-radius, radius + 1):
|
||||||
|
row = []
|
||||||
|
for dx in range(-radius, radius + 1):
|
||||||
|
tx = player.x + dx
|
||||||
|
ty = player.y + dy
|
||||||
|
wx, wy = zone.wrap(tx, ty)
|
||||||
|
row.append(zone.terrain[wy][wx])
|
||||||
|
rows.append(row)
|
||||||
|
|
||||||
|
player.writer.send_gmcp(
|
||||||
|
"Room.Map",
|
||||||
|
{
|
||||||
|
"x": player.x,
|
||||||
|
"y": player.y,
|
||||||
|
"radius": radius,
|
||||||
|
"terrain": rows,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def send_msdp_vitals(player: Player) -> None:
|
||||||
|
"""Send MSDP variable updates for real-time gauges."""
|
||||||
|
player.writer.send_msdp(
|
||||||
|
{
|
||||||
|
"HEALTH": str(round(player.pl, 1)),
|
||||||
|
"HEALTH_MAX": str(round(player.pl, 1)),
|
||||||
|
"STAMINA": str(round(player.stamina, 1)),
|
||||||
|
"STAMINA_MAX": str(round(player.max_stamina, 1)),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
@ -44,6 +44,16 @@ class Player(Entity):
|
||||||
self.writer.write(message)
|
self.writer.write(message)
|
||||||
await self.writer.drain()
|
await self.writer.drain()
|
||||||
|
|
||||||
|
def send_gmcp(self, package: str, data: Any = None) -> None:
|
||||||
|
"""Send a GMCP message to the client (no-op if unsupported)."""
|
||||||
|
if self.writer is not None:
|
||||||
|
self.writer.send_gmcp(package, data)
|
||||||
|
|
||||||
|
def send_msdp(self, variables: dict) -> None:
|
||||||
|
"""Send MSDP variables to the client (no-op if unsupported)."""
|
||||||
|
if self.writer is not None:
|
||||||
|
self.writer.send_msdp(variables)
|
||||||
|
|
||||||
|
|
||||||
# Global registry of connected players
|
# Global registry of connected players
|
||||||
players: dict[str, Player] = {}
|
players: dict[str, Player] = {}
|
||||||
|
|
|
||||||
248
tests/test_gmcp.py
Normal file
248
tests/test_gmcp.py
Normal file
|
|
@ -0,0 +1,248 @@
|
||||||
|
"""Tests for GMCP and MSDP support."""
|
||||||
|
|
||||||
|
from unittest.mock import AsyncMock, MagicMock
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from mudlib.gmcp import (
|
||||||
|
send_char_status,
|
||||||
|
send_char_vitals,
|
||||||
|
send_map_data,
|
||||||
|
send_msdp_vitals,
|
||||||
|
send_room_info,
|
||||||
|
)
|
||||||
|
from mudlib.player import Player, players
|
||||||
|
from mudlib.zone import Zone
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def clear_state():
|
||||||
|
"""Clear players before and after each test."""
|
||||||
|
players.clear()
|
||||||
|
yield
|
||||||
|
players.clear()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def test_zone():
|
||||||
|
"""Create a small test zone with known terrain."""
|
||||||
|
terrain = [["." for _ in range(25)] for _ in range(25)] # 25x25 simple terrain
|
||||||
|
# Add some impassable tiles for exit testing
|
||||||
|
terrain[5][10] = "^" # mountain north of player
|
||||||
|
zone = Zone(
|
||||||
|
name="testzone",
|
||||||
|
width=25,
|
||||||
|
height=25,
|
||||||
|
toroidal=True,
|
||||||
|
terrain=terrain,
|
||||||
|
impassable={"^", "~"},
|
||||||
|
)
|
||||||
|
return zone
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def mock_writer():
|
||||||
|
writer = MagicMock()
|
||||||
|
writer.write = MagicMock()
|
||||||
|
writer.drain = AsyncMock()
|
||||||
|
writer.send_gmcp = MagicMock()
|
||||||
|
writer.send_msdp = MagicMock()
|
||||||
|
return writer
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def player(mock_writer, test_zone):
|
||||||
|
p = Player(name="Goku", x=10, y=10, writer=mock_writer)
|
||||||
|
p.location = test_zone
|
||||||
|
p.pl = 85.5
|
||||||
|
p.stamina = 42.3
|
||||||
|
p.max_stamina = 100.0
|
||||||
|
players[p.name] = p
|
||||||
|
return p
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_char_vitals(player):
|
||||||
|
"""Test Char.Vitals sends correct payload."""
|
||||||
|
send_char_vitals(player)
|
||||||
|
|
||||||
|
player.writer.send_gmcp.assert_called_once_with(
|
||||||
|
"Char.Vitals",
|
||||||
|
{
|
||||||
|
"pl": 85.5,
|
||||||
|
"stamina": 42.3,
|
||||||
|
"max_stamina": 100.0,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_char_status_normal_mode(player):
|
||||||
|
"""Test Char.Status sends correct payload in normal mode."""
|
||||||
|
player.flying = False
|
||||||
|
player.resting = False
|
||||||
|
player.mode_stack = ["normal"]
|
||||||
|
|
||||||
|
send_char_status(player)
|
||||||
|
|
||||||
|
player.writer.send_gmcp.assert_called_once_with(
|
||||||
|
"Char.Status",
|
||||||
|
{
|
||||||
|
"flying": False,
|
||||||
|
"resting": False,
|
||||||
|
"mode": "normal",
|
||||||
|
"in_combat": False,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_char_status_in_combat(player):
|
||||||
|
"""Test Char.Status detects combat mode."""
|
||||||
|
player.mode_stack = ["normal", "combat"]
|
||||||
|
|
||||||
|
send_char_status(player)
|
||||||
|
|
||||||
|
args = player.writer.send_gmcp.call_args[0]
|
||||||
|
assert args[0] == "Char.Status"
|
||||||
|
assert args[1]["mode"] == "combat"
|
||||||
|
assert args[1]["in_combat"] is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_char_status_flying_and_resting(player):
|
||||||
|
"""Test Char.Status reflects flying and resting states."""
|
||||||
|
player.flying = True
|
||||||
|
player.resting = True
|
||||||
|
|
||||||
|
send_char_status(player)
|
||||||
|
|
||||||
|
args = player.writer.send_gmcp.call_args[0]
|
||||||
|
assert args[1]["flying"] is True
|
||||||
|
assert args[1]["resting"] is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_room_info(player):
|
||||||
|
"""Test Room.Info sends zone, coordinates, terrain, and exits."""
|
||||||
|
send_room_info(player)
|
||||||
|
|
||||||
|
player.writer.send_gmcp.assert_called_once()
|
||||||
|
args = player.writer.send_gmcp.call_args[0]
|
||||||
|
assert args[0] == "Room.Info"
|
||||||
|
|
||||||
|
data = args[1]
|
||||||
|
assert data["zone"] == "testzone"
|
||||||
|
assert data["x"] == 10
|
||||||
|
assert data["y"] == 10
|
||||||
|
assert data["terrain"] == "."
|
||||||
|
# All adjacent tiles should be passable except north (mountain at 10,5)
|
||||||
|
# north is y-1 = 9, but mountain is at y=5
|
||||||
|
# Player at 10,10 so adjacent tiles are all "." and passable
|
||||||
|
assert "north" in data["exits"]
|
||||||
|
assert "south" in data["exits"]
|
||||||
|
assert "east" in data["exits"]
|
||||||
|
assert "west" in data["exits"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_room_info_blocked_exit(player, test_zone):
|
||||||
|
"""Test Room.Info excludes blocked exits."""
|
||||||
|
# Put player next to mountain at 10,5
|
||||||
|
player.y = 6
|
||||||
|
test_zone.terrain[5][10] = "^" # mountain to the north
|
||||||
|
|
||||||
|
send_room_info(player)
|
||||||
|
|
||||||
|
args = player.writer.send_gmcp.call_args[0]
|
||||||
|
data = args[1]
|
||||||
|
# North (y-1=5) should be impassable
|
||||||
|
assert "north" not in data["exits"]
|
||||||
|
assert "south" in data["exits"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_room_info_no_zone(player):
|
||||||
|
"""Test Room.Info with no zone does not crash."""
|
||||||
|
player.location = None
|
||||||
|
|
||||||
|
send_room_info(player)
|
||||||
|
|
||||||
|
# Should not have called send_gmcp
|
||||||
|
player.writer.send_gmcp.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_map_data(player):
|
||||||
|
"""Test Room.Map sends terrain grid."""
|
||||||
|
send_map_data(player)
|
||||||
|
|
||||||
|
player.writer.send_gmcp.assert_called_once()
|
||||||
|
args = player.writer.send_gmcp.call_args[0]
|
||||||
|
assert args[0] == "Room.Map"
|
||||||
|
|
||||||
|
data = args[1]
|
||||||
|
assert data["x"] == 10
|
||||||
|
assert data["y"] == 10
|
||||||
|
assert data["radius"] == 10
|
||||||
|
# Grid should be 21x21 (radius 10 = -10 to +10 inclusive)
|
||||||
|
assert len(data["terrain"]) == 21
|
||||||
|
assert len(data["terrain"][0]) == 21
|
||||||
|
# Center tile should be "."
|
||||||
|
assert data["terrain"][10][10] == "."
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_map_data_toroidal_wrapping(player, test_zone):
|
||||||
|
"""Test Room.Map wraps correctly in toroidal zones."""
|
||||||
|
# Put player near edge to test wrapping
|
||||||
|
player.x = 2
|
||||||
|
player.y = 2
|
||||||
|
|
||||||
|
send_map_data(player)
|
||||||
|
|
||||||
|
args = player.writer.send_gmcp.call_args[0]
|
||||||
|
data = args[1]
|
||||||
|
# Should still be 21x21 even near edge (wraps around)
|
||||||
|
assert len(data["terrain"]) == 21
|
||||||
|
assert len(data["terrain"][0]) == 21
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_map_data_no_zone(player):
|
||||||
|
"""Test Room.Map with no zone does not crash."""
|
||||||
|
player.location = None
|
||||||
|
|
||||||
|
send_map_data(player)
|
||||||
|
|
||||||
|
player.writer.send_gmcp.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_msdp_vitals(player):
|
||||||
|
"""Test MSDP sends vitals with correct variable names."""
|
||||||
|
send_msdp_vitals(player)
|
||||||
|
|
||||||
|
player.writer.send_msdp.assert_called_once_with(
|
||||||
|
{
|
||||||
|
"HEALTH": "85.5",
|
||||||
|
"HEALTH_MAX": "85.5",
|
||||||
|
"STAMINA": "42.3",
|
||||||
|
"STAMINA_MAX": "100.0",
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_player_send_gmcp(player):
|
||||||
|
"""Test Player.send_gmcp convenience method."""
|
||||||
|
player.send_gmcp("Test.Package", {"key": "value"})
|
||||||
|
|
||||||
|
player.writer.send_gmcp.assert_called_once_with("Test.Package", {"key": "value"})
|
||||||
|
|
||||||
|
|
||||||
|
def test_player_send_gmcp_no_writer():
|
||||||
|
"""Test Player.send_gmcp with no writer does not crash."""
|
||||||
|
p = Player(name="NoWriter", writer=None)
|
||||||
|
p.send_gmcp("Test.Package", {}) # Should not raise
|
||||||
|
|
||||||
|
|
||||||
|
def test_player_send_msdp(player):
|
||||||
|
"""Test Player.send_msdp convenience method."""
|
||||||
|
player.send_msdp({"VAR": "value"})
|
||||||
|
|
||||||
|
player.writer.send_msdp.assert_called_once_with({"VAR": "value"})
|
||||||
|
|
||||||
|
|
||||||
|
def test_player_send_msdp_no_writer():
|
||||||
|
"""Test Player.send_msdp with no writer does not crash."""
|
||||||
|
p = Player(name="NoWriter", writer=None)
|
||||||
|
p.send_msdp({}) # Should not raise
|
||||||
Loading…
Reference in a new issue