117 lines
3.8 KiB
Python
117 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from abc import ABC, abstractmethod
|
|
|
|
from collections import deque
|
|
from dataclasses import dataclass, field
|
|
from typing import Deque, Optional, Pattern, Set, Tuple
|
|
|
|
|
|
class Tool(ABC):
|
|
"""Interface for autonomous Telnet tools."""
|
|
|
|
@abstractmethod
|
|
def observe(self, output: str) -> None:
|
|
"""Receive the latest text emitted by the server."""
|
|
|
|
@abstractmethod
|
|
def decide(self) -> Optional[str]:
|
|
"""Return the next command to send, or ``None`` to stay idle."""
|
|
|
|
|
|
@dataclass
|
|
class SimpleTool(Tool):
|
|
"""Minimal tool that always returns the same command."""
|
|
|
|
default_command: str = "schau"
|
|
last_output: str = field(default="", init=False)
|
|
|
|
def observe(self, output: str) -> None:
|
|
if output:
|
|
self.last_output = output
|
|
|
|
def decide(self) -> Optional[str]:
|
|
return self.default_command
|
|
|
|
|
|
@dataclass
|
|
class ExploreTool(Tool):
|
|
"""Tool that inspects every noun it discovers in the room description."""
|
|
|
|
look_command: str = "schau"
|
|
inspect_command: str = "untersuche"
|
|
nouns_pattern: Pattern[str] = field(
|
|
default_factory=lambda: re.compile(r"\b[A-ZÄÖÜ][A-Za-zÄÖÜäöüß-]*\b")
|
|
)
|
|
last_output: str = field(default="", init=False)
|
|
look_sent: bool = field(default=False, init=False)
|
|
pending_targets: Deque[str] = field(default_factory=deque, init=False)
|
|
seen_targets: Set[str] = field(default_factory=set, init=False)
|
|
inspected_targets: Set[str] = field(default_factory=set, init=False)
|
|
|
|
def observe(self, output: str) -> None:
|
|
if not output:
|
|
return
|
|
self.last_output = output
|
|
if not self.look_sent:
|
|
return
|
|
|
|
for noun in self.nouns_pattern.findall(output):
|
|
target = noun.strip()
|
|
key = target.lower()
|
|
if not target or key in self.seen_targets:
|
|
continue
|
|
self.seen_targets.add(key)
|
|
self.pending_targets.append(target)
|
|
|
|
def decide(self) -> Optional[str]:
|
|
if not self.look_sent:
|
|
self.look_sent = True
|
|
print("[Tool] Exploring room, sending 'schau'")
|
|
return self.look_command
|
|
|
|
if self.pending_targets:
|
|
target = self.pending_targets.popleft()
|
|
key = target.lower()
|
|
self.inspected_targets.add(key)
|
|
progress = f"[Tool] Explored {len(self.inspected_targets)}/{len(self.seen_targets)} — untersuche {target}"
|
|
print(progress)
|
|
return f"{self.inspect_command} {target}"
|
|
|
|
return None
|
|
|
|
|
|
@dataclass
|
|
class CommunicationTool(Tool):
|
|
"""Tool that replies to private tells."""
|
|
|
|
reply_template: str = "teile {player} mit Hallo! Ich bin Mistle und ein Bot."
|
|
tell_pattern: Pattern[str] = field(
|
|
default_factory=lambda: re.compile(
|
|
r"^(?P<player>[^\s]+) teilt (d|D)ir mit: (?P<message>.+)$",
|
|
re.MULTILINE,
|
|
)
|
|
)
|
|
last_output: str = field(default="", init=False)
|
|
pending_replies: Deque[Tuple[str, str]] = field(default_factory=deque, init=False)
|
|
|
|
def observe(self, output: str) -> None:
|
|
if not output:
|
|
return
|
|
self.last_output = output
|
|
for match in self.tell_pattern.finditer(output):
|
|
player = match.group("player").strip()
|
|
message = match.group("message").strip()
|
|
if not player:
|
|
continue
|
|
self.pending_replies.append((player, message))
|
|
print(f"[Tool] Received message from {player}: {message}")
|
|
|
|
def decide(self) -> Optional[str]:
|
|
if not self.pending_replies:
|
|
return None
|
|
player, _ = self.pending_replies.popleft()
|
|
reply = self.reply_template.format(player=player)
|
|
print(f"[Tool] Replying to {player}")
|
|
return reply
|