297 lines
9.7 KiB
Python
297 lines
9.7 KiB
Python
"""Combat command handlers."""
|
|
|
|
import asyncio
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
from mudlib.combat.encounter import CombatState
|
|
from mudlib.combat.engine import get_encounter, start_encounter
|
|
from mudlib.combat.moves import CombatMove, load_moves
|
|
from mudlib.combat.stamina import check_stamina_cues
|
|
from mudlib.commands import CommandDefinition, register
|
|
from mudlib.player import Player, players
|
|
from mudlib.render.colors import colorize
|
|
from mudlib.render.pov import render_pov
|
|
|
|
# Combat moves will be injected after loading
|
|
combat_moves: dict[str, CombatMove] = {}
|
|
combat_content_dir: Path | None = None
|
|
|
|
|
|
async def do_attack(player: Player, target_args: str, move: CombatMove) -> None:
|
|
"""Core attack logic with a resolved move.
|
|
|
|
Args:
|
|
player: The attacking player
|
|
target_args: Remaining args after move resolution (just the target name)
|
|
move: The resolved combat move
|
|
"""
|
|
encounter = get_encounter(player)
|
|
|
|
# Parse target from args
|
|
target = None
|
|
target_name = target_args.strip()
|
|
if encounter is None and target_name:
|
|
target = players.get(target_name)
|
|
if target is None and player.location is not None:
|
|
from mudlib.mobs import get_nearby_mob
|
|
from mudlib.zone import Zone
|
|
|
|
if isinstance(player.location, Zone):
|
|
target = get_nearby_mob(
|
|
target_name, player.x, player.y, player.location
|
|
)
|
|
|
|
# Check stamina
|
|
if player.stamina < move.stamina_cost:
|
|
await player.send("You don't have enough stamina for that move.\r\n")
|
|
return
|
|
|
|
if encounter is None:
|
|
# Not in combat - need a target
|
|
if target is None:
|
|
await player.send("You need a target to start combat.\r\n")
|
|
return
|
|
|
|
# Check altitude match before starting combat
|
|
if getattr(player, "flying", False) != getattr(target, "flying", False):
|
|
await player.send("You can't reach them from here!\r\n")
|
|
return
|
|
|
|
# Start new encounter
|
|
try:
|
|
encounter = start_encounter(player, target)
|
|
player.mode_stack.append("combat")
|
|
await player.send(f"You engage {target.name} in combat!\r\n")
|
|
|
|
# Send telegraph to defender if they can receive messages
|
|
if hasattr(target, "send") and move.telegraph:
|
|
telegraph = render_pov(move.telegraph, target, player, target)
|
|
telegraph_color = move.telegraph_color
|
|
if telegraph_color:
|
|
color_depth = getattr(target, "color_depth", None)
|
|
telegraph = colorize(
|
|
f"{{{telegraph_color}}}{telegraph}{{/}}", color_depth
|
|
)
|
|
await target.send(f"{telegraph}\r\n")
|
|
|
|
except ValueError as e:
|
|
await player.send(f"Cannot start combat: {e}\r\n")
|
|
return
|
|
else:
|
|
# Already in combat - send telegraph to defender
|
|
if encounter.attacker is player:
|
|
defender = encounter.defender
|
|
else:
|
|
defender = encounter.attacker
|
|
if hasattr(defender, "send") and move.telegraph:
|
|
telegraph = render_pov(move.telegraph, defender, player, defender)
|
|
telegraph_color = move.telegraph_color
|
|
if telegraph_color:
|
|
color_depth = getattr(defender, "color_depth", None)
|
|
telegraph = colorize(
|
|
f"{{{telegraph_color}}}{telegraph}{{/}}", color_depth
|
|
)
|
|
await defender.send(f"{telegraph}\r\n")
|
|
|
|
# Detect switch before attack() modifies state
|
|
switching = encounter.state in (
|
|
CombatState.TELEGRAPH,
|
|
CombatState.WINDOW,
|
|
)
|
|
|
|
# Execute the attack (deducts stamina)
|
|
encounter.attack(move)
|
|
|
|
# Send vitals update immediately after stamina deduction
|
|
from mudlib.gmcp import send_char_vitals
|
|
|
|
send_char_vitals(player)
|
|
|
|
# Check stamina cues after attack cost
|
|
await check_stamina_cues(player)
|
|
|
|
if switching:
|
|
await player.send(f"You switch to {move.name}!\r\n")
|
|
else:
|
|
await player.send(f"You use {move.name}!\r\n")
|
|
|
|
|
|
async def do_defend(player: Player, _args: str, move: CombatMove) -> None:
|
|
"""Core defense logic with a resolved move.
|
|
|
|
Works both in and outside combat. Applies a recovery lock
|
|
(based on timing_window_ms) so defenses have commitment.
|
|
|
|
Args:
|
|
player: The defending player
|
|
_args: Unused (defense moves don't take a target)
|
|
move: The resolved combat move
|
|
"""
|
|
# Check stamina
|
|
if player.stamina < move.stamina_cost:
|
|
await player.send("You don't have enough stamina for that move.\r\n")
|
|
return
|
|
|
|
player.stamina -= move.stamina_cost
|
|
|
|
# Send vitals update immediately after stamina deduction
|
|
from mudlib.gmcp import send_char_vitals
|
|
|
|
send_char_vitals(player)
|
|
|
|
# Check stamina cues after defense cost
|
|
await check_stamina_cues(player)
|
|
|
|
# If in combat, queue the defense on the encounter
|
|
encounter = get_encounter(player)
|
|
if encounter is not None:
|
|
encounter.defend(move)
|
|
|
|
# Broadcast to nearby players
|
|
from mudlib.commands.movement import send_nearby_message
|
|
|
|
await send_nearby_message(
|
|
player,
|
|
player.x,
|
|
player.y,
|
|
f"{player.name} {move.command}s!\r\n",
|
|
)
|
|
|
|
# Commitment: block for the timing window (inputs queue naturally)
|
|
await asyncio.sleep(move.timing_window_ms / 1000.0)
|
|
|
|
if encounter is not None:
|
|
await player.send(f"You {move.name}!\r\n")
|
|
else:
|
|
await player.send(f"You {move.command} the air!\r\n")
|
|
|
|
|
|
def make_direct_handler(move: CombatMove, handler_fn):
|
|
"""Create a handler bound to a specific move.
|
|
|
|
Used for simple moves (roundhouse) and variant aliases (pl, pr).
|
|
Args are just the target name.
|
|
"""
|
|
|
|
async def handler(player: Player, args: str) -> None:
|
|
await handler_fn(player, args, move)
|
|
|
|
return handler
|
|
|
|
|
|
def make_variant_handler(
|
|
base_name: str, variant_moves: dict[str, CombatMove], handler_fn
|
|
):
|
|
"""Create a handler for a move with directional variants.
|
|
|
|
Used for base commands like "punch" where the first arg is the direction.
|
|
"""
|
|
|
|
async def handler(player: Player, args: str) -> None:
|
|
parts = args.strip().split(maxsplit=1)
|
|
if not parts:
|
|
variants = "/".join(variant_moves.keys())
|
|
await player.send(f"{base_name.capitalize()} which way? ({variants})\r\n")
|
|
return
|
|
|
|
variant_key = parts[0].lower()
|
|
|
|
# Try exact match first
|
|
move = variant_moves.get(variant_key)
|
|
if move is None:
|
|
# Fall back to prefix matching
|
|
matches = [k for k in variant_moves if k.startswith(variant_key)]
|
|
if len(matches) == 1:
|
|
move = variant_moves[matches[0]]
|
|
elif len(matches) > 1:
|
|
variants = "/".join(sorted(matches))
|
|
await player.send(
|
|
f"Ambiguous {base_name} direction: {variant_key}. ({variants})\r\n"
|
|
)
|
|
return
|
|
else:
|
|
variants = "/".join(variant_moves.keys())
|
|
await player.send(
|
|
f"Unknown {base_name} direction: {variant_key}. Try: {variants}\r\n"
|
|
)
|
|
return
|
|
|
|
target_args = parts[1] if len(parts) > 1 else ""
|
|
await handler_fn(player, target_args, move)
|
|
|
|
return handler
|
|
|
|
|
|
def register_moves(moves: list[CombatMove]) -> None:
|
|
"""Register a list of combat moves as commands.
|
|
|
|
Args:
|
|
moves: List of combat moves to register
|
|
"""
|
|
# Group variant moves by their base command
|
|
variant_groups: dict[str, dict[str, CombatMove]] = defaultdict(dict)
|
|
simple_moves: list[CombatMove] = []
|
|
registered_names: set[str] = set()
|
|
|
|
for move in moves:
|
|
if move.name in registered_names:
|
|
continue
|
|
registered_names.add(move.name)
|
|
|
|
if move.variant:
|
|
variant_groups[move.command][move.variant] = move
|
|
else:
|
|
simple_moves.append(move)
|
|
|
|
# Register simple moves (roundhouse, sweep, duck, jump)
|
|
for move in simple_moves:
|
|
handler_fn = do_attack if move.move_type == "attack" else do_defend
|
|
mode = "*"
|
|
action = "Attack" if move.move_type == "attack" else "Defend"
|
|
register(
|
|
CommandDefinition(
|
|
name=move.name,
|
|
handler=make_direct_handler(move, handler_fn),
|
|
aliases=[],
|
|
mode=mode,
|
|
help=f"{action} with {move.name}",
|
|
)
|
|
)
|
|
|
|
# Register variant moves (punch, dodge, parry)
|
|
for base_name, variants in variant_groups.items():
|
|
# Determine type from first variant
|
|
first_variant = next(iter(variants.values()))
|
|
handler_fn = do_attack if first_variant.move_type == "attack" else do_defend
|
|
mode = "*"
|
|
|
|
# Register base command with variant handler (e.g. "punch")
|
|
action = "Attack" if first_variant.move_type == "attack" else "Defend"
|
|
register(
|
|
CommandDefinition(
|
|
name=base_name,
|
|
handler=make_variant_handler(base_name, variants, handler_fn),
|
|
aliases=[],
|
|
mode=mode,
|
|
help=f"{action} with {base_name}",
|
|
)
|
|
)
|
|
|
|
|
|
def register_combat_commands(content_dir: Path) -> None:
|
|
"""Load and register all combat moves as commands.
|
|
|
|
Args:
|
|
content_dir: Path to directory containing combat move TOML files
|
|
"""
|
|
global combat_moves, combat_content_dir
|
|
|
|
# Save content directory for use by edit command
|
|
combat_content_dir = content_dir
|
|
|
|
# Load all moves from content directory
|
|
combat_moves = load_moves(content_dir)
|
|
|
|
# Register all loaded moves
|
|
register_moves(list(combat_moves.values()))
|