File size: 18,097 Bytes
4c75d73 6c6097e 4c75d73 6c6097e 4c75d73 6c6097e 4c75d73 6c6097e 4c75d73 6c6097e 4c75d73 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 |
"""
NPC management service implementation.
"""
import random
import time
import threading
import os
from typing import Dict, Optional, List
from ..interfaces.service_interfaces import INPCService
from ..interfaces.game_interfaces import IGameWorld
from ..interfaces.npc_interfaces import INPCBehavior
class DonaldBehavior(INPCBehavior):
"""Deal maker NPC behavior with funny responses."""
def __init__(self):
self.responses = [
"Listen, I make the best deals. Nobody makes deals like me, believe me!",
"This game is tremendous, absolutely tremendous. I helped design it, you know.",
"I have the best NPCs, the most fantastic NPCs you've ever seen!",
"Fake news! I never said that. Well, maybe I did, but it was taken out of context.",
"You know what? You're fired! Just kidding, I can't fire you here.",
"I built a wall around my shop. Mexico paid for it. Well, not really, but...",
"My inventory is huge, absolutely huge. Some say it's the biggest ever.",
"I tweeted about this earlier. Did you see my tweet? It was perfect.",
"Nobody knows trading like I know trading. I wrote the book on it!",
"This is a witch hunt! I've done nothing wrong! Perfect conversation!",
"I have tremendous respect for you, tremendous. But you're wrong.",
"Covfefe! Wait, wrong universe. But you get the idea, right?",
"I'm a stable genius when it comes to trading. Very stable, very genius.",
"That's fake gold! I only deal in real gold. 24 karat, the best gold.",
"I love all my customers equally. Except the ones who don't buy anything."
]
self.greetings = [
"Welcome to Donald's Tremendous Trading Post! The best shop, believe me!",
"You've never seen deals like these. Nobody has deals like these!",
"Step right up to the most incredible shop in the entire game world!",
"I'm Donald, and I approve this shopping experience. It's tremendous!",
"Welcome! I have the best items, folks. The absolute best!"
]
def get_response(self, npc_id: str, message: str, player_id: str = None) -> str:
"""Generate Deal Maker-style response."""
# Check for keywords to give specific responses
message_lower = message.lower()
if any(word in message_lower for word in ['buy', 'sell', 'trade', 'shop', 'item']):
trade_responses = [
"You want to make a deal? I love deals! What do you got?",
"Trading is my specialty. I literally wrote 'The Art of the Deal'!",
"Let me tell you about my inventory - it's incredible, absolutely incredible!",
"Nobody, and I mean NOBODY, has better prices than me!",
"This is the best trade you'll ever make, believe me!"
]
return random.choice(trade_responses)
elif any(word in message_lower for word in ['hello', 'hi', 'hey', 'greetings']):
return random.choice(self.greetings)
elif any(word in message_lower for word in ['help', 'assist', 'support']):
help_responses = [
"Help? I don't need help! But I'll help you because I'm generous.",
"The best help you can get is right here, from me, Donald!",
"I'm always helping people. It's what I do. I'm very helpful.",
"You came to the right NPC for help. The absolute right NPC!"
]
return random.choice(help_responses)
elif any(word in message_lower for word in ['wall', 'build', 'construction']):
return "I built the most beautiful wall around my shop. It's incredible!"
elif any(word in message_lower for word in ['fake', 'news', 'media']):
return "Fake news! The media is so dishonest. But you seem trustworthy!"
else:
return random.choice(self.responses)
def get_greeting(self, npc_id: str, player_id: str = None) -> str:
"""Get greeting message."""
return random.choice(self.greetings)
def can_handle(self, npc_id: str) -> bool:
"""Check if this behavior handles Donald NPC."""
return npc_id == "donald"
class GenericNPCBehavior(INPCBehavior):
"""Generic NPC behavior for standard NPCs."""
def __init__(self):
self.personality_responses = {
'merchant': [
"Welcome to my shop! What would you like to buy?",
"I have the finest items in the realm!",
"Special discount today - 10% off all potions!"
],
'wise_teacher': [
"Ah, a curious mind! What would you like to learn?",
"Knowledge is the greatest treasure. Ask me anything!",
"I have studied the ancient texts for decades.",
"Wisdom comes through questioning. What puzzles you?"
],
'comedian': [
"Haha! Want to hear a joke? Why don't skeletons fight? They don't have the guts!",
"What do you call a sleeping bull? A bulldozer! *laughs*",
"I've got jokes for days! Life's too short not to laugh!",
"Why did the scarecrow win an award? He was outstanding in his field!"
],
'tough_trainer': [
"Ready for combat training? Show me your stance!",
"A true warrior trains every day. Are you committed?",
"Strength comes from discipline and practice!",
"The blade is an extension of your will. Focus!"
],
'gentle_healer': [
"Blessings upon you, traveler. Do you need healing?",
"The light guides my hands. I can mend your wounds.",
"Health of body and spirit go hand in hand.",
"May the divine light restore your vitality!"
],
'mystical_sage': [
"Magic flows through all things, young one.",
"The arcane arts require patience and understanding.",
"Ancient powers stir... can you feel them?",
"Wisdom and magic are closely intertwined."
],
'traveler': [
"The road calls to me... always moving, always exploring!",
"I've seen wonders beyond imagination in my travels.",
"Adventure awaits around every corner!",
"Sometimes the journey is more important than the destination."
]
}
def get_response(self, npc_id: str, message: str, player_id: str = None) -> str:
"""Generate generic NPC response based on personality."""
# This would typically get NPC data from the world
# For now, we'll use a simple mapping
personality_map = {
'merchant': 'merchant',
'scholar': 'wise_teacher',
'jester': 'comedian',
'warrior': 'tough_trainer',
'healer': 'gentle_healer',
'sage': 'mystical_sage',
'wanderer': 'traveler'
}
personality = personality_map.get(npc_id, 'merchant')
responses = self.personality_responses.get(personality, [
"Hello there, traveler!",
"How can I help you?",
"Nice weather we're having."
])
return random.choice(responses)
def get_greeting(self, npc_id: str, player_id: str = None) -> str:
"""Get greeting message."""
return self.get_response(npc_id, "hello", player_id)
def can_handle(self, npc_id: str) -> bool:
"""This behavior can handle any NPC."""
return True
class NPCService(INPCService):
"""Service for managing NPCs in the game world."""
def __init__(self, game_world: IGameWorld):
self.game_world = game_world
self.behaviors = [
DonaldBehavior(),
GenericNPCBehavior()
]
# Movement system
self._movement_thread = None
self._movement_active = False
# Only start movement system if not in test environment
if not os.environ.get('PYTEST_CURRENT_TEST') and not hasattr(game_world, '_is_test_instance'):
self.start_movement_system()
def start_movement_system(self):
"""Start the NPC movement system for moving NPCs like Roaming Rick."""
if self._movement_thread is None or not self._movement_thread.is_alive():
self._movement_active = True
self._movement_thread = threading.Thread(target=self._movement_loop, daemon=True)
self._movement_thread.start()
def stop_movement_system(self):
"""Stop the NPC movement system."""
self._movement_active = False
if self._movement_thread and self._movement_thread.is_alive():
self._movement_thread.join(timeout=1.0)
self._movement_thread = None
def _movement_loop(self):
"""Main movement loop for updating moving NPCs."""
while self._movement_active:
try:
self.update_moving_npcs()
time.sleep(2.0) # Update every 2 seconds
except Exception as e:
print(f"[NPCService] Movement loop error: {e}") # Break the loop on critical errors during testing
if "Mock" in str(e):
print(f"[NPCService] Stopping movement loop due to mock error")
break
time.sleep(5.0) # Wait longer on error
def update_moving_npcs(self):
"""Update positions of moving NPCs like Roaming Rick."""
current_time = time.time()
try:
npcs = self.get_all_npcs()
# Handle case when npcs is a Mock object during testing
if hasattr(npcs, '__iter__'):
try:
npc_items = npcs.items() if hasattr(npcs, 'items') else []
except Exception:
# If iteration fails (e.g., with Mock), just return
print("[NPCService] Error in update_moving_npcs: Mock object iteration failed")
return
else:
print("[NPCService] Error in update_moving_npcs: npcs is not iterable")
return
for npc_id, npc in npc_items:
# Check if this NPC has movement configuration
if npc.get('type') == 'moving' and 'movement' in npc:
movement = npc['movement']
# Check if enough time has passed for movement (based on speed)
time_since_last_move = current_time - movement.get('last_move', 0)
movement_interval = 2.0 / movement.get('speed', 1) # Faster speed = shorter interval
if time_since_last_move >= movement_interval:
# Store old position
old_x, old_y = npc['x'], npc['y']
# Calculate new position
move_distance = 25 # Same as player movement
direction_x = movement.get('direction_x', 1)
direction_y = movement.get('direction_y', 1)
new_x = npc['x'] + (move_distance * direction_x)
new_y = npc['y'] + (move_distance * direction_y)
# Check boundaries and bounce if needed (assuming 800x600 world)
world_width = getattr(self.game_world, 'world_width', 475)
world_height = getattr(self.game_world, 'world_height', 375)
if new_x <= 0 or new_x >= world_width:
direction_x *= -1 # Reverse X direction
new_x = max(0, min(world_width, new_x))
if new_y <= 0 or new_y >= world_height:
direction_y *= -1 # Reverse Y direction
new_y = max(0, min(world_height, new_y))
# Update NPC position and movement data
npc['x'] = new_x
npc['y'] = new_y
movement['direction_x'] = direction_x
movement['direction_y'] = direction_y
movement['last_move'] = current_time
# Add world event if NPC moved
if old_x != new_x or old_y != new_y:
if hasattr(self.game_world, 'add_world_event'):
self.game_world.add_world_event(f"🚶 {npc['name']} roams to ({int(new_x)}, {int(new_y)})")
# Check if any players are now near this moved NPC
try:
players = self.game_world.get_all_players()
for player_id, player in players.items():
distance = ((player.x - new_x)**2 + (player.y - new_y)**2)**0.5
if distance < 50: # Close enough to notice
if hasattr(self.game_world, 'add_world_event'):
self.game_world.add_world_event(f"👀 {player.name} notices {npc['name']} nearby")
except Exception as e:
print(f"[NPCService] Error checking player proximity: {e}")
except Exception as e:
print(f"[NPCService] Error in update_moving_npcs: {e}")
def get_npc_response(self, npc_id: str, message: str, player_id: str = None) -> str:
"""Get NPC response to player message."""
try:
# Find appropriate behavior for this NPC
for behavior in self.behaviors:
if behavior.can_handle(npc_id):
return behavior.get_response(npc_id, message, player_id)
# Fallback response
return "I don't understand what you're saying."
except Exception as e:
print(f"[NPCService] Error generating response: {e}")
return "Sorry, I'm having trouble understanding you right now."
def register_npc(self, npc_id: str, npc_data: Dict) -> bool:
"""Register a new NPC in the game world."""
try:
if hasattr(self.game_world, 'add_npc'):
self.game_world.add_npc(npc_id, npc_data)
return True
else:
# Direct access to npcs dict
npcs = getattr(self.game_world, 'npcs', {})
npcs[npc_id] = npc_data
return True
except Exception as e:
print(f"[NPCService] Error registering NPC: {e}")
return False
def get_npc(self, npc_id: str) -> Optional[Dict]:
"""Get NPC data by ID."""
try:
if hasattr(self.game_world, 'get_npc'):
return self.game_world.get_npc(npc_id)
else:
npcs = getattr(self.game_world, 'npcs', {})
return npcs.get(npc_id)
except Exception as e:
print(f"[NPCService] Error getting NPC: {e}")
return None
def update_npc_position(self, npc_id: str, x: int, y: int) -> bool:
"""Update NPC position."""
try:
npc = self.get_npc(npc_id)
if npc:
npc['x'] = x
npc['y'] = y
return True
return False
except Exception as e:
print(f"[NPCService] Error updating NPC position: {e}")
return False
def get_npc_greeting(self, npc_id: str, player_id: str = None) -> str:
"""Get greeting message from NPC."""
for behavior in self.behaviors:
if behavior.can_handle(npc_id):
return behavior.get_greeting(npc_id, player_id)
return "Hello, traveler!"
def add_behavior(self, behavior: INPCBehavior) -> bool:
"""Add a new NPC behavior."""
try:
# Insert before the generic behavior (which should be last)
self.behaviors.insert(-1, behavior)
return True
except Exception as e:
print(f"[NPCService] Error adding behavior: {e}")
return False
def remove_behavior(self, behavior_class: type) -> bool:
"""Remove an NPC behavior by class."""
try:
self.behaviors = [b for b in self.behaviors if not isinstance(b, behavior_class)]
return True
except Exception as e:
print(f"[NPCService] Error removing behavior: {e}")
return False
def get_all_npcs(self) -> Dict[str, Dict]:
"""Get all NPCs from the game world."""
try:
if hasattr(self.game_world, 'get_all_npcs'):
return self.game_world.get_all_npcs()
else:
return getattr(self.game_world, 'npcs', {})
except Exception as e:
print(f"[NPCService] Error getting all NPCs: {e}")
return {}
|