Add pluggable parser system, perception loop, and project infrastructure
Phase 0 (housekeeping): - Add .gitignore to protect credentials and runtime files - Track all project source code, docs, and config Phase 1 (perception loop): - Implement pluggable parser architecture (BaseMudParser ABC + ParserFactory) - Add DeadSoulsParser with regex for HP/MP/SP prompt, room descriptions, exits (multiple formats), NPC/item detection, and combat events - Add OutputBuffer for accumulating text between prompts - Update GameState with SP support and update_from_prompt/room/combat methods - Rewire main loop to use parser + buffer for real-time game state updates - Add reconnection logic to TelnetClient (retry with backoff) - Remove old TextParser (replaced by pluggable system) - Add mud_type config field for selecting parser at runtime - Add comprehensive test suite (50 tests, all passing)
This commit is contained in:
truetype committed
1 parent
0f370ec089
commit
f85c27a742
42 files changed
+3490
-134
No files matched your search
+39
@@ -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.*
|
||||
+151
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
+160
@@ -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.
|
||||
@@ -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)
|
||||
@@ -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
|
||||
+306
@@ -0,0 +1,306 @@
|
||||
<img src="https://r2cdn.perplexity.ai/pplx-full-logo-primary-dark%402x.png" style="height:64px;margin-right:32px"/>
|
||||
|
||||
# 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
|
||||
|
||||
@@ -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 <your.email@example.com>"]
|
||||
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"
|
||||
@@ -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
|
||||
@@ -0,0 +1,4 @@
|
||||
"""MUD AI Player - Main package"""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
__author__ = "Your Name"
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Configuration management"""
|
||||
|
||||
__all__ = ["load_config", "Config"]
|
||||
@@ -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)
|
||||
@@ -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"]
|
||||
@@ -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')]
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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<hp>\d+)/(?P<max_hp>\d+)\s+"
|
||||
r"mp:\s*(?P<mp>\d+)/(?P<max_mp>\d+)\s+"
|
||||
r"sp:\s*(?P<sp>\d+)/(?P<max_sp>\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<attacker>.+?)\s+hits?\s+(?P<target>.+?)\s+for\s+(?P<damage>\d+)\s+damage",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_COMBAT_MISS_RE = re.compile(
|
||||
r"(?P<attacker>.+?)\s+misses?\s+(?P<target>.+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_COMBAT_KILL_RE = re.compile(
|
||||
r"(?P<attacker>.+?)\s+killed?\s+(?P<target>.+?)!?$",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_COMBAT_ATTACK_RE = re.compile(
|
||||
r"(?P<attacker>.+?)\s+attacks?\s+(?P<target>.+?)!?$",
|
||||
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]
|
||||
@@ -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())
|
||||
@@ -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
|
||||
+221
@@ -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()
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Specialized modules package"""
|
||||
|
||||
__all__ = []
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Character creation module"""
|
||||
|
||||
__all__ = ["CharacterCreator"]
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Decision engine module"""
|
||||
|
||||
__all__ = ["DecisionEngine", "CombatAI", "GoalManager"]
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Auto-mapper module"""
|
||||
|
||||
__all__ = ["WorldMapper", "Pathfinder", "Explorer", "ActiveStorage", "HistoricalStorage", "SemanticSearch"]
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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 []
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Progression optimizer module"""
|
||||
|
||||
__all__ = ["ProgressionOptimizer"]
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Quest system module"""
|
||||
|
||||
__all__ = ["QuestManager"]
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Utilities package"""
|
||||
|
||||
__all__ = ["setup_logger", "HumanBehavior"]
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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"))
|
||||
@@ -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
|
||||
@@ -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]
|
||||
@@ -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 ") == []
|
||||
@@ -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)
|
||||
@@ -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
|
||||
Reference in new issue
Block a user