MMORPG_AI_NPC_MCP_CLIENT_SERVER / src /ui /interface_manager.py
Chris4K's picture
Upload 202 files
1e51656 verified
"""
Interface Manager Module
This module manages the complete user interface for the MMORPG application,
connecting the HuggingFace-style UI with the game services and handling all
user interactions and event bindings.
"""
import gradio as gr
import asyncio
import time
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import asdict
from ..core.game_engine import GameEngine
from ..core.player import Player
from ..core.world import GameWorld
from ..facades.game_facade import GameFacade
from .huggingface_ui import HuggingFaceUI
class InterfaceManager:
"""Manages the complete user interface and event handling for the MMORPG."""
def __init__(self, game_facade: GameFacade, ui: HuggingFaceUI):
from ..core.game_engine import get_game_engine
self.game_engine = get_game_engine()
self.game_facade = game_facade
self.ui = ui # reference to UI for HTML generation
# Interface state
self.interface = None
# Auto-refresh timers
self.auto_refresh_interval = 2.0 # seconds
# Entity name to ID mapping for private chat
self.entity_name_to_id = {}
# Current player tracking for enhanced features
self.current_player_id = None
def create_interface(self) -> gr.Blocks:
"""Create and configure the complete interface with event handlers."""
# Create the interface with all event handlers properly wired
self._create_interface_with_events()
return self.interface
def _create_interface_with_events(self) -> None:
"""Create the interface with all event handlers properly wired."""
# Get keyboard script for head injection (critical for working keyboard controls)
keyboard_js = self.ui.get_keyboard_script()
with gr.Blocks(
title="🎮 MMORPG with MCP Integration",
theme=self.ui.theme,
css=self.ui.custom_css,
head=keyboard_js
) as interface:
# Hero section
self.ui.create_hero_section()
# Keyboard status
keyboard_status = self.ui.create_keyboard_status()
# Player state management
player_state = gr.State({})
with gr.Tabs():
# Main Game Tab
with gr.Tab("🌍 Game World"):
with gr.Row():
with gr.Column(scale=2):
# Player registration
with gr.Group():
gr.Markdown("### 🎮 Join the Adventure")
with gr.Row():
player_name = gr.Textbox(
label="Player Name",
placeholder="Enter your character name",
scale=3
)
join_btn = gr.Button("Join Game", variant="primary", scale=1)
leave_btn = gr.Button("Leave Game", variant="secondary", scale=1)
# Enhanced controls info
# gr.Markdown("""
# ### ⌨️ Enhanced Controls
# **Mouse:** Click movement buttons below
# **Keyboard:** Use **WASD** or **Arrow Keys** for movement
# **Action:** **Spacebar** for special action
# 💡 *Keyboard controls automatically activate when buttons are ready*
# """)
# Game world view with bigger map (800x600)
game_view = gr.HTML(
value=self.ui._generate_world_html(800, 600),
label="🌍 Game World"
)
# Enhanced movement controls
with gr.Group():
gr.Markdown("### 🕹️ Movement Controls")
gr.Markdown("**Keyboard:** WASD or Arrow Keys | **Mouse:** Click buttons below")
with gr.Row():
gr.HTML("")
move_up = gr.Button("↑", size="lg")
gr.HTML("")
with gr.Row():
move_left = gr.Button("←", size="lg")
action_btn = gr.Button("⚔️", size="lg", variant="secondary")
move_right = gr.Button("→", size="lg")
with gr.Row():
gr.HTML("")
move_down = gr.Button("↓", size="lg")
gr.HTML("")
with gr.Column(scale=1):
# Player stats
player_info = gr.JSON(
label="🧝‍♂️ Player Stats",
value={"status": "Not connected", "info": "Join the game to see your stats"}
)
# Online players
online_players = gr.Dataframe(
headers=["Name", "Type", "Level"],
label="👥 Online Players" )
# World events
world_events = gr.Textbox(
label="🌍 World Events & NPC Interactions",
lines=4,
interactive=False,
placeholder="World events will appear here...<br><br>💡 Tip: Walk near NPCs (📮🏪🌤️) to interact with them!<br>Then visit the 'NPC Add-ons' tab to use their features."
)
# Enhanced chat system
with gr.Row():
with gr.Column(scale=4):
chat_display = gr.Chatbot(
label="💬 Game Chat",
height=300,
type='messages',
value=[{"role": "assistant", "content": "Welcome! Join the game to start chatting!"}]
)
with gr.Row():
chat_input = gr.Textbox(
placeholder="Type your message... (use /help for commands)",
scale=4,
container=False
)
chat_send = gr.Button("Send", scale=1, variant="primary")
# Private chat system
with gr.Column(scale=2):
# Auto-refresh toggle
with gr.Row():
auto_refresh_enabled = gr.Checkbox(
label="Auto-refresh (2s)",
value=True,
info="Toggle to preserve manual selections"
)
proximity_info = gr.HTML(
value="<div style='text-align: center; color: #666;'>🔍 Move near NPCs or players to chat privately</div>",
label="📱 Nearby Entities"
)
# Private chat interface with multi-tab support
with gr.Group(visible=False) as private_chat_group:
nearby_entities = gr.Dropdown(
label="💬 Start new chat with",
choices=[],
interactive=True
)
with gr.Row():
start_chat_btn = gr.Button("Start Chat", variant="primary", scale=1)
clear_all_tabs_btn = gr.Button("Clear All", variant="secondary", scale=1)
# Chat tabs display
chat_tabs_state = gr.State({})
active_tabs_display = gr.HTML(
value="<div style='text-align: center; color: #666; padding: 10px;'>No active chats</div>",
label="Active Chats"
)
# Current chat display
current_chat_display = gr.Chatbot(
label="🔒 Private Messages",
height=200,
type='messages',
value=[],
visible=False
)
with gr.Row(visible=False) as chat_input_row:
private_message_input = gr.Textbox(
placeholder="Type private message...",
scale=4,
container=False
)
private_send_btn = gr.Button("Send", scale=1, variant="secondary")
# Documentation Tab
with gr.Tab("📚 Documentation"):
self.ui._create_documentation_content()
# NPC Add-ons Tab
with gr.Tab("🤖 NPC Add-ons"):
self.ui._create_npc_addons_content()
# Plugin System Tab
with gr.Tab("🔌 Plugin System"):
self.ui._create_plugin_system_content()
# MCP Integration Tab
with gr.Tab("🔗 MCP Integration"):
self.ui._create_mcp_integration_content()
# AI Agent Testing Tab
with gr.Tab("🧪 AI Agent Testing"):
gr.Markdown("## 🧪 Test AI Agent Integration")
gr.Markdown("*Use this to simulate AI agent connections for testing*")
with gr.Row():
ai_name = gr.Textbox(
label="AI Agent Name",
placeholder="Claude the Explorer",
scale=3
)
register_ai_btn = gr.Button("Register AI Agent", variant="primary", scale=1)
with gr.Row():
ai_action = gr.Dropdown(
choices=["move up", "move down", "move left", "move right", "chat"],
label="AI Action",
scale=2
)
ai_message = gr.Textbox(
label="AI Message (for chat)",
placeholder="Hello humans!",
scale=3
)
execute_ai_btn = gr.Button("Execute AI Action", variant="secondary")
ai_result = gr.Textbox(label="AI Action Result", interactive=False, lines=5)
# Raw Data Testing Section
gr.Markdown("### 🧪 Test Raw Data Access")
gr.Markdown("*Test accessing comprehensive world data without HTML parsing*")
with gr.Row():
test_raw_data_btn = gr.Button("Get Raw World Data", variant="secondary")
test_available_npcs_btn = gr.Button("Get Available NPCs", variant="secondary")
raw_data_result = gr.JSON(label="Raw Data Output", visible=True)
# Architecture Tab
with gr.Tab("🏗️ Architecture"):
self.ui._create_architecture_content()
# =================================================================
# EVENT HANDLERS
# =================================================================
# Join/Leave game handlers
# Join game handler with name and state inputs
join_btn.click(
self._handle_join_game,
inputs=[player_name, player_state],
outputs=[player_state, player_info, online_players, game_view]
)
# Leave game handler with state input
leave_btn.click(
self._handle_leave_game,
inputs=[player_state],
outputs=[player_state, player_info, online_players, game_view]
)
# Movement handlers
move_up.click(
self._handle_movement,
inputs=[gr.State("up"), player_state],
outputs=[player_state, game_view, world_events]
)
move_down.click(
self._handle_movement,
inputs=[gr.State("down"), player_state],
outputs=[player_state, game_view, world_events]
)
move_left.click(
self._handle_movement,
inputs=[gr.State("left"), player_state],
outputs=[player_state, game_view, world_events]
)
move_right.click(
self._handle_movement,
inputs=[gr.State("right"), player_state],
outputs=[player_state, game_view, world_events]
)
action_btn.click(
self._handle_action,
inputs=[player_state],
outputs=[world_events]
)
# Chat handlers
chat_send.click(
self._handle_chat_message,
inputs=[chat_input, player_state],
outputs=[chat_display, chat_input]
)
chat_input.submit(
self._handle_chat_message,
inputs=[chat_input, player_state],
outputs=[chat_display, chat_input]
)
# Private chat handlers
start_chat_btn.click(
self._handle_start_chat,
inputs=[nearby_entities, chat_tabs_state, player_state],
outputs=[chat_tabs_state, active_tabs_display, current_chat_display, chat_input_row]
)
private_send_btn.click(
self._handle_private_message,
inputs=[private_message_input, chat_tabs_state, player_state],
outputs=[current_chat_display, private_message_input]
)
# Clear all private chat tabs
clear_all_tabs_btn.click(
self._handle_clear_all_chats,
inputs=[chat_tabs_state],
outputs=[chat_tabs_state, active_tabs_display, current_chat_display, chat_input_row]
)
# AI Agent Testing handlers
register_ai_btn.click(
self._handle_register_ai_agent,
inputs=[ai_name],
outputs=[ai_result]
)
execute_ai_btn.click(
self._handle_execute_ai_action,
inputs=[ai_action, ai_message],
outputs=[ai_result]
)
# Raw Data Testing handlers
test_raw_data_btn.click(
self._handle_test_raw_world_data,
inputs=[],
outputs=[raw_data_result]
)
test_available_npcs_btn.click(
self._handle_test_available_npcs,
inputs=[],
outputs=[raw_data_result]
)
# Auto-refresh handlers via Timer (Gradio v3+)
refresh_timer = gr.Timer(value=self.auto_refresh_interval)
refresh_timer.tick(
self._auto_refresh_game_state,
inputs=[player_state, auto_refresh_enabled],
outputs=[player_info, online_players, game_view, proximity_info, nearby_entities, private_chat_group, chat_display, world_events]
)
self.interface = interface
# =================================================================
# EVENT HANDLER METHODS
# =================================================================
def _handle_join_game(self, name: str, current_state: Dict) -> Tuple[Dict, Dict, List[List[str]], str]:
"""Handle player joining the game."""
# Validate name
if not name or not name.strip():
return current_state, {"error": "Please enter a valid name"}, [], self.ui._generate_world_html(800, 600)
# Attempt to join via facade
player_id = self.game_facade.join_game(name.strip())
if not player_id:
return current_state, {"error": "Failed to join game"}, [], self.ui._generate_world_html(800, 600)
# Update state
current_state.update({"player_id": player_id, "joined": True, "name": name.strip()})
# Set current player ID for enhanced features (glow effect, etc.)
self.current_player_id = player_id
# Build UI outputs
status = {"player_id": player_id, "status": f"Joined as {name.strip()}"}
rows = self._get_online_players_data()
world_html = self._generate_world_html_with_players()
return current_state, status, rows, world_html
def _handle_leave_game(self, current_state: Dict) -> Tuple[Dict, Dict, List[List[str]], str]:
"""Handle player leaving the game: clear state and return updated UI components."""
# Get player_id from current state
player_id = current_state.get("player_id") # Attempt to leave via facade
success = self.game_facade.leave_game(player_id)
# Reset state and clear current player tracking
new_state: Dict = {}
self.current_player_id = None # Build UI outputs
status = {"status": "Left game" if success else "Leave failed"}
rows = self._get_online_players_data()
world_html = self._generate_world_html_with_players()
return new_state, status, rows, world_html
def _handle_movement(self, direction: str, current_state: Dict) -> Tuple[Dict, str, str]:
"""Handle player movement with enhanced collision detection."""
if not current_state.get("joined"):
return current_state, self._generate_world_html_with_players(), "Join the game to move!"
try:
player_id = current_state.get("player_id")
success, new_position, events = self.game_facade.move_player(player_id, direction)
if success:
current_state.update(new_position)
world_html = self._generate_world_html_with_players()
event_text = "<br>".join(events) if events else f"Moved {direction}"
return current_state, world_html, event_text
else:
return current_state, self._generate_world_html_with_players(), f"Cannot move {direction} - blocked by obstacle!"
except Exception as e:
return current_state, self._generate_world_html_with_players(), f"Movement error: {str(e)}"
def _handle_action(self, current_state: Dict) -> str:
"""Handle special action button."""
if not current_state.get("joined"):
return "Join the game to use actions!"
try:
player_id = current_state.get("player_id")
result = self.game_facade.handle_action(player_id)
return result
except Exception as e:
return f"Action error: {str(e)}"
def _handle_chat_message(self, message: str, current_state: Dict) -> Tuple[List[Dict], str]:
"""Handle public chat messages with state context."""
# Validate message and join status
if not message or not message.strip() or not current_state.get("joined"):
return [], ""
try:
player_id = current_state.get("player_id")
# Send message via facade
self.game_facade.send_chat_message(player_id, message.strip())
# Get recent history
history = self.game_facade.get_chat_history(20)
formatted = self._format_chat_messages(history)
# Clear input after sending
return formatted, ""
except Exception:
return [{"role": "assistant", "content": "Chat error"}], ""
def _handle_start_chat(self, entity_selection: str, chat_tabs_state: Dict, current_state: Dict) -> Tuple[Dict, str, gr.Chatbot, gr.Group]:
"""Handle starting a private chat."""
if not entity_selection or not current_state.get("joined"):
return chat_tabs_state, gr.update(value=self._generate_chat_tabs_html(chat_tabs_state)), gr.update(visible=False), gr.update(visible=False)
try:
# Convert display name back to entity ID using the stored mapping
entity_id = self.entity_name_to_id.get(entity_selection, entity_selection)
entity_name = entity_selection # We already have the display name
if entity_id not in chat_tabs_state:
chat_tabs_state[entity_id] = {
"name": entity_name,
"active": True,
"pinned": False,
"unread": 0 }
# Set as active
for tab_id in chat_tabs_state:
chat_tabs_state[tab_id]["active"] = (tab_id == entity_id)
# Get chat messages for this entity
player_id = current_state.get("player_id")
messages = self.game_facade.get_private_messages(player_id, entity_id)
tabs_html = self._generate_chat_tabs_html(chat_tabs_state)
return chat_tabs_state, gr.update(value=tabs_html), gr.update(value=messages, visible=True), gr.update(visible=True)
except Exception as e:
return chat_tabs_state, gr.update(value=f"Error: {str(e)}"), gr.update(visible=False), gr.update(visible=False)
def _handle_private_message(self, message: str, chat_tabs_state: Dict, current_state: Dict) -> Tuple[List[Dict], str]:
"""Handle private chat messages."""
# Validate input and state
if not message or not message.strip() or not current_state.get("joined"):
return [], ""
# Determine active entity for private chat
entity_id = next((eid for eid, info in chat_tabs_state.items() if info.get("active")), None)
if not entity_id:
return [], ""
print(f"[InterfaceManager] Sending private message from {current_state.get('player_id')} to {entity_id}: '{message.strip()}'")
# Send private message via game facade
player_id = current_state.get("player_id")
player_name = current_state.get("name", "Unknown") # Get player name from state
success = self.game_facade.send_private_message(player_id, player_name, entity_id, message.strip())
if success:
# Fetch updated messages
messages = self.game_facade.get_private_messages(player_id, entity_id)
# Format messages for Gradio Chatbot (using backup version formatting)
formatted = []
for msg in messages[-20:]: # last 20 messages
# Determine role based on sender_id vs current player
role = "user" if msg.get("sender_id") == player_id else "assistant"
# Convert newlines to HTML breaks for proper display
message_content = msg.get('message', '').replace('\n', '<br>')
content = f"[{msg.get('timestamp', '')}] {msg.get('sender', 'Unknown')}: {message_content}"
formatted.append({"role": role, "content": content})
print(f"[InterfaceManager] Formatted {len(formatted)} private messages for display")
return formatted, ""
else:
print(f"[InterfaceManager] Failed to send private message")
return [], ""
def _handle_clear_all_chats(self, chat_tabs_state: Dict) -> Tuple[Dict, str, gr.Chatbot, gr.Group]:
"""Handle clearing all private chat tabs."""
chat_tabs_state.clear()
tabs_html = self._generate_chat_tabs_html(chat_tabs_state)
return chat_tabs_state, tabs_html, gr.update(visible=False), gr.update(visible=False)
def _handle_register_ai_agent(self, ai_name: str) -> str:
"""Handle AI agent registration."""
if not ai_name or not ai_name.strip():
return "❌ Please enter a valid AI agent name"
try:
# Generate a unique agent ID
import uuid
agent_id = f"test_agent_{uuid.uuid4().hex[:8]}"
# Register AI agent via game facade
result = self.game_facade.register_ai_agent(agent_id, ai_name.strip())
if result:
# Store the agent ID for later use
if not hasattr(self, 'registered_agents'):
self.registered_agents = {}
self.registered_agents[agent_id] = ai_name.strip()
return f"✅ AI agent '{ai_name.strip()}' registered successfully!\nAgent ID: {agent_id}"
else:
return f"❌ Failed to register AI agent '{ai_name.strip()}'"
except Exception as e:
return f"❌ Error registering AI agent: {str(e)}"
def _handle_execute_ai_action(self, action: str, message: str) -> str:
"""Handle AI agent action execution."""
if not hasattr(self, 'registered_agents') or not self.registered_agents:
return "❌ No AI agents registered. Please register an AI agent first!"
if not action:
return "❌ Please select an action to execute"
try:
# Use the first registered agent for demo purposes
agent_id = list(self.registered_agents.keys())[0]
agent_name = self.registered_agents[agent_id]
if action.startswith("move"):
# Extract direction from action (e.g., "move up" -> "up")
direction = action.split()[1] if len(action.split()) > 1 else action
result = self.game_facade.move_ai_agent(agent_id, direction)
if result.get("success"):
position = result.get("position", {})
return f"🤖 {agent_name} moved {direction} successfully!\nNew position: ({position.get('x', '?')}, {position.get('y', '?')})"
else:
error_msg = result.get("error", "Movement failed")
return f"❌ {agent_name} movement failed: {error_msg}"
elif action == "chat":
if not message or not message.strip():
return "❌ Please enter a chat message"
result = self.game_facade.ai_agent_chat(agent_id, message.strip())
if result.get("success"):
return f"💬 {agent_name} sent chat message: '{message.strip()}'"
else:
error_msg = result.get("error", "Chat failed")
return f"❌ {agent_name} chat failed: {error_msg}"
else:
return f"❓ Unknown action: {action}"
except Exception as e:
return f"❌ Error executing AI action: {str(e)}"
def _handle_test_raw_world_data(self) -> Dict:
"""Handle testing raw world data access."""
try:
raw_data = self.game_facade.get_raw_world_data()
return raw_data
except Exception as e:
return {"error": f"Failed to get raw world data: {str(e)}"}
def _handle_test_available_npcs(self) -> List[Dict]:
"""Handle testing available NPCs data access."""
try:
npcs_data = self.game_facade.get_available_npcs()
return npcs_data
except Exception as e:
return [{"error": f"Failed to get NPCs data: {str(e)}"}]
def _auto_refresh_game_state(self, current_state: Dict, auto_refresh: bool):
"""Auto-refresh game state and UI elements with update objects.""" # If auto-refresh disabled or not joined, clear updates
if not auto_refresh or not current_state.get("joined"):
return (
gr.update(), # player_info
gr.update(), # online_players
gr.update(), # game_view
gr.update(), # proximity_info
gr.update(), # nearby_entities
gr.update(visible=False), # private_chat_group
gr.update(), # chat_display
gr.update(), # world_events
)
try:
# Get updated data
player_id = current_state.get("player_id")
# Get updated player stats via facade
player_info_data = self.game_facade.get_player_stats(player_id)
online_players_data = self._get_online_players_data()
world_html = self._generate_world_html_with_players()
proximity_html, nearby_choices, show_private_chat = self._get_proximity_info(player_id)
# Get updated chat messages
chat_history = self.game_facade.get_chat_history(20)
formatted_chat = self._format_chat_messages(chat_history)
# Get formatted world events for display
world_events_data = self._get_formatted_world_events()
# Return update objects for each component
return (
gr.update(value=player_info_data), # player_info
gr.update(value=online_players_data), # online_players
gr.update(value=world_html), # game_view
gr.update(value=proximity_html), # proximity_info
gr.update(choices=nearby_choices), # nearby_entities
gr.update(visible=show_private_chat), # private_chat_group
gr.update(value=formatted_chat), # chat_display
gr.update(value=world_events_data), # world_events
)
except Exception as e:
print(f"Auto-refresh error: {e}")
return (
gr.update(), # player_info
gr.update(), # online_players
gr.update(), # game_view
gr.update(), # proximity_info
gr.update(), # nearby_entities
gr.update(visible=False), # private_chat_group
gr.update(), # chat_display
gr.update(), # world_events
)
# =================================================================
# HELPER METHODS
# =================================================================
def _get_online_players_data(self) -> List[List[str]]:
"""Get formatted online players data."""
try:
players_data = [] # Use facade to get current players
for player in self.game_facade.get_all_players().values():
players_data.append([
player.name,
"🤖 AI" if player.type == "ai_agent" else "👤 Human",
str(player.level)
])
return players_data
except Exception:
return []
def _generate_world_html_with_players(self) -> str:
"""Generate world HTML with current player positions, glow effects, and player names."""
try:
# Get base world HTML
base_html = self.ui._generate_world_html(800, 600)
# Get current player ID for glow effect
current_player_id = getattr(self, 'current_player_id', None)
# Add players to the world with enhanced features
players_html = ""
all_players = self.game_facade.get_all_players()
for player in all_players.values():
player_icon = "🤖" if player.type == "ai_agent" else "🧑‍🦰"
is_current = player.id == current_player_id
# Player glow effect implementation
border_style = "border: 2px solid yellow; border-radius: 50%;" if is_current else ""
# Player sprite with glow effect
players_html += f'''
<div style="position: absolute; left: {player.x}px; top: {player.y}px;
font-size: 20px; z-index: 10; {border_style}
text-shadow: 2px 2px 4px rgba(0,0,0,0.5);"
title="{player.name} (Level {player.level})">
{player_icon}
</div>
'''
# Player name display implementation
name_background = "rgba(255,215,0,0.9)" if is_current else "rgba(255,215,0,0.6)"
players_html += f'''
<div style="position: absolute; left: {player.x - 15}px; top: {player.y - 15}px;
background: {name_background}; color: black; padding: 1px 4px;
border-radius: 3px; font-size: 8px; font-weight: bold; z-index: 11;">
{player.name} (Lv.{player.level})
</div>
'''
# Status line implementation
import time
player_count = len(all_players)
current_time = time.strftime('%H:%M:%S')
# Get width and height from the world HTML or set defaults
width, height = 800, 600
status_html = f'''
<!--<div style="position: absolute; top: 10px; left: 50%; transform: translateX(-50%);
text-align: center; background: rgba(0,0,0,0.1); padding: 10px;
border-radius: 5px; font-family: Arial, sans-serif; z-index: 12;">
-->
<div style="position: absolute; top: 10px; left: 10px; background: rgba(255,255,255,0.9);
padding: 10px; border-radius: 8px; font-size: 12px; box-shadow: 0 2px 5px rgba(0,0,0,0.2);">
🌍 <strong>Fantasy Realm</strong> |
Players: {player_count}/20 |
Last Update: {current_time}
<br>
<!-- 🎮 <strong>Use WASD or Arrow Keys to move, Space for Action</strong>
--> <br>
🧑‍🦰 Players | 📮🏪🚶🧙🌤️ NPCs | 🌳 Trees
</div>
'''
# Insert enhanced content into the world HTML
enhanced_content = players_html + status_html
if enhanced_content:
base_html = base_html.replace(
'<div id="players-container"></div>',
f'<div id="players-container">{enhanced_content}</div>'
)
return base_html
except Exception as e:
print(f"Error generating world HTML: {e}")
return self.ui._generate_world_html(800, 600)
def _format_chat_messages(self, messages: List[Dict]) -> List[Dict]:
"""Format chat messages for Gradio chatbot."""
formatted = []
for msg in messages[-20:]: # Show last 20 messages
role = "assistant" if msg.get("sender") == "System" else "user"
# Convert newlines to HTML breaks for proper display
message_content = msg.get('message', '').replace('\n', '<br>')
content = f"[{msg.get('timestamp', '')}] {msg.get('sender', 'Unknown')}: {message_content}"
formatted.append({"role": role, "content": content})
return formatted
def _generate_chat_tabs_html(self, chat_tabs_state: Dict) -> str:
"""Generate HTML for chat tabs display."""
if not chat_tabs_state:
return "<div style='text-align: center; color: #666; padding: 10px;'>No active chats</div>"
tabs_html = "<div style='display: flex; flex-wrap: wrap; gap: 5px; padding: 5px;'>"
for entity_id, tab_info in chat_tabs_state.items():
active_class = "active" if tab_info.get('active') else ""
pin_icon = "📌" if tab_info.get('pinned') else ""
unread_badge = f" ({tab_info.get('unread', 0)})" if tab_info.get('unread', 0) > 0 else ""
tabs_html += f'''
<div class="chat-tab {active_class}"
style="display: flex; align-items: center; gap: 5px;"
title="Entity ID: {entity_id}">
<span>{tab_info['name']}{unread_badge}</span>
{pin_icon}
<span style="color: #f44336; font-weight: bold; cursor: pointer;" title="Close tab">×</span>
</div>
'''
tabs_html += "</div>"
return tabs_html
def _get_entity_name(self, entity_id: str) -> str:
"""Get display name for an entity."""
try:
world = self.game_engine.get_world()
# Check NPCs
npc = world.npcs.get(entity_id)
if npc:
return npc.get("name", entity_id)
# Check players
player = world.players.get(entity_id)
if player:
return player.name
return entity_id
except Exception:
return entity_id
def _get_proximity_info(self, player_id: str) -> Tuple[str, List[str], bool]:
"""Get proximity information and nearby entities."""
try:
proximity_data = self.game_facade.get_proximity_info(player_id)
nearby_entities = proximity_data.get("nearby_entities", [])
if nearby_entities:
proximity_html = f"<div style='color: #4caf50;'>📡 Found {len(nearby_entities)} nearby entities</div>" # Create choices with human-readable names and maintain ID mapping
choices = []
self.entity_name_to_id = {} # Reset and store mapping for later use
for entity in nearby_entities:
entity_id = entity['id']
entity_name = self._get_entity_name(entity_id)
# Store the mapping for reverse lookup
self.entity_name_to_id[entity_name] = entity_id
choices.append(entity_name)
return proximity_html, choices, True
else:
proximity_html = "<div style='color: #666;'>🔍 Move near NPCs or players to chat privately</div>"
return proximity_html, [], False
except Exception as e:
return f"<div style='color: #f44336;'>Error: {str(e)}</div>", [], False
def _get_formatted_world_events(self) -> str:
"""Get formatted world events for display."""
try:
# Get world events from the game engine
world_events = self.game_facade.get_world_events()
if not world_events:
return "No recent world events...\n\n💡 Tip: Walk near NPCs (📮🏪🌤️) to interact with them!\nThen visit the 'NPC Add-ons' tab to use their features."
# Format events with HTML line breaks
formatted_events = []
for event in world_events[-10:]: # Show last 10 events
timestamp = event.get('timestamp', '')
event_text = event.get('event', '')
formatted_events.append(f"[{timestamp}] {event_text}")
return "\n".join(formatted_events)
except Exception as e:
print(f"Error formatting world events: {e}")
return "Error loading world events..."