"""CLI helpers for TW2002 bot."""
from __future__ import annotations
import asyncio
import logging
from pathlib import Path
import yaml
from bbsbot.games.tw2002.config import BotConfig
def setup_logging(verbose: bool = False) -> None:
level = logging.DEBUG if verbose else logging.INFO
logging.basicConfig(
level=level,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
def generate_default_config() -> str:
"""Generate example configuration with comments.
Returns:
YAML configuration string with helpful comments
"""
config = BotConfig()
data = config.model_dump(mode="json")
output = """# TW2002 Bot Configuration
# Generated by bbsbot tw2002 bot --generate-config
#
# Connection Settings:
# host: BBS server hostname or IP
# port: BBS server port
# game_password: Password for game access
#
# Character Settings:
# password: Character password for auto-login
# name_complexity: simple|medium|complex|numbered
#
# Trading Strategy:
# strategy: profitable_pairs|opportunistic|twerk_optimized|ai_strategy
#
# LLM Settings (for ai_strategy):
# llm.provider: ollama|anthropic
# llm.ollama.model: Model to use (e.g., llama3, mistral)
# llm.ollama.base_url: Ollama server URL
"""
output += yaml.dump(data, default_flow_style=False, sort_keys=False)
return output
def _apply_overrides(
config: BotConfig,
*,
host: str | None,
port: int | None,
strategy: str | None,
target_credits: int | None,
max_turns: int | None,
) -> None:
if host:
config.connection.host = host
if port:
config.connection.port = port
if strategy:
config.trading.strategy = strategy
if target_credits:
config.session.target_credits = target_credits
if max_turns:
config.session.max_turns_per_session = max_turns
def _make_watch_callback(clear: bool, show_prompt: bool = True):
def _watch(snapshot: dict) -> None:
if clear:
print("\x1b[2J\x1b[H", end="")
print(snapshot.get("screen", ""))
if show_prompt and snapshot.get("prompt_detected"):
detected = snapshot["prompt_detected"]
print("")
print(f"[prompt] {detected.get('prompt_id')} ({detected.get('input_type')})")
return _watch
async def run_bot(
config: BotConfig,
*,
watch: bool = False,
watch_interval: float = 0.0,
watch_clear: bool = True,
watch_socket: bool = False,
watch_socket_host: str = "127.0.0.1",
watch_socket_port: int = 8765,
watch_socket_protocol: str = "raw",
watch_socket_clear: bool = False,
) -> None:
"""Run the trading bot with the given configuration."""
from bbsbot.games.tw2002.bot import TradingBot
from bbsbot.games.tw2002.multi_character import MultiCharacterManager
from bbsbot.watch import WatchManager, watch_settings
print("\n" + "=" * 60)
print("TW2002 TRADING BOT")
print("=" * 60)
print(f"Strategy: {config.trading.strategy}")
print(f"Target: {config.session.target_credits:,} credits")
print(f"Max turns: {config.session.max_turns_per_session}")
print("=" * 60 + "\n")
# Set up data directory
from bbsbot.paths import default_knowledge_root
knowledge_root = default_knowledge_root()
# Include game_letter in path to separate character pools per game on same BBS
game_suffix = f"_game{config.connection.game_letter}" if config.connection.game_letter else ""
data_dir = knowledge_root / "tw2002" / f"{config.connection.host}_{config.connection.port}{game_suffix}"
data_dir.mkdir(parents=True, exist_ok=True)
multi_char = MultiCharacterManager(
config=config,
data_dir=data_dir,
sharing_mode=config.multi_character.knowledge_sharing,
)
total_characters = 0
total_profit = 0
last_bot = None
watch_manager: WatchManager | None = None
if watch_socket:
watch_settings.enabled = True
watch_settings.host = watch_socket_host
watch_settings.port = watch_socket_port
watch_settings.protocol = watch_socket_protocol
watch_settings.send_clear = watch_socket_clear
watch_manager = WatchManager()
await watch_manager.start()
if watch_socket_protocol == "json":
print(
f"[Spy] Attach: bbsbot spy --encoding utf-8 --host {watch_socket_host} --port {watch_socket_port}"
)
else:
print(f"[Spy] Attach: bbsbot spy --host {watch_socket_host} --port {watch_socket_port}")
while total_characters < config.multi_character.max_characters:
total_characters += 1
char_state = multi_char.get_or_create_next_character()
char_type = "Existing" if char_state.sessions_played > 0 else "New"
print(f"\n[Character {total_characters}] {char_state.name} ({char_type})")
bot = TradingBot(
character_name=char_state.name,
config=config,
)
last_bot = bot
if watch_manager is not None:
bot.session_manager.register_session_callback(watch_manager.attach_session)
try:
bot.set_watch_manager(watch_manager)
except Exception:
setattr(bot, "_watch_manager", watch_manager)
try:
print(f"\n[Connect] Connecting to {config.connection.host}:{config.connection.port}...")
await bot.connect(host=config.connection.host, port=config.connection.port)
print(" Connected!")
if watch and bot.session is not None:
bot.session.set_watch(_make_watch_callback(clear=watch_clear), interval_s=watch_interval)
# Use config username if provided, otherwise use generated character name
login_username = config.connection.username or char_state.name
# Use config character_password if provided, otherwise use character config password
login_password = config.connection.character_password or config.character.password
print(f"\n[Login] Logging in as {login_username}...")
await bot.login_sequence(
game_password=config.connection.game_password,
character_password=login_password,
username=login_username,
)
print(" Logged in!")
# Initialize knowledge AFTER login so we can use the detected game_letter
# This ensures different games on same BBS have separate knowledge bases
game_letter = config.connection.game_letter or bot.last_game_letter
bot.init_knowledge(config.connection.host, config.connection.port, game_letter)
bot.init_strategy()
print("\n[Orient] Getting initial state...")
state = await bot.orient(force_scan=True)
char_state.update_from_game_state(state)
print(f" Context: {state.context}")
print(f" Sector: {state.sector}")
print(f" Credits: {state.credits:,}" if state.credits else " Credits: Unknown")
await run_trading_loop(bot, config, char_state)
multi_char.save_character(char_state)
total_profit += char_state.total_profit
if char_state.credits >= config.session.target_credits:
print(f"\n{'='*60}")
print(f"TARGET REACHED: {char_state.credits:,} credits!")
print(f"{'='*60}")
break
except KeyboardInterrupt:
print("\n\nInterrupted by user")
multi_char.save_character(char_state)
break
except Exception as exc:
logging.error("Error during gameplay: %s", exc)
import traceback
traceback.print_exc()
if "destroyed" in str(exc).lower() or "died" in str(exc).lower():
print("\n*** CHARACTER DIED ***")
char_state = multi_char.handle_death(char_state)
else:
multi_char.save_character(char_state)
break
finally:
if bot.session_id:
try:
await bot.session_manager.close_session(bot.session_id)
except Exception:
pass
# Release character locks so other processes can use them
multi_char.release_all_locks()
# Display goal progression summary (AI strategy only).
try:
show_viz = (
config.trading.strategy == "ai_strategy"
and config.trading.ai_strategy.show_goal_visualization
)
except Exception:
show_viz = False
if show_viz and last_bot is not None and getattr(last_bot, "strategy", None) is not None:
strategy = last_bot.strategy
phases = getattr(strategy, "_goal_phases", None)
if phases:
from bbsbot.games.tw2002.visualization import GoalSummaryReport
print("\n" + "=" * 60)
print("GOAL PROGRESSION SUMMARY")
print("=" * 60)
report = GoalSummaryReport(
phases=phases,
max_turns=config.session.max_turns_per_session,
)
summary_text = report.render_full_summary()
print(summary_text)
print("")
emit_viz = getattr(last_bot, "emit_viz", None)
if callable(emit_viz):
current_turn = getattr(strategy, "_current_turn", None)
emit_viz("summary", summary_text, turn=current_turn)
print("\n" + "=" * 60)
print("SESSION COMPLETE")
print("=" * 60)
stats = multi_char.get_aggregate_stats()
print(f" Characters used: {stats['total_characters']}")
print(f" Deaths: {stats['total_deaths']}")
print(f" Total profit: {stats['total_profit']:,} credits")
print("=" * 60)
if watch_manager is not None:
await watch_manager.stop()
async def run_trading_loop(bot, config: BotConfig, char_state) -> None:
"""Run the main trading loop using the configured strategy."""
import random
# This function remains unchanged; existing implementation follows.
from bbsbot.games.tw2002.cli_impl import run_trading_loop as _impl
await _impl(bot, config, char_state)
def run_health_check(host: str, port: int, timeout: int) -> None:
"""Run health check on BBS server.
Args:
host: BBS server hostname or IP
port: BBS server port
timeout: Connection timeout in seconds
"""
print(f"\n{'='*60}")
print("TW2002 SERVER HEALTH CHECK")
print(f"{'='*60}")
print(f"Host: {host}")
print(f"Port: {port}")
print(f"Timeout: {timeout}s")
print(f"{'='*60}\n")
async def _check() -> None:
from bbsbot.core.session_manager import SessionManager
manager = SessionManager()
print(f"[1/3] Testing TCP connection to {host}:{port}...")
try:
session_id = await asyncio.wait_for(
manager.create_session(
host=host,
port=port,
cols=80,
rows=25,
term="ANSI",
send_newline=False,
reuse=False,
),
timeout=timeout
)
print(" ✓ Connection successful")
except asyncio.TimeoutError:
print(f" ✗ Connection timeout after {timeout}s")
print(f"\n[ERROR] Could not connect to {host}:{port}")
print("Possible causes:")
print(f" - Server is not running")
print(f" - Wrong host/port (check your configuration)")
print(f" - Firewall blocking connection")
print(f"\nTroubleshooting:")
print(f" - Verify server is running: telnet {host} {port}")
print(f" - Check firewall settings")
return
except Exception as e:
print(f" ✗ Connection failed: {e}")
print(f"\n[ERROR] Connection error: {e}")
print(f"\nIs the BBS server running at {host}:{port}?")
return
print("\n[2/3] Testing telnet negotiation...")
try:
session = await manager.get_session(session_id)
await asyncio.sleep(0.5) # Wait for negotiation
print(" ✓ Telnet negotiation complete")
except Exception as e:
print(f" ✗ Negotiation failed: {e}")
await manager.close_session(session_id)
return
print("\n[3/3] Reading initial screen...")
try:
await session.wait_for_update(timeout_ms=2000)
screen = session.snapshot().get("screen", "")
if screen and screen.strip():
print(" ✓ Server is responding")
print(f"\n[SUCCESS] Server is reachable and responding!")
print(f"\nFirst 5 lines of screen:")
lines = screen.split("\n")[:5]
for line in lines:
print(f" {line}")
else:
print(" ⚠ No screen data received")
print("\n[WARNING] Server connected but not sending data")
except Exception as e:
print(f" ✗ Read failed: {e}")
await manager.close_session(session_id)
print(f"\n{'='*60}")
asyncio.run(_check())
def run_bot_cli(
*,
config_path: str | None,
generate_config: bool,
host: str | None,
port: int | None,
verbose: bool,
strategy: str | None,
target_credits: int | None,
max_turns: int | None,
watch: bool,
watch_interval: float,
watch_clear: bool,
watch_socket: bool,
watch_socket_host: str,
watch_socket_port: int,
watch_socket_protocol: str,
watch_socket_clear: bool,
) -> None:
if generate_config:
print(generate_default_config())
return
if config_path:
try:
config = BotConfig.from_yaml(Path(config_path))
except FileNotFoundError:
print(f"\n[ERROR] Config file not found: {config_path}")
print("\nGenerate an example config with:")
print(" bbsbot tw2002 bot --generate-config > config.yaml")
print("\nOr run without config to use defaults:")
print(" bbsbot tw2002 bot --host localhost --port 2002")
return
except Exception as e:
print(f"\n[ERROR] Failed to load config: {e}")
print("\nCheck your config file syntax or generate a new one:")
print(" bbsbot tw2002 bot --generate-config > config.yaml")
return
else:
config = BotConfig()
_apply_overrides(
config,
host=host,
port=port,
strategy=strategy,
target_credits=target_credits,
max_turns=max_turns,
)
setup_logging(verbose)
try:
asyncio.run(
run_bot(
config,
watch=watch,
watch_interval=watch_interval,
watch_clear=watch_clear,
watch_socket=watch_socket,
watch_socket_host=watch_socket_host,
watch_socket_port=watch_socket_port,
watch_socket_protocol=watch_socket_protocol,
watch_socket_clear=watch_socket_clear,
)
)
except ConnectionError as e:
print(f"\n{'='*60}")
print(f"[ERROR] Connection failed: {e}")
print(f"{'='*60}")
print(f"\nIs the BBS server running at {config.connection.host}:{config.connection.port}?")
print("\nRun health check to diagnose:")
print(f" bbsbot tw2002 check --host {config.connection.host} --port {config.connection.port}")
except KeyboardInterrupt:
print("\n\nBot stopped by user")
except Exception as e:
print(f"\n{'='*60}")
print(f"[ERROR] {e}")
print(f"{'='*60}")
import traceback
traceback.print_exc()
