viandante/docs/modules.md

Moduli — tour ragionato

Un modulo per sezione. Le funzioni e le classi elencate sono quelle dell'API pubblica del modulo — leaky implementation details sono omesse (sono _underscore_prefixed e cambiano spesso).

---

ansi.py — stripper ANSI

Scopo. Funzione pura per togliere gli escape ANSI/VT100 prima che il testo venga dato in pasto a euristiche di parsing (stanze, login prompt, mapper UI). È una foglia del grafo: non importa nulla e non sa nulla del MUD.

API pubblica.

def strip_ansi(text: str) -> str: ...
def strip_ansi_bytes(data: bytes) -> bytes: ...
def visible_len(text: str) -> int: ...

Internals. Una regex (_ANSI_RE) compilata a import‑time che copre CSI, OSC, charset designation e DEC private mode. Sostituzione re.sub lineare. I single ESC byte senza sequenza sono lasciati intatti (non sono errori, sono escape non validi che la stampante gestirà).

---

deadsouls.py — euristiche Dead Souls / LP‑MUD

Scopo. Riconoscere stanze dal testo ricevuto dal server e canonicalizzare le direzioni delle uscite. Solo testo e solo euristica — nessuna conoscenza di IAC, GMCP, asyncio.

API pubblica.

EXIT_ALIASES: dict[str, str]
DIRECTION_OFFSETS: dict[str, tuple[int, int]]
OPPOSITE: dict[str, str]    # importato da mapper

def parse_exits(line: str) -> list[str]

@dataclass
class RoomBlock:
    name: str
    body: str
    exits: list[str]

def parse_room_block(block: str) -> RoomBlock | None

class RollingRoomParser:
    def __init__(self) -> None
    def feed_text(self, text: str) -> list[RoomBlock]
    def flush(self) -> list[RoomBlock]

Strategia di parsing blocco stanza. Basata su linee (non regex multi‑linea):

1. Accumula testo in un buffer interno.
2. Aggiungi il nuovo testo.
3. Split per \n; conserva l'ultimo pezzo parziale come buffer.
4. Per ogni riga completa: se contiene Exits: → chiudi il blocco corrente (nome = prima riga non vuota, body = resto) ed emetti un RoomBlock. Altrimenti appendi al blocco corrente.
5. Il blocco corrente rimane bufferizzato per la prossima chiamata.

Tolleranze. ANSI già strippato a monte. Linee con la parola "exits" in mezzo a dialoghi NPC sono al sicuro: parse_exits controlla il formato (bracketed/unbracketed, ":" o "-") prima di popolare la lista uscite.

---

gmcp.py — parser GMCP

Scopo. Decodificare il payload GMCP (IAC SB GMCP <module> [<json>] IAC SE) in GmcpEvent(module, data). Validare la forma del module name. Estrarre i campi tolleranti (sinonimi) di Char.Vitals / Room.Info.

API pubblica.

@dataclass(frozen=True)
class GmcpEvent:
    module: str
    data: Any

# moduli standard
MODULE_CHAR_LOGIN: str
MODULE_CHAR_LOGOUT: str
MODULE_CHAR_VITALS: str
MODULE_CHAR_STATUS: str
MODULE_CHAR_INFO: str
MODULE_ROOM_INFO: str
MODULE_ROOM_WALK_DIR: str

def parse_gmcp(data: bytes) -> GmcpEvent | None
def extract_vitals(payload: Any) -> dict[str, int]
def extract_room_info(payload: Any) -> tuple[str, str, str, str | None] | None

Sinonimi tollerati (vitale).

Chiave canonica	Sinonimi accettati (case‑sensitive)
hp / max_hp	hp, health, life
sp / max_sp	sp, mana, magic
mv / max_mv	mv, move, stam

Room.Info. Il primo argomento è l'id stabile della stanza: se num/vnum/id è presente e numerico, uso vnum{int(num)}; altrimenti name.strip().lower(). Questo è ciò che permette al mapper di ri‑identificare la stessa stanza tra sessioni.

---

mapper.py — grafo stanze + render ASCII

Scopo. Persistenza in‑memory delle stanze visitate e rendering ASCII del grafo. Le coordinate vengono assegnate solo al momento del render (lazy layout).

API pubblica.

CELL_W: int = 22

@dataclass
class Room:
    id: str
    name: str
    body: str
    exits: dict[str, str]   # direzione -> id vicino

class Mapper:
    def __init__(self) -> None
    def visit(self, name, body, exits, *, id_override=None) -> str
    def link_rooms(self, from_id, direction, to_id) -> None
    def move(self, direction: str) -> bool
    @property
    def current_room(self) -> Room | None
    @property
    def rooms(self) -> dict[str, Room]
    def render(self, *, max_cols=12, max_rows=16) -> str

Algoritmo di layout (3 fasi).

1. BFS dal current. Anchor (0, 0) ed esplora i forward edges noti. Su collisione, nudge +1 su y (fino a 16 tentativi).
2. Orphan placement. Per ogni stanza non ancora piazzata, prova un edge in qualsiasi direzione verso un placed‑room (forward o reverse). Itera finché non ci sono più avanzamenti. Questo gestisce il backtracking n → s dove la s‑room non era collegata via exits[e] ma tramite il movimento utente.
3. Stack in basso a destra. Le stanze veramente isolate finiscono qui. Se ne trovi mai una, hai probabilmente un bug di parsing: ogni stanza raggiungibile ha almeno un edge.

Snapshot in _place_adjacent. placed = list(coords.items()) viene copiato prima di mutare coords (altrimenti l'iterazione successiva vedrebbe il nuovo elemento appena inserito e potrebbe creare cicli).

Render. Coordinate y ordinate top‑down (y grande = nord). Celle di larghezza fissa CELL_W = 22. Marker » per la stanza corrente, · per le altre. I link e → w sono tratteggi ───, le verticali │.

---

logger.py — EventLogger

Scopo. Sink append‑only per eventi di sessione. Due formati paralleli: JSON Lines (machine‑readable) e Markdown (human‑readable). File aperti in write‑mode (non append): ogni sessione riparte pulito.

API pubblica.

class EventLogger:
    def __init__(
        self,
        json_path: str | None = None,
        md_path: str | None = None,
        *,
        host: str = "",
        port: int = 0,
        user: str = "",
    ) -> None

    def close(self) -> None
    def event(self, kind: str, data: dict[str, Any] | None = None) -> None

    # shortcut tipizzati
    def session_start(self, host: str, port: int, user: str) -> None
    def session_end(self, reason: str = "") -> None
    def disconnect(self, reason: str = "") -> None
    def text(self, raw_ansi: str) -> None
    def room(self, rid: str, name: str, exits: list[str]) -> None
    def gmcp(self, module: str, payload: Any) -> None
    def vitals(
        self, *,
        hp=None, sp=None, mv=None,
        max_hp=None, max_sp=None, max_mv=None,
    ) -> None
    def line_sent(self, line: str, *, redacted: bool = False) -> None
    def local_cmd(self, cmd, args=None, *, redacted: bool = False) -> None
    def login(self, phase: str, value: str = "", *, redact: bool = True) -> None

Thread safety. threading.Lock interno. Oggi è uncontended (asyncio è single‑threaded) ma difende da futuri worker thread. Tutte le scritture passano per _write_json_unlocked + _append_md_unlocked

• _flush_unlocked sotto il lock.

Sicurezza password. Tre punti di redazione:

• line_sent(line, redacted=True) → testo registrato come "***".
• local_cmd(cmd, args, redacted=True) → args list collassato a ["***"].
• login(phase, value, redact=True) (default) → value registrato come "***".

Idempotenza. close() è idempotente; __init__ solleva OSError in caso di errore di I/O e chiude gli handle aperti prima di rilanciare （nessun file descriptor leak).

---

telnet.py — trasporto telnet

Scopo. Implementare la state machine IAC (RFC 854) sopra un asyncio.StreamReader/StreamWriter e gestire la negoziazione opzioni (ECHO/NAWS/TTYPE/GMCP). Nessuna conoscenza del MUD o della UI.

API pubblica.

# costanti di protocollo
IAC, DO, DONT, WILL, WONT, SB, SE  = 255, 253, 254, 251, 252, 250, 240
EOR, GA                            = 239, 249
OPT_ECHO, OPT_TTYPE, OPT_NAWS, OPT_GMCP = 1, 24, 31, 201

# eventi
@dataclass
class TextEvent:     text: str
@dataclass
class NegotiateEvent: cmd: int; opt: int
@dataclass
class SubnegEvent:    opt: int; data: bytes
@dataclass
class PromptEvent:    ...

TelnetEvent = TextEvent | NegotiateEvent | SubnegEvent | PromptEvent

class IACParser:
    """Push parser: feed bytes, get events."""
    def feed(self, buf: bytes) -> list[TelnetEvent]

class TelnetConnection:
    def __init__(self, host, port, *,
                 naws_size=None, ttype="XTERM-256COLOR", want_gmcp=True) -> None
    async def connect(self) -> None
    async def close(self) -> None
    async def read_loop(self, handler: Callable[[TelnetEvent], Awaitable[None]]) -> None
    async def send(self, data: bytes) -> None
    async def send_line(self, line: str) -> None
    async def report_window_size(self, cols: int, rows: int) -> None

State machine (S enum).

NORMAL → byte normale               → NORMAL
NORMAL → IAC                        → IAC
IAC   → IAC                         → NORMAL (literal)
IAC   → DO|DONT|WILL|WONT           → NEGOTIATE
IAC   → SB                          → SB_OPTION
IAC   → SE                          → NORMAL (stray)
IAC   → EOR|GA                      → NORMAL (PromptEvent emesso)
IAC   → altro (NOP/DM/AO)           → NORMAL
NEGOTIATE → <opt>                    → NORMAL (NegotiateEvent emesso)
SB_OPTION → <opt>                    → SB_DATA
SB_DATA → IAC                        → SB_IAC
SB_IAC  → SE                         → NORMAL (SubnegEvent emesso)
SB_IAC  → IAC                        → SB_DATA (literal)
SB_IAC  → altro                      → NORMAL (abort)

Negoziazione opzioni.

cmd	opt	Risposta client
DO	ECHO/NAWS/TTYPE/GMCP se supportato	WILL (+ side‑effect: invio dimensione finestra per NAWS, set expects_ttype_send per TTYPE)
DO	altro	WONT
DONT	qualsiasi	WONT
WILL	qualsiasi	DONT (server offers non ci servono in v0.2)
WONT	qualsiasi	DONT

Side‑channel GMCP. Se want_gmcp=True, connect() invia proattivamente IAC WILL GMCP prima di leggere. Alcuni MUD preferiscono l'offerta client piuttosto che DO server.

---

cli.py — orchestratore

Scopo. Compostare tutti i moduli sopra in un'unica sessione interattiva. Possiede il prompt, la status bar, il dispatcher telnet, la macchina a stati di auto‑login e i comandi slash locali.

API pubblica.

@dataclass
class ClientState:
    mapper: Mapper
    parser: RollingRoomParser
    vitals: dict[str, int]
    gmcp_enabled: bool
    auto_user_sent: bool
    auto_password_prompted: bool
    auto_password_sent: bool
    logger: EventLogger | None

def main(argv: list[str] | None = None) -> int

Componenti.

• _format_prompt(state, user) → FormattedText con [room] <user> > .
• _format_status_bar(state) → FormattedText con room, GMCP on/off, hp/sp/mv.
• _handle_gmcp(data, state) → dispatch modulo → vitale/room/mapper.
• _maybe_autologin(text, conn, state, args) → matching regex su What name do you wish? / Password: e invio automatico.
• _on_telnet_event(ev, conn, state, args) → dispatcher principale.
• _input_loop(session, conn, state, user) → prompt loop async.
• _handle_local_command(line, conn, state) -> bool → comandi slash.
• _run(args) → orchestrata con asyncio.wait(FIRST_COMPLETED), conn.close() prima del drain delle pending task.

Comandi slash.

Slash	Implementazione	Note
/quit /exit /q	conn.close() + SystemExit(0)	non loggato
/help /?	_print_help()	loggato
/map	mapper.render() stampato	loggato
/where	room corrente + uscite	loggato
/look	conn.send_line("look")	loggato
/stats	stanze, GMCP, vitali	loggato
/vitals	state.vitals	loggato
/clear	prompt_toolkit.clear()	loggato
/password PWD...	redacted + conn.send_line(pwd)	loggato (redatto)
altro /...	return False → forward al MUD	loggato come line_sent da _input_loop

Auto‑login state machine.

                 text: "What name do you wish?"
                       ↓
        args.user exists & not auto_user_sent
                       ↓
   send_line(args.user); auto_user_sent = True
                       ↓
                text: "Password:"
                       ↓
    auto_password_prompted = True
    ├── args.password exists → send_line, auto_password_sent = True
    └── args.password ""     → stampa suggerimento /password PWD