shmup/mud
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add spawn command and wire mobs into server

Browse source 
Phase 6: spawn command creates mobs at player position from loaded
templates. Server loads mob templates from content/mobs/ at startup,
injects world into combat/commands module, and runs process_mobs()
each game loop tick after process_combat().
This commit is contained in:
Jared Miller 2026-02-08 23:04:00 -05:00
parent 42bb3fa046
commit 305f25e77a
Signed by: shmup
GPG key ID: 22B5C6D66A38B06C
7 changed files with 159 additions and 90 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
src/mudlib/combat/engine.py
View file

@ -114,9 +114,7 @@ async def process_combat() -> None:
from mudlib.mobs import despawn_mob
despawn_mob(loser)
await winner.send(
f"You have defeated the {loser.name}!\r\n"
)
await winner.send(f"You have defeated the {loser.name}!\r\n")
elif isinstance(winner, Mob):
await loser.send(
f"You have been defeated by the {winner.name}!\r\n"

5
src/mudlib/commands/look.py
View file

@ -89,10 +89,7 @@ async def cmd_look(player: Player, args: str) -> None:
if x == center_x and y == center_y:
line.append(colorize_terrain("@", player.color_depth))
# Check if this is another player's position
elif (x, y) in other_player_positions:
line.append(colorize_terrain("*", player.color_depth))
# Check if this is a mob's position
elif (x, y) in mob_positions:
elif (x, y) in other_player_positions or (x, y) in mob_positions:
line.append(colorize_terrain("*", player.color_depth))
else:
# Check for active effects at this world position

24
src/mudlib/commands/spawn.py Normal file
View file

@ -0,0 +1,24 @@
"""Spawn command for creating mobs."""
from mudlib.commands import CommandDefinition, register
from mudlib.mobs import mob_templates, spawn_mob
from mudlib.player import Player
async def cmd_spawn(player: Player, args: str) -> None:
"""Spawn a mob at the player's current position."""
name = args.strip().lower()
if not name:
await player.send("Usage: spawn <mob_type>\r\n")
return
if name not in mob_templates:
available = ", ".join(sorted(mob_templates.keys()))
await player.send(f"Unknown mob type: {name}\r\nAvailable: {available}\r\n")
return
mob = spawn_mob(mob_templates[name], player.x, player.y)
await player.send(f"A {mob.name} appears!\r\n")
register(CommandDefinition("spawn", cmd_spawn, aliases=[], mode="normal"))

13
src/mudlib/server.py
View file

@ -10,6 +10,7 @@ from typing import cast
import telnetlib3
from telnetlib3.server_shell import readline2
import mudlib.combat.commands
import mudlib.commands
import mudlib.commands.edit
import mudlib.commands.fly
@ -18,11 +19,14 @@ import mudlib.commands.look
import mudlib.commands.movement
import mudlib.commands.quit
import mudlib.commands.reload
import mudlib.commands.spawn
from mudlib.caps import parse_mtts
from mudlib.combat.commands import register_combat_commands
from mudlib.combat.engine import process_combat
from mudlib.content import load_commands
from mudlib.effects import clear_expired
from mudlib.mob_ai import process_mobs
from mudlib.mobs import load_mob_templates, mob_templates
from mudlib.player import Player, players
from mudlib.resting import process_resting
from mudlib.store import (
@ -65,6 +69,7 @@ async def game_loop() -> None:
t0 = asyncio.get_event_loop().time()
clear_expired()
await process_combat()
await process_mobs(mudlib.combat.commands.combat_moves)
await process_resting()
# Periodic auto-save (every 60 seconds)
@ -379,6 +384,7 @@ async def run_server() -> None:
mudlib.commands.fly.world = _world
mudlib.commands.look.world = _world
mudlib.commands.movement.world = _world
mudlib.combat.commands.world = _world
# Load content-defined commands from TOML files
content_dir = pathlib.Path(__file__).resolve().parents[2] / "content" / "commands"
@ -396,6 +402,13 @@ async def run_server() -> None:
register_combat_commands(combat_dir)
log.info("registered combat commands")
# Load mob templates
mobs_dir = pathlib.Path(__file__).resolve().parents[2] / "content" / "mobs"
if mobs_dir.exists():
loaded = load_mob_templates(mobs_dir)
mob_templates.update(loaded)
log.info("loaded %d mob templates from %s", len(loaded), mobs_dir)
# connect_maxwait: how long to wait for telnet option negotiation (CHARSET
# etc) before starting the shell. default is 4.0s which is painful.
# MUD clients like tintin++ reject CHARSET immediately via MTTS, but

34
tests/test_mob_ai.py
View file

@ -14,11 +14,9 @@ from mudlib.combat.engine import (
get_encounter,
start_encounter,
)
from mudlib.combat.moves import CombatMove, load_moves
from mudlib.entity import Mob
from mudlib.combat.moves import load_moves
from mudlib.mob_ai import process_mobs
from mudlib.mobs import (
despawn_mob,
load_mob_template,
mobs,
spawn_mob,
@ -68,9 +66,7 @@ def mock_reader():
@pytest.fixture
def player(mock_reader, mock_writer):
p = Player(
name="Goku", x=0, y=0, reader=mock_reader, writer=mock_writer
)
p = Player(name="Goku", x=0, y=0, reader=mock_reader, writer=mock_writer)
players[p.name] = p
return p
@ -138,9 +134,7 @@ class TestMobAttackAI:
assert encounter.current_move is not None
@pytest.mark.asyncio
async def test_mob_picks_from_its_own_moves(
self, player, goblin_toml, moves
):
async def test_mob_picks_from_its_own_moves(self, player, goblin_toml, moves):
"""Mob only picks moves from its moves list."""
template = load_mob_template(goblin_toml)
mob = spawn_mob(template, 0, 0)
@ -155,9 +149,7 @@ class TestMobAttackAI:
assert encounter.current_move.name in mob.moves
@pytest.mark.asyncio
async def test_mob_skips_when_stamina_too_low(
self, player, goblin_toml, moves
):
async def test_mob_skips_when_stamina_too_low(self, player, goblin_toml, moves):
"""Mob skips attack when stamina is too low for any move."""
template = load_mob_template(goblin_toml)
mob = spawn_mob(template, 0, 0)
@ -173,9 +165,7 @@ class TestMobAttackAI:
assert encounter.state == CombatState.IDLE
@pytest.mark.asyncio
async def test_mob_respects_cooldown(
self, player, goblin_toml, moves
):
async def test_mob_respects_cooldown(self, player, goblin_toml, moves):
"""Mob doesn't act when cooldown hasn't expired."""
template = load_mob_template(goblin_toml)
mob = spawn_mob(template, 0, 0)
@ -190,9 +180,7 @@ class TestMobAttackAI:
assert encounter.state == CombatState.IDLE
@pytest.mark.asyncio
async def test_mob_swaps_roles_when_defending(
self, player, goblin_toml, moves
):
async def test_mob_swaps_roles_when_defending(self, player, goblin_toml, moves):
"""Mob swaps attacker/defender roles when it attacks as defender."""
template = load_mob_template(goblin_toml)
mob = spawn_mob(template, 0, 0)
@ -209,9 +197,7 @@ class TestMobAttackAI:
assert encounter.defender is player
@pytest.mark.asyncio
async def test_mob_doesnt_act_outside_combat(
self, goblin_toml, moves
):
async def test_mob_doesnt_act_outside_combat(self, goblin_toml, moves):
"""Mob not in combat does nothing."""
template = load_mob_template(goblin_toml)
mob = spawn_mob(template, 0, 0)
@ -223,15 +209,13 @@ class TestMobAttackAI:
assert get_encounter(mob) is None
@pytest.mark.asyncio
async def test_mob_sets_cooldown_after_attack(
self, player, goblin_toml, moves
):
async def test_mob_sets_cooldown_after_attack(self, player, goblin_toml, moves):
"""Mob sets next_action_at after attacking."""
template = load_mob_template(goblin_toml)
mob = spawn_mob(template, 0, 0)
mob.next_action_at = 0.0
encounter = start_encounter(player, mob)
start_encounter(player, mob)
player.mode_stack.append("combat")
before = time.monotonic()

78
tests/test_mobs.py
View file

@ -1,6 +1,5 @@
"""Tests for mob templates, registry, spawn/despawn, and combat integration."""
import tomllib
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock
@ -13,7 +12,6 @@ from mudlib.combat.engine import active_encounters, get_encounter
from mudlib.combat.moves import load_moves
from mudlib.entity import Mob
from mudlib.mobs import (
MobTemplate,
despawn_mob,
get_nearby_mob,
load_mob_template,
@ -166,7 +164,7 @@ class TestGetNearbyMob:
def test_picks_closest_when_multiple(self, goblin_toml, mock_world):
template = load_mob_template(goblin_toml)
far_mob = spawn_mob(template, 8, 8)
spawn_mob(template, 8, 8)
close_mob = spawn_mob(template, 1, 1)
found = get_nearby_mob("goblin", 0, 0, mock_world)
assert found is close_mob
@ -204,9 +202,7 @@ def mock_reader():
@pytest.fixture
def player(mock_reader, mock_writer):
p = Player(
name="Goku", x=0, y=0, reader=mock_reader, writer=mock_writer
)
p = Player(name="Goku", x=0, y=0, reader=mock_reader, writer=mock_writer)
players[p.name] = p
return p
@ -233,9 +229,7 @@ def punch_right(moves):
class TestTargetResolution:
@pytest.mark.asyncio
async def test_attack_mob_by_name(
self, player, punch_right, goblin_toml
):
async def test_attack_mob_by_name(self, player, punch_right, goblin_toml):
"""do_attack with mob name finds and engages the mob."""
template = load_mob_template(goblin_toml)
mob = spawn_mob(template, 0, 0)
@ -272,9 +266,7 @@ class TestTargetResolution:
assert encounter.defender is goblin_player
@pytest.mark.asyncio
async def test_attack_mob_out_of_range(
self, player, punch_right, goblin_toml
):
async def test_attack_mob_out_of_range(self, player, punch_right, goblin_toml):
"""Mob outside viewport range is not found as target."""
template = load_mob_template(goblin_toml)
spawn_mob(template, 100, 100)
@ -283,15 +275,11 @@ class TestTargetResolution:
encounter = get_encounter(player)
assert encounter is None
messages = [
call[0][0] for call in player.writer.write.call_args_list
]
messages = [call[0][0] for call in player.writer.write.call_args_list]
assert any("need a target" in msg.lower() for msg in messages)
@pytest.mark.asyncio
async def test_encounter_mob_no_mode_push(
self, player, punch_right, goblin_toml
):
async def test_encounter_mob_no_mode_push(self, player, punch_right, goblin_toml):
"""Mob doesn't get mode_stack push (it has no mode_stack)."""
template = load_mob_template(goblin_toml)
mob = spawn_mob(template, 0, 0)
@ -318,8 +306,7 @@ class TestViewportRendering:
w.height = 256
w.get_viewport = MagicMock(
return_value=[
["." for _ in range(VIEWPORT_WIDTH)]
for _ in range(VIEWPORT_HEIGHT)
["." for _ in range(VIEWPORT_WIDTH)] for _ in range(VIEWPORT_HEIGHT)
]
)
w.wrap = lambda x, y: (x % 256, y % 256)
@ -327,9 +314,7 @@ class TestViewportRendering:
return w
@pytest.mark.asyncio
async def test_mob_renders_as_star(
self, player, goblin_toml, look_world
):
async def test_mob_renders_as_star(self, player, goblin_toml, look_world):
"""Mob within viewport renders as * in look output."""
import mudlib.commands.look as look_mod
@ -342,9 +327,7 @@ class TestViewportRendering:
await look_mod.cmd_look(player, "")
output = "".join(
call[0][0] for call in player.writer.write.call_args_list
)
output = "".join(call[0][0] for call in player.writer.write.call_args_list)
# The center is at (10, 5), mob at relative (12, 5)
# Output should contain a * character
assert "*" in output
@ -367,9 +350,7 @@ class TestViewportRendering:
await look_mod.cmd_look(player, "")
output = "".join(
call[0][0] for call in player.writer.write.call_args_list
)
output = "".join(call[0][0] for call in player.writer.write.call_args_list)
# Should only have @ (player) and . (terrain), no *
stripped = output.replace("\033[0m", "").replace("\r\n", "")
# Remove ANSI codes for terrain colors
@ -381,9 +362,7 @@ class TestViewportRendering:
look_mod.world = old
@pytest.mark.asyncio
async def test_dead_mob_not_rendered(
self, player, goblin_toml, look_world
):
async def test_dead_mob_not_rendered(self, player, goblin_toml, look_world):
"""Dead mob (alive=False) not rendered in viewport."""
import mudlib.commands.look as look_mod
@ -396,14 +375,10 @@ class TestViewportRendering:
await look_mod.cmd_look(player, "")
output = "".join(
call[0][0] for call in player.writer.write.call_args_list
)
output = "".join(call[0][0] for call in player.writer.write.call_args_list)
import re
stripped = re.sub(r"\033\[[0-9;]*m", "", output).replace(
"\r\n", ""
)
stripped = re.sub(r"\033\[[0-9;]*m", "", output).replace("\r\n", "")
assert "*" not in stripped
look_mod.world = old
@ -419,9 +394,7 @@ class TestMobDefeat:
return spawn_mob(template, 0, 0)
@pytest.mark.asyncio
async def test_mob_despawned_on_pl_zero(
self, player, goblin_mob, punch_right
):
async def test_mob_despawned_on_pl_zero(self, player, goblin_mob, punch_right):
"""Mob with PL <= 0 gets despawned after combat resolves."""
from mudlib.combat.engine import process_combat, start_encounter
@ -441,9 +414,7 @@ class TestMobDefeat:
assert goblin_mob.alive is False
@pytest.mark.asyncio
async def test_player_gets_victory_message(
self, player, goblin_mob, punch_right
):
async def test_player_gets_victory_message(self, player, goblin_mob, punch_right):
"""Player receives a victory message when mob is defeated."""
from mudlib.combat.engine import process_combat, start_encounter
@ -456,15 +427,11 @@ class TestMobDefeat:
await process_combat()
messages = [
call[0][0] for call in player.writer.write.call_args_list
]
messages = [call[0][0] for call in player.writer.write.call_args_list]
assert any("defeated" in msg.lower() for msg in messages)
@pytest.mark.asyncio
async def test_mob_stamina_depleted_despawns(
self, player, goblin_mob, punch_right
):
async def test_mob_stamina_depleted_despawns(self, player, goblin_mob, punch_right):
"""Mob is despawned when attacker stamina depleted (combat end)."""
from mudlib.combat.engine import process_combat, start_encounter
@ -482,9 +449,7 @@ class TestMobDefeat:
assert get_encounter(player) is None
@pytest.mark.asyncio
async def test_player_defeat_not_despawned(
self, player, goblin_mob, punch_right
):
async def test_player_defeat_not_despawned(self, player, goblin_mob, punch_right):
"""When player loses, player is not despawned."""
from mudlib.combat.engine import process_combat, start_encounter
@ -499,12 +464,9 @@ class TestMobDefeat:
await process_combat()
# Player should get defeat message, not be despawned
messages = [
call[0][0] for call in player.writer.write.call_args_list
]
messages = [call[0][0] for call in player.writer.write.call_args_list]
assert any(
"defeated" in msg.lower() or "damage" in msg.lower()
for msg in messages
"defeated" in msg.lower() or "damage" in msg.lower() for msg in messages
)
# Player is still in players dict (not removed)
assert player.name in players

91
tests/test_spawn_command.py Normal file
View file

@ -0,0 +1,91 @@
"""Tests for the spawn command."""
from unittest.mock import AsyncMock, MagicMock
import pytest
from mudlib.commands.spawn import cmd_spawn
from mudlib.mobs import MobTemplate, mob_templates, mobs
from mudlib.player import Player, players
@pytest.fixture(autouse=True)
def clear_state():
"""Clear mobs, templates, and players."""
mobs.clear()
mob_templates.clear()
players.clear()
yield
mobs.clear()
mob_templates.clear()
players.clear()
@pytest.fixture
def mock_writer():
writer = MagicMock()
writer.write = MagicMock()
writer.drain = AsyncMock()
return writer
@pytest.fixture
def mock_reader():
return MagicMock()
@pytest.fixture
def player(mock_reader, mock_writer):
p = Player(name="Goku", x=10, y=20, reader=mock_reader, writer=mock_writer)
players[p.name] = p
return p
@pytest.fixture
def goblin_template():
t = MobTemplate(
name="goblin",
description="a snarling goblin",
pl=50.0,
stamina=40.0,
max_stamina=40.0,
moves=["punch left"],
)
mob_templates["goblin"] = t
return t
@pytest.mark.asyncio
async def test_spawn_valid_mob(player, goblin_template):
"""Spawn creates a mob at the player's position."""
await cmd_spawn(player, "goblin")
assert len(mobs) == 1
mob = mobs[0]
assert mob.name == "goblin"
assert mob.x == player.x
assert mob.y == player.y
messages = [call[0][0] for call in player.writer.write.call_args_list]
assert any("goblin" in msg.lower() and "appears" in msg.lower() for msg in messages)
@pytest.mark.asyncio
async def test_spawn_invalid_type(player, goblin_template):
"""Spawn with unknown type shows available mobs."""
await cmd_spawn(player, "dragon")
assert len(mobs) == 0
messages = [call[0][0] for call in player.writer.write.call_args_list]
assert any("unknown" in msg.lower() for msg in messages)
assert any("goblin" in msg.lower() for msg in messages)
@pytest.mark.asyncio
async def test_spawn_no_args(player, goblin_template):
"""Spawn with no args shows usage."""
await cmd_spawn(player, "")
assert len(mobs) == 0
messages = [call[0][0] for call in player.writer.write.call_args_list]
assert any("usage" in msg.lower() for msg in messages)