80 lines
2.5 KiB
Python
80 lines
2.5 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
|
|
|
|
|
|
class Agent(ABC):
|
|
"""Interface for autonomous Telnet actors."""
|
|
|
|
@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 SimpleAgent(Agent):
|
|
"""Minimal agent 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 ExploreAgent(Agent):
|
|
"""Agent 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
|
|
return self.look_command
|
|
|
|
if self.pending_targets:
|
|
target = self.pending_targets.popleft()
|
|
key = target.lower()
|
|
self.inspected_targets.add(key)
|
|
progress = f"Explored {len(self.inspected_targets)}/{len(self.seen_targets)}"
|
|
print(progress)
|
|
return f"{self.inspect_command} {target}"
|
|
|
|
return None
|