Add mob AI for combat decisions
Phase 5: process_mobs() runs each tick, handling mob attack and defense decisions. Mobs pick random attacks from their move list when IDLE, swap roles if needed, and attempt defense during TELEGRAPH/WINDOW with a 40% chance of correct counter. 1-second cooldown between actions. Training dummies with empty moves never fight back.
This commit is contained in:
parent
a61e998252
commit
23bb814ce0
GPG key ID:
22B5C6D66A38B06C
2 changed files with 447 additions and 0 deletions
114
src/mudlib/mob_ai.py
Normal file
114
src/mudlib/mob_ai.py
Normal file
|
|
@ -0,0 +1,114 @@
|
|||
"""Mob AI — game loop processor for mob combat decisions."""
|
||||
|
||||
import random
|
||||
import time
|
||||
|
||||
from mudlib.combat.encounter import CombatState
|
||||
from mudlib.combat.engine import get_encounter
|
||||
from mudlib.combat.moves import CombatMove
|
||||
from mudlib.mobs import mobs
|
||||
|
||||
# Seconds between mob actions (gives player time to read and react)
|
||||
MOB_ACTION_COOLDOWN = 1.0
|
||||
|
||||
|
||||
async def process_mobs(combat_moves: dict[str, CombatMove]) -> None:
|
||||
"""Called once per game loop tick. Handles mob combat decisions."""
|
||||
now = time.monotonic()
|
||||
|
||||
for mob in mobs[:]: # copy list in case of modification
|
||||
if not mob.alive:
|
||||
continue
|
||||
|
||||
encounter = get_encounter(mob)
|
||||
if encounter is None:
|
||||
continue
|
||||
|
||||
if now < mob.next_action_at:
|
||||
continue
|
||||
|
||||
# Determine if mob is attacker or defender in this encounter
|
||||
mob_is_defender = encounter.defender is mob
|
||||
|
||||
# Defense AI: react during TELEGRAPH or WINDOW when mob is defender
|
||||
if mob_is_defender and encounter.state in (
|
||||
CombatState.TELEGRAPH,
|
||||
CombatState.WINDOW,
|
||||
):
|
||||
_try_defend(mob, encounter, combat_moves, now)
|
||||
continue
|
||||
|
||||
# Attack AI: act when encounter is IDLE
|
||||
if encounter.state == CombatState.IDLE:
|
||||
_try_attack(mob, encounter, combat_moves, now)
|
||||
|
||||
|
||||
def _try_attack(mob, encounter, combat_moves, now):
|
||||
"""Attempt to pick and execute an attack move."""
|
||||
# Filter to affordable attack moves from mob's move list
|
||||
affordable = []
|
||||
for move_name in mob.moves:
|
||||
move = combat_moves.get(move_name)
|
||||
if (
|
||||
move is not None
|
||||
and move.move_type == "attack"
|
||||
and mob.stamina >= move.stamina_cost
|
||||
):
|
||||
affordable.append(move)
|
||||
|
||||
if not affordable:
|
||||
return
|
||||
|
||||
move = random.choice(affordable)
|
||||
|
||||
# Swap roles if mob is currently the defender
|
||||
if encounter.defender is mob:
|
||||
encounter.attacker, encounter.defender = (
|
||||
encounter.defender,
|
||||
encounter.attacker,
|
||||
)
|
||||
|
||||
encounter.attack(move)
|
||||
mob.next_action_at = now + MOB_ACTION_COOLDOWN
|
||||
|
||||
# Send telegraph to the player (the other participant)
|
||||
# This is fire-and-forget since mob.send is a no-op
|
||||
|
||||
|
||||
def _try_defend(mob, encounter, combat_moves, now):
|
||||
"""Attempt to pick and queue a defense move."""
|
||||
# Don't double-defend
|
||||
if encounter.pending_defense is not None:
|
||||
return
|
||||
|
||||
# Filter to affordable defense moves from mob's move list
|
||||
affordable_defenses = []
|
||||
for move_name in mob.moves:
|
||||
move = combat_moves.get(move_name)
|
||||
if (
|
||||
move is not None
|
||||
and move.move_type == "defense"
|
||||
and mob.stamina >= move.stamina_cost
|
||||
):
|
||||
affordable_defenses.append(move)
|
||||
|
||||
if not affordable_defenses:
|
||||
return
|
||||
|
||||
# 40% chance to pick a correct counter, 60% random
|
||||
chosen = None
|
||||
current_move = encounter.current_move
|
||||
if current_move and random.random() < 0.4:
|
||||
# Try to find a correct counter
|
||||
counters = []
|
||||
for defense in affordable_defenses:
|
||||
if defense.name in current_move.countered_by:
|
||||
counters.append(defense)
|
||||
if counters:
|
||||
chosen = random.choice(counters)
|
||||
|
||||
if chosen is None:
|
||||
chosen = random.choice(affordable_defenses)
|
||||
|
||||
encounter.defend(chosen)
|
||||
mob.stamina -= chosen.stamina_cost
|
||||
333
tests/test_mob_ai.py
Normal file
333
tests/test_mob_ai.py
Normal file
|
|
@ -0,0 +1,333 @@
|
|||
"""Tests for mob AI behavior."""
|
||||
|
||||
import time
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
import mudlib.commands.movement as movement_mod
|
||||
from mudlib.combat import commands as combat_commands
|
||||
from mudlib.combat.encounter import CombatState
|
||||
from mudlib.combat.engine import (
|
||||
active_encounters,
|
||||
get_encounter,
|
||||
start_encounter,
|
||||
)
|
||||
from mudlib.combat.moves import CombatMove, load_moves
|
||||
from mudlib.entity import Mob
|
||||
from mudlib.mob_ai import process_mobs
|
||||
from mudlib.mobs import (
|
||||
despawn_mob,
|
||||
load_mob_template,
|
||||
mobs,
|
||||
spawn_mob,
|
||||
)
|
||||
from mudlib.player import Player, players
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def clear_state():
|
||||
"""Clear mobs, encounters, and players before and after each test."""
|
||||
mobs.clear()
|
||||
active_encounters.clear()
|
||||
players.clear()
|
||||
yield
|
||||
mobs.clear()
|
||||
active_encounters.clear()
|
||||
players.clear()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_world():
|
||||
"""Inject a mock world for movement and combat commands."""
|
||||
fake_world = MagicMock()
|
||||
fake_world.width = 256
|
||||
fake_world.height = 256
|
||||
old_movement = movement_mod.world
|
||||
old_combat = combat_commands.world
|
||||
movement_mod.world = fake_world
|
||||
combat_commands.world = fake_world
|
||||
yield fake_world
|
||||
movement_mod.world = old_movement
|
||||
combat_commands.world = old_combat
|
||||
|
||||
|
||||
@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=0, y=0, reader=mock_reader, writer=mock_writer
|
||||
)
|
||||
players[p.name] = p
|
||||
return p
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def moves():
|
||||
"""Load combat moves from content directory."""
|
||||
content_dir = Path(__file__).parent.parent / "content" / "combat"
|
||||
return load_moves(content_dir)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def inject_moves(moves):
|
||||
"""Inject loaded moves into combat commands module."""
|
||||
combat_commands.combat_moves = moves
|
||||
yield
|
||||
combat_commands.combat_moves = {}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def goblin_toml(tmp_path):
|
||||
path = tmp_path / "goblin.toml"
|
||||
path.write_text(
|
||||
'name = "goblin"\n'
|
||||
'description = "a snarling goblin with a crude club"\n'
|
||||
"pl = 50.0\n"
|
||||
"stamina = 40.0\n"
|
||||
"max_stamina = 40.0\n"
|
||||
'moves = ["punch left", "punch right", "sweep"]\n'
|
||||
)
|
||||
return path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def dummy_toml(tmp_path):
|
||||
path = tmp_path / "training_dummy.toml"
|
||||
path.write_text(
|
||||
'name = "training dummy"\n'
|
||||
'description = "a battered wooden training dummy"\n'
|
||||
"pl = 200.0\n"
|
||||
"stamina = 100.0\n"
|
||||
"max_stamina = 100.0\n"
|
||||
"moves = []\n"
|
||||
)
|
||||
return path
|
||||
|
||||
|
||||
class TestMobAttackAI:
|
||||
@pytest.mark.asyncio
|
||||
async def test_mob_attacks_when_idle_and_cooldown_expired(
|
||||
self, player, goblin_toml, moves
|
||||
):
|
||||
"""Mob attacks when encounter is IDLE and cooldown has expired."""
|
||||
template = load_mob_template(goblin_toml)
|
||||
mob = spawn_mob(template, 0, 0)
|
||||
mob.next_action_at = 0.0 # cooldown expired
|
||||
|
||||
encounter = start_encounter(player, mob)
|
||||
player.mode_stack.append("combat")
|
||||
|
||||
await process_mobs(moves)
|
||||
|
||||
# Mob should have attacked — encounter state should be TELEGRAPH
|
||||
assert encounter.state == CombatState.TELEGRAPH
|
||||
assert encounter.current_move is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
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)
|
||||
mob.next_action_at = 0.0
|
||||
|
||||
encounter = start_encounter(player, mob)
|
||||
player.mode_stack.append("combat")
|
||||
|
||||
await process_mobs(moves)
|
||||
|
||||
assert encounter.current_move is not None
|
||||
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
|
||||
):
|
||||
"""Mob skips attack when stamina is too low for any move."""
|
||||
template = load_mob_template(goblin_toml)
|
||||
mob = spawn_mob(template, 0, 0)
|
||||
mob.stamina = 0.0
|
||||
mob.next_action_at = 0.0
|
||||
|
||||
encounter = start_encounter(player, mob)
|
||||
player.mode_stack.append("combat")
|
||||
|
||||
await process_mobs(moves)
|
||||
|
||||
# Mob can't afford any move, encounter stays IDLE
|
||||
assert encounter.state == CombatState.IDLE
|
||||
|
||||
@pytest.mark.asyncio
|
||||
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)
|
||||
mob.next_action_at = time.monotonic() + 100.0 # far in the future
|
||||
|
||||
encounter = start_encounter(player, mob)
|
||||
player.mode_stack.append("combat")
|
||||
|
||||
await process_mobs(moves)
|
||||
|
||||
# Mob should not have attacked
|
||||
assert encounter.state == CombatState.IDLE
|
||||
|
||||
@pytest.mark.asyncio
|
||||
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)
|
||||
mob.next_action_at = 0.0
|
||||
|
||||
# Player is attacker, mob is defender
|
||||
encounter = start_encounter(player, mob)
|
||||
player.mode_stack.append("combat")
|
||||
|
||||
await process_mobs(moves)
|
||||
|
||||
# Mob should now be the attacker
|
||||
assert encounter.attacker is mob
|
||||
assert encounter.defender is player
|
||||
|
||||
@pytest.mark.asyncio
|
||||
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)
|
||||
mob.next_action_at = 0.0
|
||||
|
||||
await process_mobs(moves)
|
||||
|
||||
# No encounter exists for mob
|
||||
assert get_encounter(mob) is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
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)
|
||||
player.mode_stack.append("combat")
|
||||
|
||||
before = time.monotonic()
|
||||
await process_mobs(moves)
|
||||
|
||||
# next_action_at should be ~1 second in the future
|
||||
assert mob.next_action_at >= before + 0.9
|
||||
|
||||
|
||||
class TestMobDefenseAI:
|
||||
@pytest.fixture
|
||||
def punch_right(self, moves):
|
||||
return moves["punch right"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mob_defends_during_telegraph(
|
||||
self, player, goblin_toml, moves, punch_right
|
||||
):
|
||||
"""Mob attempts defense during TELEGRAPH phase."""
|
||||
template = load_mob_template(goblin_toml)
|
||||
# Give the mob defense moves
|
||||
mob = spawn_mob(template, 0, 0)
|
||||
mob.moves = ["punch left", "dodge left", "dodge right"]
|
||||
mob.next_action_at = 0.0
|
||||
|
||||
encounter = start_encounter(player, mob)
|
||||
player.mode_stack.append("combat")
|
||||
|
||||
# Player attacks, putting encounter in TELEGRAPH
|
||||
encounter.attack(punch_right)
|
||||
assert encounter.state == CombatState.TELEGRAPH
|
||||
|
||||
await process_mobs(moves)
|
||||
|
||||
# Mob should have queued a defense
|
||||
assert encounter.pending_defense is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mob_skips_defense_when_already_defending(
|
||||
self, player, goblin_toml, moves, punch_right
|
||||
):
|
||||
"""Mob doesn't double-defend if already has pending_defense."""
|
||||
template = load_mob_template(goblin_toml)
|
||||
mob = spawn_mob(template, 0, 0)
|
||||
mob.moves = ["dodge left", "dodge right"]
|
||||
mob.next_action_at = 0.0
|
||||
|
||||
encounter = start_encounter(player, mob)
|
||||
player.mode_stack.append("combat")
|
||||
encounter.attack(punch_right)
|
||||
|
||||
# Pre-set a defense
|
||||
existing_defense = moves["dodge left"]
|
||||
encounter.pending_defense = existing_defense
|
||||
|
||||
await process_mobs(moves)
|
||||
|
||||
# Should not have changed
|
||||
assert encounter.pending_defense is existing_defense
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mob_no_defense_without_defense_moves(
|
||||
self, player, goblin_toml, moves, punch_right
|
||||
):
|
||||
"""Mob with no defense moves in its list can't defend."""
|
||||
template = load_mob_template(goblin_toml)
|
||||
mob = spawn_mob(template, 0, 0)
|
||||
# Only attack moves
|
||||
mob.moves = ["punch left", "punch right", "sweep"]
|
||||
mob.next_action_at = time.monotonic() + 100.0 # prevent attacking
|
||||
|
||||
encounter = start_encounter(player, mob)
|
||||
player.mode_stack.append("combat")
|
||||
encounter.attack(punch_right)
|
||||
|
||||
await process_mobs(moves)
|
||||
|
||||
# No defense queued
|
||||
assert encounter.pending_defense is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dummy_never_fights_back(
|
||||
self, player, dummy_toml, moves, punch_right
|
||||
):
|
||||
"""Training dummy with empty moves never attacks or defends."""
|
||||
template = load_mob_template(dummy_toml)
|
||||
mob = spawn_mob(template, 0, 0)
|
||||
mob.next_action_at = 0.0
|
||||
|
||||
encounter = start_encounter(player, mob)
|
||||
player.mode_stack.append("combat")
|
||||
|
||||
# Player attacks
|
||||
encounter.attack(punch_right)
|
||||
|
||||
await process_mobs(moves)
|
||||
|
||||
# Dummy should not have defended (empty moves list)
|
||||
assert encounter.pending_defense is None
|
||||
Loading…
Reference in a new issue