214 lines
7.2 KiB
Python
214 lines
7.2 KiB
Python
"""Interactive fiction session management via dfrotz subprocess."""
|
|
|
|
import asyncio
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from mudlib.player import Player
|
|
|
|
|
|
@dataclass
|
|
class IFResponse:
|
|
"""Response from IF session input handling."""
|
|
|
|
output: str
|
|
done: bool
|
|
|
|
|
|
class IFSession:
|
|
"""Manages an interactive fiction session via dfrotz subprocess."""
|
|
|
|
def __init__(self, player, story_path: str, game_name: str = ""):
|
|
self.player = player
|
|
self.story_path = story_path
|
|
self.game_name = game_name or Path(story_path).stem
|
|
self.process: asyncio.subprocess.Process | None = None
|
|
# data/ directory is at project root (2 levels up from src/mudlib/)
|
|
self._data_dir = Path(__file__).resolve().parents[2] / "data"
|
|
# Track whether we've saved (prevents double-save on quit+stop)
|
|
self._saved = False
|
|
|
|
@property
|
|
def save_path(self) -> Path:
|
|
"""Return path to save file for this player/game combo."""
|
|
# Sanitize player name to prevent path traversal attacks
|
|
# Account creation doesn't validate names, so defensive check here
|
|
safe_name = re.sub(r"[^a-zA-Z0-9_-]", "_", self.player.name)
|
|
return self._data_dir / "if_saves" / safe_name / f"{self.game_name}.qzl"
|
|
|
|
def _ensure_save_dir(self) -> None:
|
|
"""Create save directory if it doesn't exist."""
|
|
self.save_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
async def start(self) -> str:
|
|
"""Spawn dfrotz and return intro text."""
|
|
self.process = await asyncio.create_subprocess_exec(
|
|
"dfrotz",
|
|
"-p", # plain ASCII, no formatting
|
|
"-w",
|
|
"80", # screen width
|
|
"-m", # no MORE prompts
|
|
self.story_path,
|
|
stdin=asyncio.subprocess.PIPE,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
# Read intro text until we see the prompt
|
|
intro = await self._read_response()
|
|
return intro
|
|
|
|
async def handle_input(self, text: str) -> IFResponse:
|
|
"""Handle player input. Route to dfrotz or handle escape commands."""
|
|
# Check for escape commands (:: prefix)
|
|
if text.lower() == "::quit":
|
|
# Auto-save before quitting
|
|
await self._do_save()
|
|
return IFResponse(output="game saved.", done=True)
|
|
|
|
if text.lower() == "::help":
|
|
help_text = """escape commands:
|
|
::quit - exit the game
|
|
::save - save game progress
|
|
::help - show this help"""
|
|
return IFResponse(output=help_text, done=False)
|
|
|
|
if text.lower() == "::save":
|
|
confirmation = await self._do_save()
|
|
return IFResponse(output=confirmation, done=False)
|
|
|
|
# Regular game input - send to dfrotz
|
|
# Reset saved flag since game state has changed
|
|
self._saved = False
|
|
if self.process and self.process.stdin:
|
|
stripped = text.strip()
|
|
self.process.stdin.write(f"{stripped}\n".encode())
|
|
await self.process.stdin.drain()
|
|
|
|
# Read response
|
|
output = await self._read_response()
|
|
return IFResponse(output=output, done=False)
|
|
|
|
# Process not running
|
|
return IFResponse(output="error: game not running", done=True)
|
|
|
|
async def stop(self):
|
|
"""Terminate dfrotz process."""
|
|
if self.process and self.process.returncode is None:
|
|
# Auto-save before terminating
|
|
await self._do_save()
|
|
self.process.terminate()
|
|
await self.process.wait()
|
|
|
|
async def _do_save(self) -> str:
|
|
"""Save game state to disk. Returns confirmation message."""
|
|
# Skip if already saved (prevents double-save on quit+stop)
|
|
if self._saved:
|
|
return "already saved"
|
|
|
|
if not self.process or not self.process.stdin:
|
|
return "error: game not running"
|
|
|
|
try:
|
|
# Ensure save directory exists
|
|
self._ensure_save_dir()
|
|
|
|
# Send "save" command to dfrotz
|
|
self.process.stdin.write(b"save\n")
|
|
await self.process.stdin.drain()
|
|
|
|
# Read filename prompt
|
|
await self._read_response()
|
|
|
|
# Send save file path
|
|
save_path_str = str(self.save_path)
|
|
self.process.stdin.write(f"{save_path_str}\n".encode())
|
|
await self.process.stdin.drain()
|
|
|
|
# Read confirmation
|
|
confirmation = await self._read_response()
|
|
|
|
# Mark as saved
|
|
self._saved = True
|
|
return confirmation
|
|
except Exception as e:
|
|
return f"error: save failed ({e})"
|
|
|
|
async def _do_restore(self) -> str:
|
|
"""Restore game state from disk. Returns game text or empty string."""
|
|
# Check if save file exists
|
|
if not self.save_path.exists():
|
|
return ""
|
|
|
|
if not self.process or not self.process.stdin:
|
|
return ""
|
|
|
|
try:
|
|
# Send "restore" command to dfrotz
|
|
self.process.stdin.write(b"restore\n")
|
|
await self.process.stdin.drain()
|
|
|
|
# Read filename prompt
|
|
await self._read_response()
|
|
|
|
# Send save file path
|
|
save_path_str = str(self.save_path)
|
|
self.process.stdin.write(f"{save_path_str}\n".encode())
|
|
await self.process.stdin.drain()
|
|
|
|
# Read game text response
|
|
game_text = await self._read_response()
|
|
return game_text
|
|
except Exception:
|
|
# If restore fails, return empty string (player starts fresh)
|
|
return ""
|
|
|
|
async def _read_response(self) -> str:
|
|
"""Read dfrotz output until the '>' prompt appears."""
|
|
if not self.process or not self.process.stdout:
|
|
return ""
|
|
|
|
buf = b""
|
|
idle_timeout = 0.1 # return quickly once data stops flowing
|
|
loop = asyncio.get_running_loop()
|
|
overall_deadline = loop.time() + 5.0
|
|
|
|
try:
|
|
while True:
|
|
remaining = overall_deadline - loop.time()
|
|
if remaining <= 0:
|
|
break
|
|
timeout = min(idle_timeout, remaining)
|
|
try:
|
|
chunk = await asyncio.wait_for(
|
|
self.process.stdout.read(1024), timeout=timeout
|
|
)
|
|
except TimeoutError:
|
|
# No data for idle_timeout - dfrotz is done talking
|
|
break
|
|
if not chunk:
|
|
break
|
|
buf += chunk
|
|
|
|
# Check for prompt at end of buffer
|
|
text = buf.decode("latin-1")
|
|
if text.endswith("\n>"):
|
|
text = text[:-2]
|
|
return text.rstrip()
|
|
if text == ">":
|
|
return ""
|
|
except TimeoutError:
|
|
pass
|
|
|
|
return buf.decode("latin-1").rstrip()
|
|
|
|
|
|
async def broadcast_to_spectators(player: "Player", message: str) -> None:
|
|
"""Send message to all other players at the same location."""
|
|
from mudlib.player import players
|
|
|
|
for other in players.values():
|
|
if other.name != player.name and other.x == player.x and other.y == player.y:
|
|
await other.send(message)
|