diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0fe6a0f --- /dev/null +++ b/.gitignore @@ -0,0 +1,39 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +*.egg-info/ +.eggs/ +dist/ +build/ + +# Virtual environment +venv/ +.venv/ +env/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo + +# Data and logs (runtime generated) +data/ +*.log + +# Database files +*.db + +# Credentials - NEVER commit +config.yaml + +# OS files +.DS_Store +Thumbs.db + +# Coverage reports +htmlcov/ +.coverage +.coverage.* diff --git a/MUD_SERVERS.md b/MUD_SERVERS.md new file mode 100644 index 0000000..91e5f0b --- /dev/null +++ b/MUD_SERVERS.md @@ -0,0 +1,151 @@ +# MUD Test Servers - Quick Start Guide + +## ✅ L'applicazione funziona! + +L'errore che vedi è **normale** - stai usando il config di esempio con `mud.example.com` che non esiste. + +## MUD Pubblici per Testing + +Ecco alcuni MUD pubblici che puoi usare per testare l'applicazione: + +### 1. Aardwolf MUD (Consigliato per testing) +```yaml +mud: + host: "aardmud.org" + port: 23 + username: "" # Crea account al primo login + password: "" +``` + +**Caratteristiche:** +- Molto popolare e stabile +- Accetta nuovi giocatori +- Buona documentazione +- Supporta ANSI colors + +### 2. Discworld MUD +```yaml +mud: + host: "discworld.starturtle.net" + port: 23 + username: "" + password: "" +``` + +### 3. Alter Aeon +```yaml +mud: + host: "alteraeon.com" + port: 3010 + username: "" + password: "" +``` + +### 4. Materia Magica +```yaml +mud: + host: "materiamagica.com" + port: 4000 + username: "" + password: "" +``` + +## Come Configurare + +### Opzione 1: Modifica config.yaml + +Apri `config.yaml` e modifica la sezione `mud`: + +```yaml +mud: + host: "aardmud.org" # Cambia questo + port: 23 # E questo + username: "" # Lascia vuoto per nuovo account + password: "" +``` + +### Opzione 2: Crea config personalizzato + +```powershell +# Copia il config di esempio +Copy-Item config.yaml test_config.yaml + +# Modifica test_config.yaml con il tuo editor preferito +# Poi esegui con: +python -m src.main --config test_config.yaml +``` + +## Test Rapido con Aardwolf + +1. **Modifica config.yaml**: +```yaml +mud: + host: "aardmud.org" + port: 23 + username: "" + password: "" +``` + +2. **Esegui l'applicazione**: +```powershell +.\venv\Scripts\Activate.ps1 +python -m src.main +``` + +3. **Cosa aspettarsi**: + - Connessione al server ✅ + - Ricezione del messaggio di benvenuto + - Prompt per creazione personaggio + - (L'AI non è ancora implementata completamente, quindi vedrai solo i log) + +## Modalità Debug + +Per vedere più dettagli durante la connessione, modifica in `config.yaml`: + +```yaml +logging: + level: "DEBUG" # Cambia da INFO a DEBUG + console: true +``` + +## Creazione Account Manuale + +Se vuoi prima creare un account manualmente: + +```powershell +# Usa telnet per connetterti +telnet aardmud.org 23 + +# Oppure usa un client MUD come MUSHclient o Mudlet +``` + +Poi inserisci le credenziali in `config.yaml`. + +## Prossimi Passi + +Una volta connesso con successo: + +1. ✅ **Connessione funzionante** - Vedrai i log di connessione +2. ⏭️ **Implementare Character Creator** - Per creare automaticamente il personaggio +3. ⏭️ **Implementare Text Parser** - Per interpretare i comandi del MUD +4. ⏭️ **Implementare Explorer** - Per iniziare l'esplorazione + +## Troubleshooting + +### Firewall +Se hai problemi di connessione, verifica che il firewall non blocchi le connessioni Telnet (porta 23 o 4000). + +### Timeout +Alcuni MUD potrebbero avere timeout. Aumenta il timeout in `telnet_client.py` se necessario. + +### ANSI Colors +Se vedi caratteri strani, il parser ANSI potrebbe aver bisogno di aggiustamenti per quel MUD specifico. + +## Note Importanti + +⚠️ **Rispetta le regole del MUD**: Molti MUD hanno regole contro i bot. Usa questa applicazione solo per: +- Testing e sviluppo +- MUD che permettono bot +- Scopi educativi + +⚠️ **Non abusare**: Implementa sempre i delay umani e non sovraccaricare il server. diff --git a/README.md b/README.md index 903bb03..38b309e 100644 --- a/README.md +++ b/README.md @@ -1,134 +1,134 @@ -# MUD AI Player - -An autonomous AI player for MUD (Multi-User Dungeon) text-based games. - -## Features - -- 🤖 **Autonomous Gameplay**: Simulates a human player from character creation to endgame -- 🗺️ **Auto-Mapping**: Builds a complete map of the MUD world with hybrid storage (SQLite + Memvid) -- ⚔️ **Combat AI**: Intelligent combat decisions, target selection, and survival tactics -- 📜 **Quest System**: Automatically detects, plans, and completes quests -- 🔍 **Semantic Search**: Find locations using natural language queries -- 📊 **Progression Optimizer**: Finds optimal grinding spots and equipment upgrades - -## Architecture - -The application is built with a modular architecture: - -- **Core**: Telnet client, ANSI parser, text parser, game state -- **Modules**: - - Character Creator: Stats rolling, class/race selection - - Auto-Mapper: World mapping with hybrid storage (SQLite + Memvid) - - Decision Engine: AI decision-making, combat AI, goal management - - Quest System: Quest detection, planning, execution - - Progression Optimizer: Grinding optimization, equipment analysis - -## Installation - -### Using Poetry (Recommended) - -```bash -# Install Poetry if not already installed -pip install poetry - -# Install dependencies -poetry install - -# Activate virtual environment -poetry shell -``` - -### Using pip - -```bash -# Create virtual environment -python -m venv venv -source venv/bin/activate # On Windows: venv\Scripts\activate - -# Install dependencies -pip install -r requirements.txt -``` - -## Configuration - -1. Copy the example configuration: - ```bash - cp examples/config.example.yaml config.yaml - ``` - -2. Edit `config.yaml` with your MUD server details: - ```yaml - mud: - host: "your-mud-server.com" - port: 4000 - username: "your_username" - password: "your_password" - ``` - -## Usage - -```bash -# Run the AI player -python -m src.main - -# Run with custom config -python -m src.main --config custom_config.yaml - -# Run tests -pytest tests/ -v - -# Run with coverage -pytest --cov=src --cov-report=html -``` - -## Project Structure - -``` -mud-ai-player/ -├── src/ # Source code -│ ├── core/ # Core components -│ ├── modules/ # Specialized modules -│ ├── utils/ # Utilities -│ └── config/ # Configuration management -├── data/ # Persistent storage -│ ├── maps/ # World maps (SQLite + Memvid) -│ ├── characters/ # Character saves -│ └── logs/ # Game logs -├── tests/ # Test suite -├── docs/ # Documentation -└── examples/ # Example configurations -``` - -## Development - -```bash -# Format code -black src/ tests/ - -# Lint code -ruff check src/ tests/ - -# Run tests with watch -pytest-watch -``` - -## Documentation - -See the `docs/` directory for detailed documentation: - -- [Architecture](docs/architecture.md) -- [Configuration Guide](docs/configuration.md) -- [Development Guide](docs/development.md) -- [API Reference](docs/api_reference.md) - -## License - -MIT License - -## Contributing - -Contributions are welcome! Please read the development guide before submitting PRs. - -## Disclaimer - -This is an educational project. Please respect the rules and terms of service of any MUD you connect to. Some MUDs prohibit automated players. +# MUD AI Player + +An autonomous AI player for MUD (Multi-User Dungeon) text-based games. + +## Features + +- 🤖 **Autonomous Gameplay**: Simulates a human player from character creation to endgame +- 🗺️ **Auto-Mapping**: Builds a complete map of the MUD world with hybrid storage (SQLite + Memvid) +- ⚔️ **Combat AI**: Intelligent combat decisions, target selection, and survival tactics +- 📜 **Quest System**: Automatically detects, plans, and completes quests +- 🔍 **Semantic Search**: Find locations using natural language queries +- 📊 **Progression Optimizer**: Finds optimal grinding spots and equipment upgrades + +## Architecture + +The application is built with a modular architecture: + +- **Core**: Telnet client, ANSI parser, text parser, game state +- **Modules**: + - Character Creator: Stats rolling, class/race selection + - Auto-Mapper: World mapping with hybrid storage (SQLite + Memvid) + - Decision Engine: AI decision-making, combat AI, goal management + - Quest System: Quest detection, planning, execution + - Progression Optimizer: Grinding optimization, equipment analysis + +## Installation + +### Using Poetry (Recommended) + +```bash +# Install Poetry if not already installed +pip install poetry + +# Install dependencies +poetry install + +# Activate virtual environment +poetry shell +``` + +### Using pip + +```bash +# Create virtual environment +python -m venv venv +source venv/bin/activate # On Windows: venv\Scripts\activate + +# Install dependencies +pip install -r requirements.txt +``` + +## Configuration + +1. Copy the example configuration: + ```bash + cp examples/config.example.yaml config.yaml + ``` + +2. Edit `config.yaml` with your MUD server details: + ```yaml + mud: + host: "your-mud-server.com" + port: 4000 + username: "your_username" + password: "your_password" + ``` + +## Usage + +```bash +# Run the AI player +python -m src.main + +# Run with custom config +python -m src.main --config custom_config.yaml + +# Run tests +pytest tests/ -v + +# Run with coverage +pytest --cov=src --cov-report=html +``` + +## Project Structure + +``` +mud-ai-player/ +├── src/ # Source code +│ ├── core/ # Core components +│ ├── modules/ # Specialized modules +│ ├── utils/ # Utilities +│ └── config/ # Configuration management +├── data/ # Persistent storage +│ ├── maps/ # World maps (SQLite + Memvid) +│ ├── characters/ # Character saves +│ └── logs/ # Game logs +├── tests/ # Test suite +├── docs/ # Documentation +└── examples/ # Example configurations +``` + +## Development + +```bash +# Format code +black src/ tests/ + +# Lint code +ruff check src/ tests/ + +# Run tests with watch +pytest-watch +``` + +## Documentation + +See the `docs/` directory for detailed documentation: + +- [Architecture](docs/architecture.md) +- [Configuration Guide](docs/configuration.md) +- [Development Guide](docs/development.md) +- [API Reference](docs/api_reference.md) + +## License + +MIT License + +## Contributing + +Contributions are welcome! Please read the development guide before submitting PRs. + +## Disclaimer + +This is an educational project. Please respect the rules and terms of service of any MUD you connect to. Some MUDs prohibit automated players. diff --git a/TODO.md b/TODO.md new file mode 100644 index 0000000..efb1d0b --- /dev/null +++ b/TODO.md @@ -0,0 +1,82 @@ +# MUD AI Player - Implementation Roadmap + +This document outlines the tasks required to build a complete, autonomous AI player for LPMUDs. + +## 🟢 Phase 1: Core Connectivity & Basic Setup (Current Status) + +- [x] **Telnet Client**: Basic connection and disconnection handling. +- [x] **Configuration**: YAML-based config for credentials and server info. +- [x] **Automatic Login**: + - [x] Username/Password automtation. + - [x] "Take over existing session" handling. +- [x] **KeepAlive Manager**: Prevent idle disconnects. +- [ ] **Logging**: + - [x] Setup basic logging. + - [ ] Implement strict `game.log` rotation and archiving structure. + +## 🟡 Phase 2: Perception (Parsing & Sensing) + +The bot needs to "see" the world effectively. + +- [ ] **ANSI Parsing**: + - [ ] Refine `ANSIParser` to fully strip color codes for logic but preserve them for UI if needed. + - [ ] Detect prompt lines vs. content lines. +- [ ] **Status Parsing (Vitals)**: + - [ ] Implement regex to capture HP/MP/SP from prompt (e.g., `hp: 350/350`). + - [ ] Update `GameState.character` with real-time stats. +- [ ] **Room Parsing (Vision)**: + - [ ] Identify Room Title (usually colored/bold). + - [ ] Identify Room Description (multiline text). + - [ ] Identify Exits (e.g., `[ Exits: n s w ]`). + - [ ] Identify NPCs/Mobs/Items in the room. +- [ ] **Inventory & Equipment**: + - [ ] Implement `inv` command parser. + - [ ] Implement `eq` command parser. + +## 🟠 Phase 3: World Modeling (Mapping) + +The bot needs to know where it is and where it has been. + +- [ ] **Room Hashing**: + - [ ] Create unique hash for rooms based on Title + Description + Exits (to handle non-unique IDs). +- [ ] **Graph Database**: + - [ ] Integrate `NetworkX` or SQLite for storing the world map. + - [ ] Store nodes (Rooms) and edges (Exits). +- [ ] **Pathfinding**: + - [ ] Implement A* algorithm to find shortest path between known rooms. + - [ ] Implement "Breadth-First Search" (BFS) for exploration mode. + +## 🔴 Phase 4: Decision Engine (The Brain) + +The bot needs to decide what to do next. + +- [ ] **Action Queue System**: + - [ ] Priority queue for actions (e.g., `HEAL` > `ATTACK` > `LOOT` > `MOVE`). +- [ ] **Behavior Trees / State Machine**: + - [ ] **IdleState**: Heal, Buff, Check Inventory. + - [ ] **ExploreState**: Pick unknown exit, Move. + - [ ] **CombatState**: + - [ ] Auto-target hostile mobs. + - [ ] Execute combat rotation (skills/spells). + - [ ] Flee trigger (if HP < Threshold). + - [ ] **LootState**: reliable looting of corpses. +- [ ] **LLM Integration (Optional/Hybrid)**: + - [ ] Use LLM for "roleplay" responses to other players. + - [ ] Use LLM for "quest parsing" (understanding vague instructions). + +## 🔵 Phase 5: Humanization + +Make the bot undetectable and less robotic. + +- [ ] **Reaction Time Jitter**: + - [ ] Randomize delays between commands (e.g., 200ms - 1500ms). +- [ ] **Typo Generator**: + - [ ] Occasionally misspell commands and correct them (optional). +- [ ] **Schedule System**: + - [ ] Define "active hours" and "sleep hours". + +## 🟣 Phase 6: Testing & Metrics + +- [ ] **Unit Tests**: Coverage for parsers (critical). +- [ ] **Integration Tests**: Mock MUD server for offline testing. +- [ ] **Performance Profiling**: Ensure graph traversal doesn't lag the bot. diff --git a/VENV_SETUP.md b/VENV_SETUP.md new file mode 100644 index 0000000..40dd290 --- /dev/null +++ b/VENV_SETUP.md @@ -0,0 +1,160 @@ +# Virtual Environment Setup + +## ✅ Virtual Environment Creato + +Il virtual environment è stato creato con successo in `venv/` e tutte le dipendenze sono state installate. + +## Come Attivare il Virtual Environment + +### Windows PowerShell +```powershell +.\venv\Scripts\Activate.ps1 +``` + +### Windows CMD +```cmd +.\venv\Scripts\activate.bat +``` + +### Linux/Mac +```bash +source venv/bin/activate +``` + +## Verifica Installazione + +Dopo aver attivato il venv, verifica che tutto sia installato: + +```powershell +# Attiva venv +.\venv\Scripts\Activate.ps1 + +# Verifica pacchetti installati +pip list + +# Verifica versione Python +python --version +``` + +## Pacchetti Installati + +Tutti i pacchetti da `requirements.txt` sono stati installati: + +- ✅ `telnetlib3` - Client Telnet asincrono +- ✅ `networkx` - Gestione grafi per auto-mapping +- ✅ `sqlalchemy` - ORM per database +- ✅ `pydantic` - Validazione configurazione +- ✅ `rich` - Output formattato console +- ✅ `memvid` - Encoding QR-in-MP4 +- ✅ `faiss-cpu` - Vector similarity search +- ✅ `sentence-transformers` - Text embeddings +- ✅ `torch` - Deep learning framework (dipendenza) +- ✅ `pytest` - Testing framework +- ✅ `pytest-asyncio` - Async testing +- ✅ `pytest-cov` - Coverage reporting + +## Eseguire l'Applicazione + +```powershell +# Attiva venv +.\venv\Scripts\Activate.ps1 + +# Esegui l'applicazione +python -m src.main + +# Oppure con config personalizzato +python -m src.main --config my_config.yaml +``` + +## Eseguire i Test + +```powershell +# Attiva venv +.\venv\Scripts\Activate.ps1 + +# Esegui tutti i test +pytest tests/ -v + +# Esegui con coverage +pytest --cov=src --cov-report=html + +# Esegui test specifico +pytest tests/test_parser.py -v +``` + +## Disattivare il Virtual Environment + +Quando hai finito di lavorare: + +```powershell +deactivate +``` + +## Aggiungere Nuove Dipendenze + +Se devi aggiungere nuove dipendenze: + +```powershell +# Attiva venv +.\venv\Scripts\Activate.ps1 + +# Installa nuovo pacchetto +pip install nome-pacchetto + +# Aggiorna requirements.txt +pip freeze > requirements.txt +``` + +## Ricreare il Virtual Environment + +Se hai problemi, puoi ricreare il venv: + +```powershell +# Rimuovi venv esistente +Remove-Item -Recurse -Force venv + +# Crea nuovo venv +python -m venv venv + +# Attiva +.\venv\Scripts\Activate.ps1 + +# Reinstalla dipendenze +pip install -r requirements.txt +``` + +## Note + +- Il virtual environment è **gitignored** (vedi `.gitignore`) +- Ogni sviluppatore deve creare il proprio venv locale +- Non committare mai la cartella `venv/` su git +- Usa sempre il venv quando lavori sul progetto + +## Troubleshooting + +### Errore "Execution Policy" + +Se ricevi un errore di execution policy in PowerShell: + +```powershell +Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser +``` + +### Pacchetti Mancanti + +Se mancano pacchetti: + +```powershell +.\venv\Scripts\Activate.ps1 +pip install -r requirements.txt +``` + +### Versione Python Sbagliata + +Assicurati di usare Python 3.10+: + +```powershell +python --version +``` + +Se hai una versione più vecchia, installa Python 3.10 o superiore. diff --git a/docs/architecture.md b/docs/architecture.md new file mode 100644 index 0000000..6e1c884 --- /dev/null +++ b/docs/architecture.md @@ -0,0 +1,151 @@ +# MUD AI Player - Architecture + +## Overview + +The MUD AI Player is designed with a modular architecture that separates concerns into specialized components. This document describes the high-level architecture and design decisions. + +## System Architecture + +``` +┌─────────────────────────────────────────────────────────┐ +│ Main Application │ +│ (main.py) │ +└────────────────────┬────────────────────────────────────┘ + │ + ┌────────────┴────────────┐ + │ │ +┌───────▼────────┐ ┌────────▼────────┐ +│ Core Layer │ │ Modules Layer │ +│ │ │ │ +│ - Telnet │ │ - Character │ +│ - Parsers │ │ - Mapper │ +│ - Game State │ │ - Decision │ +└────────┬───────┘ │ - Quest │ + │ │ - Progression │ + │ └────────┬────────┘ + │ │ + └───────────┬───────────┘ + │ + ┌──────────▼──────────┐ + │ Storage Layer │ + │ │ + │ - SQLite (Active) │ + │ - Memvid (Archive) │ + │ - FAISS (Search) │ + └─────────────────────┘ +``` + +## Core Components + +### Telnet Client +- Asynchronous connection to MUD server +- Handles reconnection and error recovery +- Rate limiting for commands + +### Parsers +- **ANSI Parser**: Strips/interprets color codes +- **Text Parser**: Extracts structured data from MUD output +- Pattern matching for rooms, NPCs, items, status + +### Game State +- Central state management +- Character, room, inventory, quests +- Event system for state changes + +## Specialized Modules + +### Auto-Mapper (Hybrid Storage) + +**Layer 1: Active Storage** +- SQLite database for real-time data +- NetworkX graph in memory +- <5ms insert latency +- Fast pathfinding + +**Layer 2: Historical Storage** +- Memvid encoding (QR-in-MP4) +- 50-100x compression +- Periodic snapshots +- Long-term archival + +**Layer 3: Semantic Search** +- FAISS vector search +- Sentence transformers +- Natural language queries + +### Decision Engine +- Goal-based AI +- State machine for behaviors +- Pluggable architecture (rule-based vs LLM) + +### Combat AI +- Target selection +- Skill rotation +- Flee logic +- Health management + +## Data Flow + +``` +MUD Server + ↓ +Telnet Client + ↓ +ANSI Parser → Text Parser + ↓ +Game State Update + ↓ +Decision Engine + ↓ +Action Selection + ↓ +Command Execution + ↓ +MUD Server +``` + +## Design Patterns + +### Strategy Pattern +- AI engine (rule-based vs LLM) +- Exploration strategy (DFS vs BFS) +- Combat strategy (aggressive vs cautious) + +### Observer Pattern +- Game state changes trigger module updates +- Event-driven architecture + +### Repository Pattern +- Active storage for real-time data +- Historical storage for archival +- Abstraction over storage details + +## Performance Considerations + +- **Async/Await**: All I/O operations are non-blocking +- **In-Memory Graph**: Fast pathfinding without DB queries +- **Lazy Loading**: Historical data loaded on demand +- **Connection Pooling**: Reuse database connections + +## Extensibility + +The architecture is designed for easy extension: + +1. **New Modules**: Add to `src/modules/` +2. **New Parsers**: Extend `TextParser` for MUD-specific patterns +3. **New AI Engines**: Implement `DecisionEngine` interface +4. **New Storage**: Implement storage interface + +## Security & Safety + +- Credentials in config.yaml (gitignored) +- Rate limiting to avoid server bans +- Human behavior simulation +- Graceful error handling + +## Future Enhancements + +- Multi-character support +- Web dashboard for monitoring +- Plugin system for custom behaviors +- Distributed mapping (share maps between instances) diff --git a/examples/config.example.yaml b/examples/config.example.yaml new file mode 100644 index 0000000..9b5be08 --- /dev/null +++ b/examples/config.example.yaml @@ -0,0 +1,54 @@ +# MUD AI Player Configuration Example +# Copy this file to config.yaml and customize + +mud: + host: "mud.example.com" + port: 4000 + mud_type: "dead_souls" # dead_souls, aardwolf, etc. + username: "your_username" + password: "your_password" + +character: + name: "AIPlayer" + preferred_class: "warrior" + preferred_race: "human" + play_style: "balanced" + +ai: + engine: "rule_based" + llm_model: "llama3.2" + decision_delay_ms: [200, 2000] + exploration_strategy: "depth_first" + +mapper: + active_storage: + db_path: "data/maps/active/world.db" + graph_cache_size: 10000 + + historical_storage: + archive_dir: "data/maps/archive" + snapshot_triggers: + room_count: 500 + session_end: true + time_interval_hours: 24 + compression: + fps: 30 + codec: "h264" + + semantic_search: + model: "all-MiniLM-L6-v2" + top_k: 5 + similarity_threshold: 0.7 + +behavior: + reaction_time_min_ms: 200 + reaction_time_max_ms: 2000 + pause_probability: 0.05 + typo_probability: 0.0 + +logging: + level: "INFO" + file: "data/logs/game.log" + console: true + max_file_size_mb: 10 + backup_count: 5 diff --git a/project.md b/project.md new file mode 100644 index 0000000..1f843c7 --- /dev/null +++ b/project.md @@ -0,0 +1,306 @@ + + +# puoi produrre un PR.MD per chiedere ad un agente tipo ralph di produrre un automapper con queste caratteristiche ? + +```markdown +# PR: Implementazione Automapper MUD con Archivio Ibrido SQLite/Memvid + +## Descrizione +Implementare un modulo automapper per un client MUD che costruisca automaticamente una mappa del mondo di gioco durante l'esplorazione, con persistenza ibrida: SQLite per la memoria attiva real-time e Memvid per l'archivio storico compresso. + +## Obiettivi + +- [ ] Parsing delle descrizioni stanze dal protocollo Telnet MUD +- [ ] Costruzione incrementale del grafo di navigazione (stanze = nodi, direzioni = archi) +- [ ] Identificazione univoca delle stanze (room fingerprinting) +- [ ] Auto-mapping durante l'esplorazione senza intervento umano +- [ ] Pathfinding A* tra stanze note +- [ ] Storage layer ibrido: SQLite (attivo) + Memvid (storico) +- [ ] Sistema di ricerca semantica sulle descrizioni stanze + +## Architettura Proposta + +### 1. Telnet Parser & Room Extractor +```python +class RoomExtractor: + """Estrae informazioni stanza dal testo MUD grezzo""" + + def parse(self, raw_output: str) -> RoomData: + # Estrae: nome stanza, descrizione, exits disponibili, + # oggetti presenti, mob/NPC visibili + pass +``` + +**Input**: Testo raw dal server MUD (con ANSI color codes) +**Output**: `RoomData` strutturato con: + +- `room_id`: hash univoco (descrizione + exits + nome) +- `name`: nome breve stanza +- `description`: descrizione lunga +- `exits`: dict {direzione: room_id_target} o None se non esplorato +- `entities`: lista mob/NPC/oggetti presenti +- `metadata`: timestamp, coordinate approssimative (se disponibili) + + +### 2. Graph Manager (NetworkX) + +```python +class MUDGraph: + """Grafo delle stanze con metodi di navigazione""" + + def add_room(self, room: RoomData) -> bool: + # Aggiunge nodo, aggiorna archi bidirezionali + pass + + def find_path(self, from_id: str, to_id: str) -> List[str]: + # A* pathfinding, ritorna lista di direzioni [n, n, e, u, ...] + pass + + def get_unexplored_exits(self, room_id: str) -> List[str]: + # Ritorna direzioni da esplorare (exit con target=None) + pass +``` + + +### 3. Storage Layer Ibrido + +#### Layer A - SQLite (Memoria Attiva) + +Per operazioni real-time durante l'esplorazione: + +```sql +-- rooms: cache attiva delle stanze esplorate +CREATE TABLE rooms ( + room_id TEXT PRIMARY KEY, + name TEXT, + description TEXT, + embedding BLOB, -- vettore per ricerca semantica + discovered_at TIMESTAMP, + visit_count INTEGER DEFAULT 1, + last_visit TIMESTAMP +); + +-- exits: grafo di navigazione +CREATE TABLE exits ( + from_room TEXT, + direction TEXT, + to_room TEXT, -- NULL se non esplorato + FOREIGN KEY (from_room) REFERENCES rooms(room_id) +); + +-- exploration_queue: stanze con exit non esplorati +CREATE TABLE exploration_queue ( + room_id TEXT, + direction TEXT, + priority INTEGER, + added_at TIMESTAMP +); +``` + + +#### Layer B - Memvid (Archivio Storico) + +Per storage compresso e ricerca semantica a lungo termine: + +```python +class MemvidArchive: + """Archivio storico in MP4 con QR code frames""" + + def snapshot(self, db_path: str) -> str: + # 1. Estrae tutte le stanze da SQLite + # 2. Chunking descrizioni + metadati + # 3. Embedding con sentence-transformers + # 4. Encoding in QR code frames + # 5. Genera MP4 H.264 + indice FAISS + pass + + def semantic_search(self, query: str, top_k: int = 5) -> List[RoomData]: + # Ricerca vettoriale sul MP4 storico + pass +``` + + +### 4. Auto-Explorer Controller + +```python +class AutoExplorer: + """Orchestra l'esplorazione automatica""" + + def decide_next_move(self, current_room: RoomData) -> str: + # Logica: + # 1. Se ci sono exit non esplorati, scegli quella più vicina + # 2. Altrimenti, BFS verso la stanza con exit non esplorati + # 3. Evita loop (no back-and-forth) + pass + + def update_graph(self, move: str, result: str) -> None: + # Aggiorna grafo dopo ogni movimento + # Se result è "You can't go that way", marca exit come blocked + pass +``` + + +### 5. Sync Manager + +```python +class SyncManager: + """Gestisce la sincronizzazione tra SQLite e Memvid""" + + def __init__(self, sync_interval: int = 100): + # Ogni N stanze nuove, triggera snapshot Memvid + pass + + def should_sync(self) -> bool: + # Controlla se ci sono abbastanza nuove stanze + pass + + def perform_sync(self): + # 1. Crea backup SQLite + # 2. Genera snapshot Memvid + # 3. Verifica integrità (sample random decode) + # 4. Opzionale: pulisce SQLite delle stanze meno recenti + pass +``` + + +## Criteri di Accettazione + +### Test 1: Parsing Base + +```python +raw_mud_output = """ +[Exits: n e w] +The Village Square +You are standing in the village square. A fountain bubbles here. +A guard is standing here. +A signpost points north to the Forest, east to the Shop. +""" +room = extractor.parse(raw_mud_output) +assert room.name == "The Village Square" +assert room.exits == {"n": None, "e": None, "w": None} +assert "guard" in room.entities +``` + + +### Test 2: Costruzione Grafo + +```python +# Simula esplorazione: Square → North → Forest +graph = MUDGraph() +# ... aggiungi stanze ... +path = graph.find_path("square_123", "forest_456") +assert path == ["n"] # diretta, una sola mossa +``` + + +### Test 3: Persistenza SQLite + +```python +# Dopo 10 stanze esplorate +assert db.count_rooms() == 10 +assert db.get_room("square_123").visit_count >= 1 +``` + + +### Test 4: Memvid Snapshot + +```python +# Dopo sync, verifica che MP4 sia leggibile +archive = MemvidArchive("world_map.mp4") +results = archive.semantic_search("forest with dark trees") +assert len(results) > 0 +assert "forest" in results.description.lower() +``` + + +### Test 5: Auto-Explorer (Integration) + +```python +# Simula 20 movimenti in un mondo di test +explorer = AutoExplorer(graph) +for _ in range(20): + move = explorer.decide_next_move(current_room) + # Simula risposta server + new_room = simulate_mud_response(move) + explorer.update_graph(move, new_room) + +# Dovrebbe aver esplorato almeno 5 stanze uniche +assert len(graph.rooms) >= 5 +``` + + +## Considerazioni Tecniche + +### Room ID Generation + +Usare hash consistente (SHA256 truncato) di: + +- Nome stanza normalizzato (lowercase, no color codes) +- Descrizione normalizzata (prime 100 char) +- Lista di exits disponibili (ordinata) + + +### Gestione Stanze Dinamiche + +Alcuni MUD hanno stanze che cambiano (portali random, maze). Implementare: + +- `is_dynamic`: flag se la stanza cambia tra visite +- `confidence`: score 0-1 sulla stabilità dell'ID + + +### Performance Targets + +- Parsing stanza: < 50ms +- Query pathfinding: < 100ms per 1000 stanze +- Sync SQLite→Memvid: background, non bloccare < 500ms UI + + +### Dipendenze + +``` +# requirements.txt +networkx +sqlite3 +numpy +sentence-transformers +pillow # QR generation +opencv-python # QR decoding +faiss-cpu # o faiss-gpu +``` + + +## Checklist Implementazione + +- [ ] Implementare `RoomExtractor` con regex MUD +- [ ] Implementare `MUDGraph` con NetworkX +- [ ] Setup schema SQLite +- [ ] Implementare `AutoExplorer` con BFS per exit non esplorati +- [ ] Implementare `MemvidArchive` encoding/decoding +- [ ] Implementare `SyncManager` con trigger automatico +- [ ] Scrivere test unitari per ogni modulo +- [ ] Scrivere test di integrazione con server MUD mock +- [ ] Documentazione API (docstring) +- [ ] Esempio di usage in `examples/basic_explorer.py` + + +## Note per l'Agente + +1. **Inizia dal SQLite**: implementa prima il grafo in-memory + SQLite, il Memvid è un "nice to have" iniziale +2. **Mock del server**: crea un `MockMUDServer` che risponde con testo predefinito per i test +3. **Logging**: usa `logging` per tracciare ogni decisione dell'explorer (utile per debug) +4. **Configurazione**: usa `pydantic.BaseSettings` per parametri (sync_interval, embedding model, ecc.) + +## Definizione di Completato + +L'automapper è considerato completato quando: + +1. ✅ Può esplorare autonomamente un mondo MUD di test con 50+ stanze +2. ✅ Costruisce un grafo navigabile con pathfinding funzionante +3. ✅ Persiste i dati in SQLite senza perdita di informazioni +4. ✅ Genera correttamente un archivio Memvid ricercabile +5. ✅ Tutti i test passano (`pytest tests/ -v`) + +--- + +**Priorità**: Alta + diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..8af54a5 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,41 @@ +[tool.poetry] +name = "mud-ai-player" +version = "0.1.0" +description = "AI player for MUD (Multi-User Dungeon) games" +authors = ["Your Name "] +readme = "README.md" +packages = [{include = "src"}] + +[tool.poetry.dependencies] +python = "^3.10" +telnetlib3 = "^2.0" +networkx = "^3.0" +sqlalchemy = "^2.0" +pydantic = "^2.0" +rich = "^13.0" +memvid = "^0.1.0" +faiss-cpu = "^1.7.4" +sentence-transformers = "^2.2.0" + +[tool.poetry.group.dev.dependencies] +pytest = "^7.0" +pytest-asyncio = "^0.21" +pytest-cov = "^4.0" +black = "^23.0" +ruff = "^0.1" + +[build-system] +requires = ["poetry-core"] +build-backend = "poetry.core.masonry.api" + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] + +[tool.black] +line-length = 100 +target-version = ['py310'] + +[tool.ruff] +line-length = 100 +target-version = "py310" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..eb5ffbf --- /dev/null +++ b/requirements.txt @@ -0,0 +1,11 @@ +telnetlib3>=2.0.4 +networkx>=3.2.1 +sqlalchemy>=2.0.23 +pydantic>=2.5.3 +rich>=13.7.0 +memvid>=0.1.0 +faiss-cpu>=1.7.4 +sentence-transformers>=2.2.0 +pytest>=7.4.3 +pytest-asyncio>=0.21.1 +pytest-cov>=4.1.0 diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..8c01541 --- /dev/null +++ b/src/__init__.py @@ -0,0 +1,4 @@ +"""MUD AI Player - Main package""" + +__version__ = "0.1.0" +__author__ = "Your Name" diff --git a/src/config/__init__.py b/src/config/__init__.py new file mode 100644 index 0000000..a8a7f12 --- /dev/null +++ b/src/config/__init__.py @@ -0,0 +1,3 @@ +"""Configuration management""" + +__all__ = ["load_config", "Config"] diff --git a/src/config/settings.py b/src/config/settings.py new file mode 100644 index 0000000..083bdac --- /dev/null +++ b/src/config/settings.py @@ -0,0 +1,95 @@ +"""Configuration loading and validation""" + +import yaml +from pydantic import BaseModel, Field +from typing import List, Optional + + +class MudConfig(BaseModel): + """MUD server configuration""" + host: str + port: int + mud_type: str = "dead_souls" + username: str = "" + password: str = "" + keep_alive_interval: int = 60 + keep_alive_command: str = "" + + +class CharacterConfig(BaseModel): + """Character preferences""" + name: str = "AIPlayer" + preferred_class: str = "warrior" + preferred_race: str = "human" + play_style: str = "balanced" + + +class AIConfig(BaseModel): + """AI engine configuration""" + engine: str = "rule_based" + llm_model: str = "llama3.2" + decision_delay_ms: List[int] = Field(default=[200, 2000]) + exploration_strategy: str = "depth_first" + + +class MapperConfig(BaseModel): + """Mapper configuration""" + class ActiveStorageConfig(BaseModel): + db_path: str = "data/maps/active/world.db" + graph_cache_size: int = 10000 + + class HistoricalStorageConfig(BaseModel): + archive_dir: str = "data/maps/archive" + + class SnapshotTriggers(BaseModel): + room_count: int = 500 + session_end: bool = True + time_interval_hours: int = 24 + + snapshot_triggers: SnapshotTriggers = Field(default_factory=SnapshotTriggers) + + active_storage: ActiveStorageConfig = Field(default_factory=ActiveStorageConfig) + historical_storage: HistoricalStorageConfig = Field(default_factory=HistoricalStorageConfig) + + +class BehaviorConfig(BaseModel): + """Behavior simulation configuration""" + reaction_time_min_ms: int = 200 + reaction_time_max_ms: int = 2000 + pause_probability: float = 0.05 + typo_probability: float = 0.0 + + +class LoggingConfig(BaseModel): + """Logging configuration""" + level: str = "INFO" + file: str = "data/logs/game.log" + console: bool = True + max_file_size_mb: int = 10 + backup_count: int = 5 + + +class Config(BaseModel): + """Main configuration""" + mud: MudConfig + character: CharacterConfig = Field(default_factory=CharacterConfig) + ai: AIConfig = Field(default_factory=AIConfig) + mapper: MapperConfig = Field(default_factory=MapperConfig) + behavior: BehaviorConfig = Field(default_factory=BehaviorConfig) + logging: LoggingConfig = Field(default_factory=LoggingConfig) + + +def load_config(config_path: str = "config.yaml") -> Config: + """ + Load configuration from YAML file + + Args: + config_path: Path to config file + + Returns: + Validated Config object + """ + with open(config_path, 'r') as f: + config_dict = yaml.safe_load(f) + + return Config(**config_dict) diff --git a/src/core/__init__.py b/src/core/__init__.py new file mode 100644 index 0000000..ee08677 --- /dev/null +++ b/src/core/__init__.py @@ -0,0 +1,15 @@ +"""Core components package""" + +from .ansi_parser import ANSIParser +from .game_state import GameState + +# TelnetClient requires telnetlib3 which may not be available in all +# environments (e.g. test runners without the full venv). Import it +# lazily so that modules which only need the parser/state can import +# from core without pulling in the telnet dependency. +try: + from .telnet_client import TelnetClient +except ImportError: + TelnetClient = None # type: ignore[assignment,misc] + +__all__ = ["TelnetClient", "ANSIParser", "GameState"] diff --git a/src/core/ansi_parser.py b/src/core/ansi_parser.py new file mode 100644 index 0000000..1fd96cd --- /dev/null +++ b/src/core/ansi_parser.py @@ -0,0 +1,40 @@ +"""ANSI color code parser""" + +import re +from typing import Tuple + + +class ANSIParser: + """Parser for ANSI escape codes and color formatting""" + + # ANSI escape sequence pattern + ANSI_PATTERN = re.compile(r'\x1b\[[0-9;]*m') + + @staticmethod + def strip_ansi(text: str) -> str: + """ + Remove all ANSI escape codes from text + + Args: + text: Text containing ANSI codes + + Returns: + Clean text without ANSI codes + """ + return ANSIParser.ANSI_PATTERN.sub('', text) + + @staticmethod + def extract_with_colors(text: str) -> list[Tuple[str, str]]: + """ + Extract text segments with their associated colors + + Args: + text: Text with ANSI codes + + Returns: + List of (text_segment, color_code) tuples + """ + # TODO: Implement color extraction for semantic parsing + # For now, just return clean text + clean = ANSIParser.strip_ansi(text) + return [(clean, 'default')] diff --git a/src/core/game_state.py b/src/core/game_state.py new file mode 100644 index 0000000..a2d4faa --- /dev/null +++ b/src/core/game_state.py @@ -0,0 +1,210 @@ +"""Game state management""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Optional, List, Dict, TYPE_CHECKING +from enum import Enum +import logging + +if TYPE_CHECKING: + from .parsers.base import ParsedPrompt, ParsedRoom, ParsedCombat + +logger = logging.getLogger(__name__) + + +class CombatState(Enum): + """Combat state enumeration""" + IDLE = "idle" + IN_COMBAT = "in_combat" + FLEEING = "fleeing" + + +@dataclass +class Stats: + """Character statistics""" + strength: int = 10 + intelligence: int = 10 + wisdom: int = 10 + dexterity: int = 10 + constitution: int = 10 + charisma: int = 10 + + +@dataclass +class Equipment: + """Character equipment""" + weapon: Optional[str] = None + armor: Optional[str] = None + helmet: Optional[str] = None + boots: Optional[str] = None + gloves: Optional[str] = None + ring: Optional[str] = None + + +@dataclass +class Character: + """Player character data""" + name: str = "" + race: str = "" + class_type: str = "" + level: int = 1 + hp: int = 100 + max_hp: int = 100 + mp: int = 50 + max_mp: int = 50 + sp: int = 0 + max_sp: int = 0 + exp: int = 0 + gold: int = 0 + stats: Stats = field(default_factory=Stats) + equipment: Equipment = field(default_factory=Equipment) + skills: Dict[str, int] = field(default_factory=dict) + + +@dataclass +class Room: + """Current room information""" + id: Optional[int] = None + fingerprint: Optional[str] = None + title: Optional[str] = None + description: str = "" + short_desc: Optional[str] = None + zone: Optional[str] = None + npcs: List[str] = field(default_factory=list) + items: List[str] = field(default_factory=list) + exits: Dict[str, int] = field(default_factory=dict) + + +@dataclass +class Quest: + """Quest information""" + id: int + name: str + description: str + objectives: List[str] = field(default_factory=list) + completed: bool = False + + +@dataclass +class GameState: + """Central game state""" + character: Character = field(default_factory=Character) + current_room: Optional[Room] = None + previous_room: Optional[Room] = None + inventory: List[str] = field(default_factory=list) + active_quests: List[Quest] = field(default_factory=list) + combat_state: CombatState = CombatState.IDLE + target: Optional[str] = None + + def is_in_combat(self) -> bool: + """Check if currently in combat""" + return self.combat_state == CombatState.IN_COMBAT + + def is_healthy(self, threshold: float = 0.7) -> bool: + """Check if character health is above threshold""" + if self.character.max_hp == 0: + return False + return (self.character.hp / self.character.max_hp) >= threshold + + # ------------------------------------------------------------------ + # Update methods — fed by the parser output + # ------------------------------------------------------------------ + + def update_from_prompt(self, prompt: "ParsedPrompt") -> None: + """Update character vitals from a parsed prompt. + + Parameters + ---------- + prompt: + A ``ParsedPrompt`` instance (from ``core.parsers.base``). + """ + if prompt.hp is not None: + self.character.hp = prompt.hp + if prompt.max_hp is not None: + self.character.max_hp = prompt.max_hp + if prompt.mp is not None: + self.character.mp = prompt.mp + if prompt.max_mp is not None: + self.character.max_mp = prompt.max_mp + if prompt.sp is not None: + self.character.sp = prompt.sp + if prompt.max_sp is not None: + self.character.max_sp = prompt.max_sp + logger.debug( + "Vitals updated: HP=%d/%d MP=%d/%d SP=%d/%d", + self.character.hp, self.character.max_hp, + self.character.mp, self.character.max_mp, + self.character.sp, self.character.max_sp, + ) + + def update_room(self, parsed_room: "ParsedRoom") -> None: + """Update current room from a parsed room description. + + Parameters + ---------- + parsed_room: + A ``ParsedRoom`` instance (from ``core.parsers.base``). + """ + # Preserve previous room for navigation tracking + if self.current_room is not None: + self.previous_room = self.current_room + + self.current_room = Room( + title=parsed_room.title, + description=parsed_room.description or "", + npcs=list(parsed_room.npcs), + items=list(parsed_room.items), + # Exits stored as direction -> room_id mapping; + # room_id is 0 (unknown) until the mapper resolves it. + exits={d: 0 for d in parsed_room.exits}, + ) + logger.debug( + "Room updated: '%s' | exits=%s | npcs=%s | items=%s", + self.current_room.title, + list(self.current_room.exits.keys()), + self.current_room.npcs, + self.current_room.items, + ) + + def update_combat(self, parsed_combat: "ParsedCombat") -> None: + """Update combat state from a parsed combat event. + + Parameters + ---------- + parsed_combat: + A ``ParsedCombat`` instance (from ``core.parsers.base``). + """ + if not parsed_combat.is_combat: + return + + action = parsed_combat.action + + if action == "flee": + self.combat_state = CombatState.FLEEING + self.target = None + logger.info("Combat: FLEEING") + elif action == "kill": + # If *we* killed the target, combat ends + if parsed_combat.attacker and "you" in parsed_combat.attacker.lower(): + self.combat_state = CombatState.IDLE + self.target = None + logger.info("Combat: target killed, returning to IDLE") + else: + # Something else died; we might still be in combat + pass + elif action in ("hit", "miss", "attack"): + self.combat_state = CombatState.IN_COMBAT + # Track the entity we are fighting + if parsed_combat.attacker and "you" in parsed_combat.attacker.lower(): + self.target = parsed_combat.target + elif parsed_combat.target and "you" in parsed_combat.target.lower(): + self.target = parsed_combat.attacker + logger.debug( + "Combat: %s -> %s (%s, dmg=%s)", + parsed_combat.attacker, + parsed_combat.target, + action, + parsed_combat.damage, + ) + diff --git a/src/core/parsers/__init__.py b/src/core/parsers/__init__.py new file mode 100644 index 0000000..acf32db --- /dev/null +++ b/src/core/parsers/__init__.py @@ -0,0 +1,23 @@ +"""Pluggable MUD parser system""" + +from .base import ( + BaseMudParser, + ParsedRoom, + ParsedPrompt, + ParsedCombat, + ParsedEvent, + EventType, +) +from .buffer import OutputBuffer +from .factory import ParserFactory + +__all__ = [ + "BaseMudParser", + "ParsedRoom", + "ParsedPrompt", + "ParsedCombat", + "ParsedEvent", + "EventType", + "OutputBuffer", + "ParserFactory", +] diff --git a/src/core/parsers/base.py b/src/core/parsers/base.py new file mode 100644 index 0000000..3980440 --- /dev/null +++ b/src/core/parsers/base.py @@ -0,0 +1,176 @@ +"""Base classes for the pluggable MUD parser system. + +This module defines the abstract interface that all MUD-specific parsers +must implement, along with the common data classes used to represent +parsed game output. +""" + +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from enum import Enum +from typing import Optional, List + + +# --------------------------------------------------------------------------- +# Data classes for parsed output +# --------------------------------------------------------------------------- + +@dataclass +class ParsedRoom: + """Structured representation of a parsed room description.""" + + title: Optional[str] = None + description: Optional[str] = None + exits: List[str] = field(default_factory=list) + npcs: List[str] = field(default_factory=list) + items: List[str] = field(default_factory=list) + raw_text: str = "" + + +@dataclass +class ParsedPrompt: + """Structured representation of the player status prompt. + + Dead Souls format example: ``hp: 350/350 mp: 440/440 sp: 431/431 >`` + """ + + hp: Optional[int] = None + max_hp: Optional[int] = None + mp: Optional[int] = None + max_mp: Optional[int] = None + sp: Optional[int] = None + max_sp: Optional[int] = None + raw_text: str = "" + + +@dataclass +class ParsedCombat: + """Structured representation of a combat event.""" + + is_combat: bool = False + attacker: Optional[str] = None + target: Optional[str] = None + damage: Optional[int] = None + action: Optional[str] = None # "hit", "miss", "kill", "flee" + raw_text: str = "" + + +class EventType(Enum): + """Types of events the parser can detect.""" + + ROOM = "room" + PROMPT = "prompt" + COMBAT = "combat" + SYSTEM = "system" # login messages, server notices + COMMUNICATION = "comm" # say, tell, shout + UNKNOWN = "unknown" + + +@dataclass +class ParsedEvent: + """Wrapper that pairs an event type with its parsed payload.""" + + event_type: EventType + room: Optional[ParsedRoom] = None + prompt: Optional[ParsedPrompt] = None + combat: Optional[ParsedCombat] = None + raw_text: str = "" + + +# --------------------------------------------------------------------------- +# Abstract base parser +# --------------------------------------------------------------------------- + +class BaseMudParser(ABC): + """Abstract base class for MUD-specific parsers. + + Every supported MUD must provide a concrete subclass that implements + the methods below. The ``ParserFactory`` uses ``mud_type`` to look up + the correct subclass at runtime. + + Typical data flow:: + + raw server bytes + -> ANSIParser.strip_ansi() + -> OutputBuffer.feed() # accumulates lines + -> OutputBuffer.flush() # returns complete block + -> BaseMudParser.parse_block() # returns list[ParsedEvent] + """ + + @abstractmethod + def parse_prompt(self, line: str) -> Optional[ParsedPrompt]: + """Try to parse a single line as a status prompt. + + Returns ``None`` if the line is not a prompt. + """ + + @abstractmethod + def parse_room(self, text_block: str) -> Optional[ParsedRoom]: + """Parse a multi-line text block as a room description. + + Returns ``None`` if the block does not contain a valid room. + """ + + @abstractmethod + def parse_combat(self, line: str) -> Optional[ParsedCombat]: + """Try to parse a single line as a combat event. + + Returns ``None`` if the line is not combat-related. + """ + + @abstractmethod + def is_prompt_line(self, line: str) -> bool: + """Return ``True`` if *line* looks like a status prompt.""" + + @abstractmethod + def is_room_description(self, text_block: str) -> bool: + """Return ``True`` if *text_block* looks like a room description.""" + + # ------------------------------------------------------------------ + # Default high-level helper (concrete subclasses may override) + # ------------------------------------------------------------------ + + def parse_block(self, text_block: str) -> List[ParsedEvent]: + """Parse a complete text block and return all detected events. + + The default implementation processes line-by-line for prompts and + combat, then attempts a room parse on the full block. Subclasses + can override for MUD-specific ordering. + """ + events: List[ParsedEvent] = [] + + for line in text_block.splitlines(): + line = line.strip() + if not line: + continue + + # Check prompt + prompt = self.parse_prompt(line) + if prompt: + events.append(ParsedEvent( + event_type=EventType.PROMPT, + prompt=prompt, + raw_text=line, + )) + continue + + # Check combat + combat = self.parse_combat(line) + if combat and combat.is_combat: + events.append(ParsedEvent( + event_type=EventType.COMBAT, + combat=combat, + raw_text=line, + )) + continue + + # Attempt room parse on the whole block + room = self.parse_room(text_block) + if room and room.title: + events.append(ParsedEvent( + event_type=EventType.ROOM, + room=room, + raw_text=text_block, + )) + + return events diff --git a/src/core/parsers/buffer.py b/src/core/parsers/buffer.py new file mode 100644 index 0000000..881ac9b --- /dev/null +++ b/src/core/parsers/buffer.py @@ -0,0 +1,103 @@ +"""Output buffer that accumulates text between prompts. + +MUD servers send output in arbitrary chunks over Telnet. A room +description may arrive as one big block or split across several small +reads. The ``OutputBuffer`` collects lines until a prompt is detected +(via the parser's ``is_prompt_line``), then flushes the accumulated +block so the parser can process it as a complete unit. +""" + +import logging +from typing import List, Optional, TYPE_CHECKING + +if TYPE_CHECKING: + from .base import BaseMudParser + +logger = logging.getLogger(__name__) + + +class OutputBuffer: + """Accumulates server output and flushes complete blocks on prompt detection. + + Usage:: + + buffer = OutputBuffer(parser) + + # Called every time data arrives from the server + blocks = buffer.feed(clean_text) + for block in blocks: + events = parser.parse_block(block) + ... + """ + + def __init__(self, parser: "BaseMudParser", max_lines: int = 200): + self._parser = parser + self._lines: List[str] = [] + self._max_lines = max_lines + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def feed(self, text: str) -> List[str]: + """Feed new text from the server and return any complete blocks. + + A *complete block* is all the text that was accumulated before a + prompt line was detected. If no prompt is found in *text*, an + empty list is returned and the text is buffered internally. + + Parameters + ---------- + text: + Clean text (ANSI already stripped) received from the server. + + Returns + ------- + list[str]: + Zero or more complete text blocks, each terminated by a prompt. + The prompt line itself is **not** included in the block — it is + returned separately so the caller can parse it as a prompt event. + """ + blocks: List[str] = [] + + for line in text.splitlines(): + if self._parser.is_prompt_line(line.strip()): + # Flush accumulated lines as a complete block + if self._lines: + block = "\n".join(self._lines) + blocks.append(block) + self._lines = [] + # The prompt line itself is a separate "block" (single line) + blocks.append(line.strip()) + else: + self._lines.append(line) + + # Safety valve: avoid unbounded memory growth + if len(self._lines) > self._max_lines: + logger.warning( + "OutputBuffer exceeded %d lines without a prompt, " + "flushing partial block", + self._max_lines, + ) + block = "\n".join(self._lines) + blocks.append(block) + self._lines = [] + + return blocks + + def flush(self) -> Optional[str]: + """Force-flush any remaining buffered text. + + Useful when the connection drops and you want to process whatever + was left in the buffer. + """ + if not self._lines: + return None + block = "\n".join(self._lines) + self._lines = [] + return block + + @property + def pending_lines(self) -> int: + """Number of lines currently buffered.""" + return len(self._lines) diff --git a/src/core/parsers/dead_souls.py b/src/core/parsers/dead_souls.py new file mode 100644 index 0000000..aa36e0a --- /dev/null +++ b/src/core/parsers/dead_souls.py @@ -0,0 +1,319 @@ +"""Dead Souls LPMUD parser. + +Implements ``BaseMudParser`` for the Dead Souls mudlib (FluffOS driver). + +Known output formats (captured from real game sessions) +------------------------------------------------------- + +**Prompt**:: + + hp: 350/350 mp: 440/440 sp: 431/431 > + +**Room description** (typical):: + + Town Square + You are standing in the town square of the village. A large fountain + dominates the center of the square. Roads lead off in several + directions. + There are three obvious exits: north, east, and west. + A town guard is standing here. + A small potion is here. + +**Exits variations**:: + + There is one obvious exit: north. + There are two obvious exits: north and south. + There are three obvious exits: north, south, and east. + Obvious exits: north, south, east + +**Combat**:: + + You attack the goblin! + You hit the goblin for 15 damage. + The goblin hits you for 8 damage. + The goblin misses you. + You killed the goblin! + You flee north! +""" + +import re +import logging +from typing import Optional, List + +from .base import BaseMudParser, ParsedRoom, ParsedPrompt, ParsedCombat + +logger = logging.getLogger(__name__) + + +class DeadSoulsParser(BaseMudParser): + """Parser tuned for the Dead Souls LPMUD (FluffOS v2.x, DS 3.8.x).""" + + # ------------------------------------------------------------------ + # Compiled regex patterns + # ------------------------------------------------------------------ + + # Prompt: "hp: 350/350 mp: 440/440 sp: 431/431 >" + _PROMPT_RE = re.compile( + r"hp:\s*(?P\d+)/(?P\d+)\s+" + r"mp:\s*(?P\d+)/(?P\d+)\s+" + r"sp:\s*(?P\d+)/(?P\d+)", + re.IGNORECASE, + ) + + # Exits — several Dead Souls variants: + # "There is one obvious exit: north." + # "There are two obvious exits: north and south." + # "There are three obvious exits: north, south, and east." + # "Obvious exits: north, south, east" + # "[ Exits: n s w ]" + _EXITS_LONG_RE = re.compile( + r"There (?:is|are) \w+ obvious exits?:\s*(.+?)\.", + re.IGNORECASE, + ) + _EXITS_SHORT_RE = re.compile( + r"Obvious exits?:\s*(.+)", + re.IGNORECASE, + ) + _EXITS_BRACKET_RE = re.compile( + r"\[\s*[Ee]xits?:\s*(.+?)\s*\]", + ) + + # Combat patterns + _COMBAT_HIT_RE = re.compile( + r"(?P.+?)\s+hits?\s+(?P.+?)\s+for\s+(?P\d+)\s+damage", + re.IGNORECASE, + ) + _COMBAT_MISS_RE = re.compile( + r"(?P.+?)\s+misses?\s+(?P.+)", + re.IGNORECASE, + ) + _COMBAT_KILL_RE = re.compile( + r"(?P.+?)\s+killed?\s+(?P.+?)!?$", + re.IGNORECASE, + ) + _COMBAT_ATTACK_RE = re.compile( + r"(?P.+?)\s+attacks?\s+(?P.+?)!?$", + re.IGNORECASE, + ) + _COMBAT_FLEE_RE = re.compile( + r"(?:You|you)\s+flee\s+(\w+)", + re.IGNORECASE, + ) + + # Lines that typically contain NPCs (Dead Souls convention) + # e.g. "A town guard is standing here." + # "Some goblins are sitting here." + # Note: plain "is here" (without standing/sitting/etc.) is treated + # as an item, not an NPC. + _NPC_RE = re.compile( + r"^(?:A|An|The|Some)\s+(.+?)\s+(?:is|are)\s+" + r"(?:standing|sitting|lying|sleeping)\s+here", + re.IGNORECASE, + ) + + # Items on the ground + # e.g. "A small potion is here." + # "A rusty sword is lying here." + _ITEM_RE = re.compile( + r"^(?:A|An|The|Some)\s+(.+?)\s+(?:is|are)\s+" + r"(?:here|lying here|on the ground)", + re.IGNORECASE, + ) + + # Known direction words + _DIRECTIONS = frozenset({ + "north", "south", "east", "west", + "northeast", "northwest", "southeast", "southwest", + "up", "down", + "n", "s", "e", "w", "ne", "nw", "se", "sw", "u", "d", + }) + + # ------------------------------------------------------------------ + # BaseMudParser implementation + # ------------------------------------------------------------------ + + def parse_prompt(self, line: str) -> Optional[ParsedPrompt]: + m = self._PROMPT_RE.search(line) + if not m: + return None + return ParsedPrompt( + hp=int(m.group("hp")), + max_hp=int(m.group("max_hp")), + mp=int(m.group("mp")), + max_mp=int(m.group("max_mp")), + sp=int(m.group("sp")), + max_sp=int(m.group("max_sp")), + raw_text=line, + ) + + def is_prompt_line(self, line: str) -> bool: + return self._PROMPT_RE.search(line) is not None + + def parse_room(self, text_block: str) -> Optional[ParsedRoom]: + lines = text_block.splitlines() + if not lines: + return None + + title: Optional[str] = None + description_lines: List[str] = [] + exits: List[str] = [] + npcs: List[str] = [] + items: List[str] = [] + + # State machine: TITLE -> DESCRIPTION -> ENTITIES/EXITS + state = "TITLE" + + for raw_line in lines: + line = raw_line.strip() + if not line: + # Blank line after title means description follows + if state == "TITLE" and title: + state = "DESCRIPTION" + continue + + # Skip prompt lines that might be embedded + if self.is_prompt_line(line): + continue + + # --- Extract exits from any line --- + exit_dirs = self._extract_exits_from_line(line) + if exit_dirs: + exits.extend(exit_dirs) + continue + + # --- Item detection (check before NPC: "is here" alone = item) --- + item_match = self._ITEM_RE.match(line) + if item_match: + items.append(item_match.group(1).strip()) + continue + + # --- NPC detection --- + npc_match = self._NPC_RE.match(line) + if npc_match: + npcs.append(npc_match.group(1).strip()) + continue + + # --- Title / Description --- + if state == "TITLE": + # In Dead Souls, the first non-empty, non-exit, non-entity + # line is the room title. Titles are typically short (< 60 + # chars) and do not end with a period. + if len(line) < 80 and not line.endswith("."): + title = line + else: + # Long first line — might be description without a + # separate title. + title = None + description_lines.append(line) + state = "DESCRIPTION" + elif state == "DESCRIPTION": + description_lines.append(line) + + if not title and not description_lines and not exits: + return None + + return ParsedRoom( + title=title, + description="\n".join(description_lines) if description_lines else None, + exits=exits, + npcs=npcs, + items=items, + raw_text=text_block, + ) + + def is_room_description(self, text_block: str) -> bool: + """Heuristic: it looks like a room if it has exits.""" + return bool( + self._EXITS_LONG_RE.search(text_block) + or self._EXITS_SHORT_RE.search(text_block) + or self._EXITS_BRACKET_RE.search(text_block) + ) + + def parse_combat(self, line: str) -> Optional[ParsedCombat]: + line = line.strip() + if not line: + return None + + # Flee + m = self._COMBAT_FLEE_RE.search(line) + if m: + return ParsedCombat( + is_combat=True, + attacker="You", + action="flee", + raw_text=line, + ) + + # Kill + m = self._COMBAT_KILL_RE.search(line) + if m: + return ParsedCombat( + is_combat=True, + attacker=m.group("attacker").strip(), + target=m.group("target").strip(), + action="kill", + raw_text=line, + ) + + # Hit with damage + m = self._COMBAT_HIT_RE.search(line) + if m: + return ParsedCombat( + is_combat=True, + attacker=m.group("attacker").strip(), + target=m.group("target").strip(), + damage=int(m.group("damage")), + action="hit", + raw_text=line, + ) + + # Miss + m = self._COMBAT_MISS_RE.search(line) + if m: + return ParsedCombat( + is_combat=True, + attacker=m.group("attacker").strip(), + target=m.group("target").strip(), + action="miss", + raw_text=line, + ) + + # Attack (initiation) + m = self._COMBAT_ATTACK_RE.search(line) + if m: + return ParsedCombat( + is_combat=True, + attacker=m.group("attacker").strip(), + target=m.group("target").strip(), + action="attack", + raw_text=line, + ) + + return None + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + def _extract_exits_from_line(self, line: str) -> List[str]: + """Try all exit patterns on a single line.""" + for pattern in (self._EXITS_LONG_RE, self._EXITS_SHORT_RE, self._EXITS_BRACKET_RE): + m = pattern.search(line) + if m: + return self._parse_exit_list(m.group(1)) + return [] + + def _parse_exit_list(self, raw: str) -> List[str]: + """Normalize an exit list string into a list of direction words. + + Handles formats like: + - ``"north, south, and east"`` + - ``"north and south"`` + - ``"n s w"`` + - ``"north, south, east"`` + """ + # Remove "and" and commas, then split on whitespace + cleaned = raw.replace(",", " ").replace(" and ", " ") + tokens = cleaned.split() + # Only keep known directions + return [t.lower() for t in tokens if t.lower() in self._DIRECTIONS] diff --git a/src/core/parsers/factory.py b/src/core/parsers/factory.py new file mode 100644 index 0000000..6b91b55 --- /dev/null +++ b/src/core/parsers/factory.py @@ -0,0 +1,55 @@ +"""Factory for creating MUD-specific parsers. + +New MUD support is added by: + +1. Creating a subclass of ``BaseMudParser`` (e.g. ``AardwolfParser``). +2. Registering it here via ``ParserFactory.register()`` or by adding it + to ``_BUILTIN_PARSERS``. +3. Setting ``mud.mud_type`` in ``config.yaml`` to the registered name. +""" + +import logging +from typing import Dict, Type + +from .base import BaseMudParser +from .dead_souls import DeadSoulsParser + +logger = logging.getLogger(__name__) + +# Built-in parsers shipped with the project +_BUILTIN_PARSERS: Dict[str, Type[BaseMudParser]] = { + "dead_souls": DeadSoulsParser, +} + + +class ParserFactory: + """Creates the correct ``BaseMudParser`` subclass for a given MUD type.""" + + _registry: Dict[str, Type[BaseMudParser]] = dict(_BUILTIN_PARSERS) + + @classmethod + def create(cls, mud_type: str) -> BaseMudParser: + """Instantiate a parser for *mud_type*. + + Raises ``ValueError`` if *mud_type* is not registered. + """ + parser_cls = cls._registry.get(mud_type) + if parser_cls is None: + available = ", ".join(sorted(cls._registry.keys())) + raise ValueError( + f"Unknown MUD type '{mud_type}'. " + f"Available parsers: {available}" + ) + logger.info("Creating parser for MUD type: %s", mud_type) + return parser_cls() + + @classmethod + def register(cls, mud_type: str, parser_cls: Type[BaseMudParser]) -> None: + """Register a custom parser at runtime.""" + cls._registry[mud_type] = parser_cls + logger.info("Registered parser '%s' -> %s", mud_type, parser_cls.__name__) + + @classmethod + def available_types(cls) -> list: + """Return the list of registered MUD type names.""" + return sorted(cls._registry.keys()) diff --git a/src/core/telnet_client.py b/src/core/telnet_client.py new file mode 100644 index 0000000..3e32b58 --- /dev/null +++ b/src/core/telnet_client.py @@ -0,0 +1,166 @@ +"""Telnet client for MUD connection""" + +import asyncio +import telnetlib3 +from typing import Optional, Callable +import logging + +logger = logging.getLogger(__name__) + + +class TelnetClient: + """Asynchronous Telnet client for MUD servers""" + + def __init__( + self, + host: str, + port: int, + max_retries: int = 3, + retry_delay: float = 5.0, + auto_reconnect: bool = True, + ): + self.host = host + self.port = port + self.max_retries = max_retries + self.retry_delay = retry_delay + self.auto_reconnect = auto_reconnect + + self.reader: Optional[telnetlib3.TelnetReader] = None + self.writer: Optional[telnetlib3.TelnetWriter] = None + self.connected = False + self.on_receive: Optional[Callable] = None + self._reconnect_count = 0 + + async def connect(self) -> bool: + """Connect to the MUD server (single attempt). + + Returns + ------- + bool + ``True`` if connection successful. + """ + try: + logger.info("Connecting to %s:%s", self.host, self.port) + self.reader, self.writer = await telnetlib3.open_connection( + self.host, + self.port, + ) + self.connected = True + self._reconnect_count = 0 + logger.info("Connected successfully") + return True + except Exception as e: + logger.error("Connection failed: %s", e) + return False + + async def connect_with_retry(self) -> bool: + """Try to connect up to ``max_retries`` times. + + Returns + ------- + bool + ``True`` if eventually connected. + """ + for attempt in range(1, self.max_retries + 1): + logger.info( + "Connection attempt %d/%d", attempt, self.max_retries + ) + if await self.connect(): + return True + if attempt < self.max_retries: + logger.info( + "Retrying in %.1f seconds...", self.retry_delay + ) + await asyncio.sleep(self.retry_delay) + + logger.error( + "Failed to connect after %d attempts", self.max_retries + ) + return False + + async def disconnect(self): + """Disconnect from the MUD server""" + self.connected = False + if self.writer: + try: + self.writer.close() + await self.writer.wait_closed() + except Exception: + pass # Best-effort close + logger.info("Disconnected") + + async def send(self, command: str): + """Send a command to the MUD server. + + Parameters + ---------- + command: + Command string to send. + """ + if not self.connected or not self.writer: + logger.warning("Not connected, cannot send command") + return + + self.writer.write(command + "\n") + await self.writer.drain() + logger.debug("Sent: %s", command) + + async def receive_loop(self): + """Main receive loop — call this in a task. + + If ``auto_reconnect`` is enabled, the loop will attempt to + re-establish the connection when it drops unexpectedly. + """ + while True: + if not self.connected or not self.reader: + break + + try: + while self.connected: + data = await self.reader.read(4096) + if not data: + logger.warning("Server closed the connection") + break + + if self.on_receive: + await self.on_receive(data) + + except asyncio.CancelledError: + raise + except Exception as e: + logger.error("Receive loop error: %s", e) + + # Connection lost + self.connected = False + + if not self.auto_reconnect: + break + + self._reconnect_count += 1 + if self._reconnect_count > self.max_retries: + logger.error( + "Max reconnect attempts (%d) reached, giving up", + self.max_retries, + ) + break + + logger.info( + "Attempting reconnect %d/%d in %.1f seconds...", + self._reconnect_count, + self.max_retries, + self.retry_delay, + ) + await asyncio.sleep(self.retry_delay) + + if await self.connect(): + logger.info("Reconnected successfully") + # Note: the caller's login logic must be re-triggered. + # This is handled by the on_receive callback seeing the + # login prompt again. + else: + logger.error("Reconnect failed") + break + + def set_receive_callback(self, callback: Callable): + """Set callback function for received data""" + self.on_receive = callback diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..a4de7cb --- /dev/null +++ b/src/main.py @@ -0,0 +1,221 @@ +"""Main application entry point""" + +import asyncio +import argparse +import sys +from pathlib import Path + +# Add src to path +sys.path.insert(0, str(Path(__file__).parent)) + +from core.telnet_client import TelnetClient +from core.ansi_parser import ANSIParser +from core.game_state import GameState +from core.parsers import ParserFactory, OutputBuffer, EventType +from config.settings import load_config +from utils.logger import setup_logger +from utils.human_behavior import HumanBehavior + +logger = None + + +async def main_loop(config): + """Main game loop""" + global logger + + # Initialize components + telnet = TelnetClient(config.mud.host, config.mud.port) + ansi_parser = ANSIParser() + game_state = GameState() + human_behavior = HumanBehavior( + config.behavior.reaction_time_min_ms, + config.behavior.reaction_time_max_ms, + config.behavior.pause_probability, + ) + + # Create MUD-specific parser via factory + mud_parser = ParserFactory.create(config.mud.mud_type) + output_buffer = OutputBuffer(mud_parser) + + logger.info( + "Parser: %s (mud_type=%s)", + type(mud_parser).__name__, + config.mud.mud_type, + ) + + # Connect to MUD + logger.info("Connecting to %s:%s", config.mud.host, config.mud.port) + if not await telnet.connect(): + logger.error("Failed to connect to MUD server") + return + + # Login state machine + login_state = "username" # username -> password -> logged_in + + # ------------------------------------------------------------------ + # Receive callback — called for every chunk from the server + # ------------------------------------------------------------------ + async def on_receive(data: str): + nonlocal login_state + + # Strip ANSI codes + clean_text = ansi_parser.strip_ansi(data) + logger.debug("Received: %s", clean_text.strip()[:200]) + + # ----- Handle login sequence ----- + if login_state == "username": + if "login" in clean_text.lower() or "name" in clean_text.lower(): + logger.info("Sending username: %s", config.mud.username) + await human_behavior.delay() + await telnet.send(config.mud.username) + login_state = "password" + return + + elif login_state == "password": + if "password" in clean_text.lower(): + logger.info("Sending password") + await human_behavior.delay() + await telnet.send(config.mud.password) + login_state = "logged_in" + return + + # Handle reconnect / take over + if "take over" in clean_text.lower() or "copy" in clean_text.lower(): + logger.info("Taking over existing session") + await human_behavior.delay() + await telnet.send("y") + login_state = "logged_in" + return + + # ----- Game logic (only after login) ----- + if login_state != "logged_in": + return + + # Feed text into the output buffer; get back complete blocks + blocks = output_buffer.feed(clean_text) + + for block in blocks: + # Parse the block into structured events + events = mud_parser.parse_block(block) + + for event in events: + if event.event_type == EventType.PROMPT and event.prompt: + game_state.update_from_prompt(event.prompt) + logger.info( + "Status: HP=%d/%d MP=%d/%d SP=%d/%d", + event.prompt.hp or 0, + event.prompt.max_hp or 0, + event.prompt.mp or 0, + event.prompt.max_mp or 0, + event.prompt.sp or 0, + event.prompt.max_sp or 0, + ) + + elif event.event_type == EventType.ROOM and event.room: + game_state.update_room(event.room) + logger.info( + "Room: '%s' | Exits: %s | NPCs: %s | Items: %s", + event.room.title, + event.room.exits, + event.room.npcs, + event.room.items, + ) + + elif event.event_type == EventType.COMBAT and event.combat: + game_state.update_combat(event.combat) + logger.info( + "Combat: %s -> %s (%s, dmg=%s)", + event.combat.attacker, + event.combat.target, + event.combat.action, + event.combat.damage, + ) + + telnet.set_receive_callback(on_receive) + + # Start receive loop + receive_task = asyncio.create_task(telnet.receive_loop()) + + try: + # Main game loop + logger.info("Starting main game loop") + + # Keep connection alive + last_keep_alive = asyncio.get_running_loop().time() + + while telnet.connected: + current_time = asyncio.get_running_loop().time() + + # Send keep-alive if logged in + if login_state == "logged_in": + if current_time - last_keep_alive >= config.mud.keep_alive_interval: + logger.debug( + "Sending keep-alive: '%s'", config.mud.keep_alive_command + ) + await telnet.send(config.mud.keep_alive_command) + last_keep_alive = current_time + + await asyncio.sleep(1.0) + + except KeyboardInterrupt: + logger.info("Shutting down...") + finally: + # Flush any remaining buffered text + remaining = output_buffer.flush() + if remaining: + events = mud_parser.parse_block(remaining) + for event in events: + if event.event_type == EventType.ROOM and event.room: + game_state.update_room(event.room) + + await telnet.disconnect() + receive_task.cancel() + + +def main(): + """Entry point""" + global logger + + # Parse arguments + parser = argparse.ArgumentParser(description="MUD AI Player") + parser.add_argument( + "--config", + default="config.yaml", + help="Path to configuration file", + ) + args = parser.parse_args() + + # Load configuration + try: + config = load_config(args.config) + except FileNotFoundError: + print(f"Error: Configuration file '{args.config}' not found") + print("Please create a config.yaml file (see examples/config.example.yaml)") + sys.exit(1) + except Exception as e: + print(f"Error loading configuration: {e}") + sys.exit(1) + + # Setup logging + logger = setup_logger( + level=config.logging.level, + log_file=config.logging.file, + console=config.logging.console, + max_file_size_mb=config.logging.max_file_size_mb, + backup_count=config.logging.backup_count, + ) + + logger.info("=" * 60) + logger.info("MUD AI Player Starting") + logger.info("=" * 60) + + # Run main loop + try: + asyncio.run(main_loop(config)) + except Exception as e: + logger.exception("Fatal error: %s", e) + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/src/modules/__init__.py b/src/modules/__init__.py new file mode 100644 index 0000000..49c597f --- /dev/null +++ b/src/modules/__init__.py @@ -0,0 +1,3 @@ +"""Specialized modules package""" + +__all__ = [] diff --git a/src/modules/character/__init__.py b/src/modules/character/__init__.py new file mode 100644 index 0000000..49230b5 --- /dev/null +++ b/src/modules/character/__init__.py @@ -0,0 +1,3 @@ +"""Character creation module""" + +__all__ = ["CharacterCreator"] diff --git a/src/modules/decision/__init__.py b/src/modules/decision/__init__.py new file mode 100644 index 0000000..0786317 --- /dev/null +++ b/src/modules/decision/__init__.py @@ -0,0 +1,3 @@ +"""Decision engine module""" + +__all__ = ["DecisionEngine", "CombatAI", "GoalManager"] diff --git a/src/modules/mapper/__init__.py b/src/modules/mapper/__init__.py new file mode 100644 index 0000000..6027938 --- /dev/null +++ b/src/modules/mapper/__init__.py @@ -0,0 +1,3 @@ +"""Auto-mapper module""" + +__all__ = ["WorldMapper", "Pathfinder", "Explorer", "ActiveStorage", "HistoricalStorage", "SemanticSearch"] diff --git a/src/modules/mapper/active_storage.py b/src/modules/mapper/active_storage.py new file mode 100644 index 0000000..3f1cdd3 --- /dev/null +++ b/src/modules/mapper/active_storage.py @@ -0,0 +1,146 @@ +"""Active storage for real-time map data""" + +import sqlite3 +import networkx as nx +from typing import Optional, Dict, List +import json +import logging + +logger = logging.getLogger(__name__) + + +class ActiveStorage: + """Layer 1: Active storage using SQLite + NetworkX for real-time exploration""" + + def __init__(self, db_path: str = "data/maps/active/world.db"): + self.db_path = db_path + self.conn: Optional[sqlite3.Connection] = None + self.graph = nx.DiGraph() + self._init_database() + self._load_graph() + + def _init_database(self): + """Initialize SQLite database with schema""" + self.conn = sqlite3.connect(self.db_path) + cursor = self.conn.cursor() + + # Create rooms table + cursor.execute(''' + CREATE TABLE IF NOT EXISTS rooms ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + fingerprint TEXT UNIQUE NOT NULL, + description TEXT, + short_desc TEXT, + zone TEXT, + coordinates TEXT, + npcs TEXT, + items TEXT, + exits TEXT, + first_visited TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + last_visited TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + visit_count INTEGER DEFAULT 1 + ) + ''') + + # Create exits table + cursor.execute(''' + CREATE TABLE IF NOT EXISTS exits ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + from_room_id INTEGER NOT NULL, + to_room_id INTEGER NOT NULL, + direction TEXT NOT NULL, + FOREIGN KEY (from_room_id) REFERENCES rooms(id), + FOREIGN KEY (to_room_id) REFERENCES rooms(id), + UNIQUE(from_room_id, direction) + ) + ''') + + # Create indexes + cursor.execute('CREATE INDEX IF NOT EXISTS idx_fingerprint ON rooms(fingerprint)') + cursor.execute('CREATE INDEX IF NOT EXISTS idx_zone ON rooms(zone)') + + self.conn.commit() + logger.info(f"Database initialized at {self.db_path}") + + def _load_graph(self): + """Load graph from database into memory""" + cursor = self.conn.cursor() + + # Load rooms as nodes + cursor.execute('SELECT id, fingerprint, description, zone FROM rooms') + for row in cursor.fetchall(): + room_id, fingerprint, description, zone = row + self.graph.add_node(room_id, fingerprint=fingerprint, description=description, zone=zone) + + # Load exits as edges + cursor.execute('SELECT from_room_id, to_room_id, direction FROM exits') + for row in cursor.fetchall(): + from_id, to_id, direction = row + self.graph.add_edge(from_id, to_id, direction=direction) + + logger.info(f"Loaded graph: {self.graph.number_of_nodes()} rooms, {self.graph.number_of_edges()} exits") + + async def add_room(self, room_data: Dict) -> int: + """ + Add new room to active storage + + Args: + room_data: Dictionary with room information + + Returns: + Room ID + """ + cursor = self.conn.cursor() + + cursor.execute(''' + INSERT OR REPLACE INTO rooms + (fingerprint, description, short_desc, zone, coordinates, npcs, items, exits) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ''', ( + room_data['fingerprint'], + room_data.get('description', ''), + room_data.get('short_desc'), + room_data.get('zone'), + json.dumps(room_data.get('coordinates', {})), + json.dumps(room_data.get('npcs', [])), + json.dumps(room_data.get('items', [])), + json.dumps(room_data.get('exits', {})) + )) + + self.conn.commit() + room_id = cursor.lastrowid + + # Update graph + self.graph.add_node(room_id, **room_data) + + # Add exits to graph + for direction, target_id in room_data.get('exits', {}).items(): + self.graph.add_edge(room_id, target_id, direction=direction) + + logger.debug(f"Added room {room_id}: {room_data.get('short_desc', 'Unknown')}") + return room_id + + def get_room(self, room_id: int) -> Optional[Dict]: + """Get room data by ID""" + cursor = self.conn.cursor() + cursor.execute('SELECT * FROM rooms WHERE id = ?', (room_id,)) + row = cursor.fetchone() + + if not row: + return None + + # Convert row to dictionary + # TODO: Implement proper row to dict conversion + return {"id": row[0], "description": row[2]} + + def get_all_rooms(self) -> List[Dict]: + """Get all rooms for snapshot""" + cursor = self.conn.cursor() + cursor.execute('SELECT * FROM rooms') + # TODO: Convert rows to dictionaries + return [] + + def close(self): + """Close database connection""" + if self.conn: + self.conn.close() diff --git a/src/modules/mapper/historical_storage.py b/src/modules/mapper/historical_storage.py new file mode 100644 index 0000000..dbb8347 --- /dev/null +++ b/src/modules/mapper/historical_storage.py @@ -0,0 +1,34 @@ +"""Historical storage using Memvid""" + +import logging +from typing import List, Dict + +logger = logging.getLogger(__name__) + + +class HistoricalStorage: + """Layer 2: Historical storage using Memvid for compressed archival""" + + def __init__(self, archive_dir: str = "data/maps/archive"): + self.archive_dir = archive_dir + # TODO: Initialize Memvid encoder/decoder + # TODO: Initialize FAISS index + logger.info(f"Historical storage initialized at {archive_dir}") + + async def create_snapshot(self, rooms: List[Dict], version: int): + """ + Create compressed MP4 snapshot of map state + + Args: + rooms: List of room dictionaries + version: Snapshot version number + """ + logger.info(f"Creating snapshot v{version} with {len(rooms)} rooms") + # TODO: Implement Memvid encoding + # TODO: Build FAISS index + pass + + def load_snapshot(self, version: int): + """Load a specific snapshot version""" + # TODO: Implement snapshot loading + pass diff --git a/src/modules/mapper/pathfinder.py b/src/modules/mapper/pathfinder.py new file mode 100644 index 0000000..fe0831a --- /dev/null +++ b/src/modules/mapper/pathfinder.py @@ -0,0 +1,55 @@ +"""Pathfinding algorithms""" + +import networkx as nx +from typing import List, Optional +import logging + +logger = logging.getLogger(__name__) + + +class Pathfinder: + """Pathfinding for navigation between rooms""" + + def __init__(self, graph: nx.DiGraph): + self.graph = graph + + def find_path(self, from_room_id: int, to_room_id: int) -> Optional[List[int]]: + """ + Find shortest path between two rooms + + Args: + from_room_id: Starting room ID + to_room_id: Destination room ID + + Returns: + List of room IDs representing the path, or None if no path exists + """ + try: + path = nx.shortest_path(self.graph, from_room_id, to_room_id) + logger.debug(f"Path found: {len(path)} rooms") + return path + except nx.NetworkXNoPath: + logger.warning(f"No path found from {from_room_id} to {to_room_id}") + return None + + def get_directions(self, path: List[int]) -> List[str]: + """ + Convert room path to direction commands + + Args: + path: List of room IDs + + Returns: + List of direction commands (n, s, e, w, etc.) + """ + directions = [] + for i in range(len(path) - 1): + from_room = path[i] + to_room = path[i + 1] + + # Get edge data + edge_data = self.graph.get_edge_data(from_room, to_room) + if edge_data and 'direction' in edge_data: + directions.append(edge_data['direction']) + + return directions diff --git a/src/modules/mapper/semantic_search.py b/src/modules/mapper/semantic_search.py new file mode 100644 index 0000000..55d2467 --- /dev/null +++ b/src/modules/mapper/semantic_search.py @@ -0,0 +1,30 @@ +"""Semantic search for map archive""" + +import logging +from typing import List, Dict + +logger = logging.getLogger(__name__) + + +class SemanticSearch: + """Semantic search on historical map archive using FAISS""" + + def __init__(self, historical_storage): + self.storage = historical_storage + # TODO: Initialize sentence transformer model + logger.info("Semantic search initialized") + + async def search(self, query: str, top_k: int = 5) -> List[Dict]: + """ + Search rooms by semantic similarity + + Args: + query: Natural language query + top_k: Number of results to return + + Returns: + List of matching rooms with similarity scores + """ + logger.info(f"Searching for: {query}") + # TODO: Implement semantic search + return [] diff --git a/src/modules/progression/__init__.py b/src/modules/progression/__init__.py new file mode 100644 index 0000000..78ea71a --- /dev/null +++ b/src/modules/progression/__init__.py @@ -0,0 +1,3 @@ +"""Progression optimizer module""" + +__all__ = ["ProgressionOptimizer"] diff --git a/src/modules/quest/__init__.py b/src/modules/quest/__init__.py new file mode 100644 index 0000000..7048325 --- /dev/null +++ b/src/modules/quest/__init__.py @@ -0,0 +1,3 @@ +"""Quest system module""" + +__all__ = ["QuestManager"] diff --git a/src/utils/__init__.py b/src/utils/__init__.py new file mode 100644 index 0000000..0eca459 --- /dev/null +++ b/src/utils/__init__.py @@ -0,0 +1,3 @@ +"""Utilities package""" + +__all__ = ["setup_logger", "HumanBehavior"] diff --git a/src/utils/human_behavior.py b/src/utils/human_behavior.py new file mode 100644 index 0000000..1702f7a --- /dev/null +++ b/src/utils/human_behavior.py @@ -0,0 +1,33 @@ +"""Human behavior simulation""" + +import random +import asyncio +import logging + +logger = logging.getLogger(__name__) + + +class HumanBehavior: + """Simulate human-like behavior to avoid bot detection""" + + def __init__( + self, + reaction_time_min_ms: int = 200, + reaction_time_max_ms: int = 2000, + pause_probability: float = 0.05 + ): + self.reaction_time_min = reaction_time_min_ms / 1000.0 + self.reaction_time_max = reaction_time_max_ms / 1000.0 + self.pause_probability = pause_probability + + async def delay(self): + """Add random human-like delay""" + delay_time = random.uniform(self.reaction_time_min, self.reaction_time_max) + await asyncio.sleep(delay_time) + + async def maybe_pause(self): + """Randomly pause to simulate human behavior""" + if random.random() < self.pause_probability: + pause_time = random.uniform(1.0, 5.0) + logger.debug(f"Taking a {pause_time:.1f}s pause") + await asyncio.sleep(pause_time) diff --git a/src/utils/logger.py b/src/utils/logger.py new file mode 100644 index 0000000..488da1c --- /dev/null +++ b/src/utils/logger.py @@ -0,0 +1,56 @@ +"""Logging configuration""" + +import logging +from logging.handlers import RotatingFileHandler +from rich.logging import RichHandler +import os + + +def setup_logger( + name: str = None, + level: str = "INFO", + log_file: str = "data/logs/game.log", + console: bool = True, + max_file_size_mb: int = 10, + backup_count: int = 5 +) -> logging.Logger: + """ + Setup logger with file and console handlers + + Args: + name: Logger name + level: Logging level (DEBUG, INFO, WARNING, ERROR) + log_file: Path to log file + console: Enable console logging + max_file_size_mb: Max log file size in MB + backup_count: Number of backup files to keep + + Returns: + Configured logger + """ + logger = logging.getLogger(name) + logger.setLevel(getattr(logging, level.upper())) + + # Remove existing handlers + logger.handlers.clear() + + # File handler with rotation + os.makedirs(os.path.dirname(log_file), exist_ok=True) + file_handler = RotatingFileHandler( + log_file, + maxBytes=max_file_size_mb * 1024 * 1024, + backupCount=backup_count + ) + file_formatter = logging.Formatter( + '%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + file_handler.setFormatter(file_formatter) + logger.addHandler(file_handler) + + # Console handler with rich formatting + if console: + console_handler = RichHandler(rich_tracebacks=True) + console_handler.setFormatter(logging.Formatter('%(message)s')) + logger.addHandler(console_handler) + + return logger diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..84386cb --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,8 @@ +"""Test configuration""" + +import pytest +import sys +from pathlib import Path + +# Add src to path for tests +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) diff --git a/tests/test_ansi_parser.py b/tests/test_ansi_parser.py new file mode 100644 index 0000000..d74bfca --- /dev/null +++ b/tests/test_ansi_parser.py @@ -0,0 +1,45 @@ +"""Tests for ANSI parser""" + +import pytest +from core.ansi_parser import ANSIParser + + +def test_strip_ansi(): + """Test ANSI code stripping""" + text_with_ansi = "\x1b[31mRed text\x1b[0m normal text" + clean_text = ANSIParser.strip_ansi(text_with_ansi) + assert clean_text == "Red text normal text" + + +def test_strip_ansi_no_codes(): + """Test text without ANSI codes""" + text = "Plain text" + clean_text = ANSIParser.strip_ansi(text) + assert clean_text == "Plain text" + + +def test_strip_ansi_multiple_codes(): + """Test multiple ANSI codes""" + text = "\x1b[1m\x1b[32mBold Green\x1b[0m and \x1b[34mBlue\x1b[0m" + clean_text = ANSIParser.strip_ansi(text) + assert clean_text == "Bold Green and Blue" + + +def test_strip_ansi_empty(): + """Test empty string""" + assert ANSIParser.strip_ansi("") == "" + + +def test_strip_ansi_prompt_with_cursor_code(): + """Test stripping cursor movement codes from Dead Souls prompt. + + Real example from game log: ``[32;1Hhp: 350/350 mp: 440/440 sp: 431/431 >`` + Note: the ``[32;1H`` is a cursor position code, not a standard color code. + The current ANSI_PATTERN only matches ``\\x1b[...m`` sequences, so this + particular code would only be stripped if it starts with \\x1b. + """ + # Standard ANSI color-wrapped prompt + text = "\x1b[32;1mhp: 350/350 mp: 440/440 sp: 431/431 >\x1b[0m" + clean = ANSIParser.strip_ansi(text) + assert "hp: 350/350" in clean + assert "sp: 431/431" in clean diff --git a/tests/test_buffer.py b/tests/test_buffer.py new file mode 100644 index 0000000..ba40de9 --- /dev/null +++ b/tests/test_buffer.py @@ -0,0 +1,111 @@ +"""Tests for the OutputBuffer.""" + +import pytest +from core.parsers.buffer import OutputBuffer +from core.parsers.dead_souls import DeadSoulsParser + + +@pytest.fixture +def parser(): + return DeadSoulsParser() + + +@pytest.fixture +def buffer(parser): + return OutputBuffer(parser) + + +class TestOutputBuffer: + + def test_no_prompt_buffers_text(self, buffer): + """Text without a prompt should be buffered, not returned.""" + blocks = buffer.feed("You are in a dark room.\nA torch flickers.") + assert blocks == [] + assert buffer.pending_lines == 2 + + def test_prompt_flushes_buffer(self, buffer): + """When a prompt arrives, the buffered text is flushed.""" + buffer.feed("Town Square\n\nYou are in the town square.\n") + blocks = buffer.feed("hp: 350/350 mp: 440/440 sp: 431/431 >") + # Should get 2 blocks: the accumulated room text + the prompt line + assert len(blocks) == 2 + assert "Town Square" in blocks[0] + assert "hp:" in blocks[1] + assert buffer.pending_lines == 0 + + def test_multiple_prompts(self, buffer): + """Two prompts in one feed should flush twice.""" + text = ( + "Room A\nhp: 100/100 mp: 50/50 sp: 75/75 >\n" + "Room B\nhp: 100/100 mp: 50/50 sp: 75/75 >\n" + ) + blocks = buffer.feed(text) + # Expect: "Room A", prompt, "Room B", prompt + assert len(blocks) == 4 + assert "Room A" in blocks[0] + assert "hp:" in blocks[1] + assert "Room B" in blocks[2] + assert "hp:" in blocks[3] + + def test_flush_remaining(self, buffer): + """flush() should return whatever is in the buffer.""" + buffer.feed("Some text without prompt\nAnother line") + remaining = buffer.flush() + assert remaining is not None + assert "Some text" in remaining + assert buffer.pending_lines == 0 + + def test_flush_empty(self, buffer): + """flush() on empty buffer returns None.""" + assert buffer.flush() is None + + def test_max_lines_safety(self, parser): + """Buffer should auto-flush when exceeding max_lines.""" + small_buffer = OutputBuffer(parser, max_lines=5) + # Feed 10 lines without a prompt + text = "\n".join(f"line {i}" for i in range(10)) + blocks = small_buffer.feed(text) + # Should have flushed at least once due to exceeding 5 lines + assert len(blocks) >= 1 + assert small_buffer.pending_lines < 5 + + def test_prompt_line_is_separate_block(self, buffer): + """The prompt itself should be its own block, not mixed with room text.""" + buffer.feed("Dark Forest\nTrees surround you.\n") + blocks = buffer.feed("hp: 200/300 mp: 100/200 sp: 150/200 >") + # blocks[0] = room text, blocks[1] = prompt + assert len(blocks) == 2 + assert "hp:" not in blocks[0] + assert "hp:" in blocks[1] + + +class TestBufferWithRealGameOutput: + """Integration-style tests using text patterns from real Dead Souls sessions.""" + + def test_login_then_room(self, buffer): + """Simulate the login sequence followed by the first room.""" + # Login messages (no prompt) + buffer.feed("Allowing login.\n") + # Then the game sends the prompt + blocks = buffer.feed("hp: 350/350 mp: 440/440 sp: 431/431 >") + assert len(blocks) == 2 + assert "Allowing login." in blocks[0] + assert "hp: 350/350" in blocks[1] + + def test_room_description_then_prompt(self, buffer): + """Simulate receiving a room description followed by prompt.""" + text = ( + "Town Square\n" + "\n" + "You are standing in the town square.\n" + "There is one obvious exit: north.\n" + "A guard is standing here.\n" + "hp: 350/350 mp: 440/440 sp: 431/431 >\n" + ) + blocks = buffer.feed(text) + # First block: room text, second: prompt + assert len(blocks) == 2 + room_block = blocks[0] + assert "Town Square" in room_block + assert "guard" in room_block + assert "hp:" in blocks[1] diff --git a/tests/test_dead_souls_parser.py b/tests/test_dead_souls_parser.py new file mode 100644 index 0000000..a455137 --- /dev/null +++ b/tests/test_dead_souls_parser.py @@ -0,0 +1,249 @@ +"""Tests for the Dead Souls LPMUD parser.""" + +import pytest +from core.parsers.dead_souls import DeadSoulsParser +from core.parsers.base import ParsedPrompt, ParsedRoom, ParsedCombat, EventType + + +@pytest.fixture +def parser(): + return DeadSoulsParser() + + +# ====================================================================== +# Prompt parsing +# ====================================================================== + +class TestPromptParsing: + + def test_standard_prompt(self, parser): + """Real prompt captured from game log.""" + line = "hp: 350/350 mp: 440/440 sp: 431/431 >" + result = parser.parse_prompt(line) + assert result is not None + assert result.hp == 350 + assert result.max_hp == 350 + assert result.mp == 440 + assert result.max_mp == 440 + assert result.sp == 431 + assert result.max_sp == 431 + + def test_prompt_with_damage(self, parser): + """Prompt after taking damage.""" + line = "hp: 250/350 mp: 400/440 sp: 431/431 >" + result = parser.parse_prompt(line) + assert result is not None + assert result.hp == 250 + assert result.max_hp == 350 + assert result.mp == 400 + + def test_prompt_case_insensitive(self, parser): + """Prompt matching should be case insensitive.""" + line = "HP: 100/200 MP: 50/100 SP: 75/150 >" + result = parser.parse_prompt(line) + assert result is not None + assert result.hp == 100 + + def test_not_a_prompt(self, parser): + """Regular text should not parse as prompt.""" + assert parser.parse_prompt("You are in a dark room.") is None + assert parser.parse_prompt("") is None + assert parser.parse_prompt("A guard says hello.") is None + + def test_is_prompt_line(self, parser): + assert parser.is_prompt_line("hp: 350/350 mp: 440/440 sp: 431/431 >") + assert not parser.is_prompt_line("You look around.") + assert not parser.is_prompt_line("") + + +# ====================================================================== +# Room parsing +# ====================================================================== + +class TestRoomParsing: + + def test_full_room_description(self, parser): + """Complete room with title, description, exits, NPC, and item.""" + text = ( + "Town Square\n" + "\n" + "You are standing in the town square of the village. A large fountain\n" + "dominates the center of the square. Roads lead off in several\n" + "directions.\n" + "There are three obvious exits: north, south, and east.\n" + "A town guard is standing here.\n" + "A small potion is here.\n" + ) + room = parser.parse_room(text) + assert room is not None + assert room.title == "Town Square" + assert "fountain" in room.description + assert set(room.exits) == {"north", "south", "east"} + assert len(room.npcs) == 1 + assert "town guard" in room.npcs[0] + assert len(room.items) == 1 + assert "small potion" in room.items[0] + + def test_single_exit(self, parser): + """Room with only one exit.""" + text = ( + "Dead End\n" + "\n" + "You have reached a dead end.\n" + "There is one obvious exit: south.\n" + ) + room = parser.parse_room(text) + assert room is not None + assert room.title == "Dead End" + assert room.exits == ["south"] + + def test_two_exits_with_and(self, parser): + """Room with two exits joined by 'and'.""" + text = ( + "Narrow Path\n" + "\n" + "A narrow path winds through the trees.\n" + "There are two obvious exits: north and south.\n" + ) + room = parser.parse_room(text) + assert room is not None + assert set(room.exits) == {"north", "south"} + + def test_bracket_exit_format(self, parser): + """Exit format: [ Exits: n s w ]""" + text = ( + "Market Street\n" + "\n" + "Stalls line both sides of the busy street.\n" + "[ Exits: n s w ]\n" + ) + room = parser.parse_room(text) + assert room is not None + assert set(room.exits) == {"n", "s", "w"} + + def test_obvious_exits_format(self, parser): + """Exit format: Obvious exits: north, south""" + text = ( + "Forest Clearing\n" + "\n" + "Sunlight filters through the canopy above.\n" + "Obvious exits: north, south\n" + ) + room = parser.parse_room(text) + assert room is not None + assert set(room.exits) == {"north", "south"} + + def test_multiple_npcs(self, parser): + """Room with multiple NPCs.""" + text = ( + "Guard Room\n" + "\n" + "A dimly lit room with weapon racks on the walls.\n" + "There are two obvious exits: north and south.\n" + "A guard captain is standing here.\n" + "A young recruit is standing here.\n" + ) + room = parser.parse_room(text) + assert room is not None + assert len(room.npcs) == 2 + + def test_empty_text_returns_none(self, parser): + """Empty input should return None.""" + assert parser.parse_room("") is None + assert parser.parse_room(" \n \n ") is None + + def test_prompt_only_not_a_room(self, parser): + """A prompt line alone is not a room.""" + room = parser.parse_room("hp: 350/350 mp: 440/440 sp: 431/431 >") + # Should return None or a room with no title + assert room is None or room.title is None + + def test_is_room_description(self, parser): + """Heuristic check for room-like text.""" + assert parser.is_room_description( + "A room.\nThere is one obvious exit: north." + ) + assert not parser.is_room_description("You pick up the sword.") + + +# ====================================================================== +# Combat parsing +# ====================================================================== + +class TestCombatParsing: + + def test_hit_with_damage(self, parser): + result = parser.parse_combat("You hit the goblin for 15 damage.") + assert result is not None + assert result.is_combat is True + assert result.action == "hit" + assert result.damage == 15 + assert "goblin" in result.target + + def test_miss(self, parser): + result = parser.parse_combat("The goblin misses you.") + assert result is not None + assert result.is_combat is True + assert result.action == "miss" + + def test_kill(self, parser): + result = parser.parse_combat("You killed the goblin!") + assert result is not None + assert result.action == "kill" + assert "goblin" in result.target + + def test_attack(self, parser): + result = parser.parse_combat("You attack the goblin!") + assert result is not None + assert result.action == "attack" + + def test_flee(self, parser): + result = parser.parse_combat("You flee north!") + assert result is not None + assert result.action == "flee" + + def test_not_combat(self, parser): + assert parser.parse_combat("You look around.") is None + assert parser.parse_combat("") is None + assert parser.parse_combat("The sun is shining.") is None + + +# ====================================================================== +# parse_block (high-level) +# ====================================================================== + +class TestParseBlock: + + def test_room_with_prompt(self, parser): + """A typical server output: room description followed by prompt.""" + text = ( + "Town Square\n" + "\n" + "You are in the town square.\n" + "There is one obvious exit: north.\n" + "A guard is standing here.\n" + "hp: 350/350 mp: 440/440 sp: 431/431 >\n" + ) + events = parser.parse_block(text) + types = [e.event_type for e in events] + assert EventType.PROMPT in types + assert EventType.ROOM in types + + def test_combat_block(self, parser): + """Combat lines should produce combat events.""" + text = ( + "You attack the goblin!\n" + "You hit the goblin for 12 damage.\n" + "The goblin hits you for 5 damage.\n" + "hp: 345/350 mp: 440/440 sp: 431/431 >\n" + ) + events = parser.parse_block(text) + combat_events = [e for e in events if e.event_type == EventType.COMBAT] + assert len(combat_events) >= 2 # at least 2 combat lines + prompt_events = [e for e in events if e.event_type == EventType.PROMPT] + assert len(prompt_events) == 1 + + def test_empty_block(self, parser): + """Empty input returns no events.""" + assert parser.parse_block("") == [] + assert parser.parse_block(" \n\n ") == [] diff --git a/tests/test_factory.py b/tests/test_factory.py new file mode 100644 index 0000000..65302a9 --- /dev/null +++ b/tests/test_factory.py @@ -0,0 +1,48 @@ +"""Tests for the ParserFactory.""" + +import pytest +from core.parsers.factory import ParserFactory +from core.parsers.dead_souls import DeadSoulsParser +from core.parsers.base import BaseMudParser + + +def test_create_dead_souls(): + parser = ParserFactory.create("dead_souls") + assert isinstance(parser, DeadSoulsParser) + + +def test_create_unknown_raises(): + with pytest.raises(ValueError, match="Unknown MUD type"): + ParserFactory.create("nonexistent_mud") + + +def test_available_types(): + types = ParserFactory.available_types() + assert "dead_souls" in types + + +def test_register_custom(): + """Register a custom parser at runtime.""" + + class DummyParser(BaseMudParser): + def parse_prompt(self, line): + return None + + def parse_room(self, text_block): + return None + + def parse_combat(self, line): + return None + + def is_prompt_line(self, line): + return False + + def is_room_description(self, text_block): + return False + + ParserFactory.register("dummy", DummyParser) + parser = ParserFactory.create("dummy") + assert isinstance(parser, DummyParser) + + # Clean up + ParserFactory._registry.pop("dummy", None) diff --git a/tests/test_game_state.py b/tests/test_game_state.py new file mode 100644 index 0000000..e745f6e --- /dev/null +++ b/tests/test_game_state.py @@ -0,0 +1,95 @@ +"""Tests for the updated GameState.""" + +import pytest +from core.game_state import GameState, CombatState +from core.parsers.base import ParsedPrompt, ParsedRoom, ParsedCombat + + +@pytest.fixture +def state(): + return GameState() + + +class TestUpdateFromPrompt: + + def test_updates_all_vitals(self, state): + prompt = ParsedPrompt(hp=300, max_hp=350, mp=400, max_mp=440, sp=431, max_sp=431) + state.update_from_prompt(prompt) + assert state.character.hp == 300 + assert state.character.max_hp == 350 + assert state.character.mp == 400 + assert state.character.max_mp == 440 + assert state.character.sp == 431 + assert state.character.max_sp == 431 + + def test_partial_update(self, state): + """Only non-None fields should update.""" + state.character.hp = 100 + state.character.max_hp = 200 + prompt = ParsedPrompt(hp=150, max_hp=200) + state.update_from_prompt(prompt) + assert state.character.hp == 150 + assert state.character.sp == 0 # untouched default + + def test_is_healthy(self, state): + prompt = ParsedPrompt(hp=280, max_hp=400, mp=100, max_mp=100, sp=100, max_sp=100) + state.update_from_prompt(prompt) + assert state.is_healthy(threshold=0.7) # 280/400 = 0.7 + assert not state.is_healthy(threshold=0.8) + + +class TestUpdateRoom: + + def test_basic_room_update(self, state): + room = ParsedRoom( + title="Town Square", + description="A busy square.", + exits=["north", "south"], + npcs=["guard"], + items=["sword"], + ) + state.update_room(room) + assert state.current_room is not None + assert state.current_room.title == "Town Square" + assert "north" in state.current_room.exits + assert state.current_room.npcs == ["guard"] + assert state.current_room.items == ["sword"] + + def test_previous_room_preserved(self, state): + room1 = ParsedRoom(title="Room A", exits=["north"]) + room2 = ParsedRoom(title="Room B", exits=["south"]) + state.update_room(room1) + state.update_room(room2) + assert state.current_room.title == "Room B" + assert state.previous_room is not None + assert state.previous_room.title == "Room A" + + +class TestUpdateCombat: + + def test_enter_combat(self, state): + combat = ParsedCombat(is_combat=True, attacker="You", target="goblin", action="attack") + state.update_combat(combat) + assert state.combat_state == CombatState.IN_COMBAT + assert state.target == "goblin" + + def test_flee(self, state): + state.combat_state = CombatState.IN_COMBAT + state.target = "goblin" + combat = ParsedCombat(is_combat=True, attacker="You", action="flee") + state.update_combat(combat) + assert state.combat_state == CombatState.FLEEING + assert state.target is None + + def test_kill_target(self, state): + state.combat_state = CombatState.IN_COMBAT + state.target = "goblin" + combat = ParsedCombat(is_combat=True, attacker="You", target="the goblin", action="kill") + state.update_combat(combat) + assert state.combat_state == CombatState.IDLE + assert state.target is None + + def test_not_combat_ignored(self, state): + combat = ParsedCombat(is_combat=False) + state.update_combat(combat) + assert state.combat_state == CombatState.IDLE