120 lines
4.2 KiB
Python
120 lines
4.2 KiB
Python
"""
|
|
Global Wandering Enemy Spawn Manager
|
|
Runs periodically to spawn/despawn enemies based on location danger levels.
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
import random
|
|
from typing import Dict, List
|
|
from bot import database
|
|
from data.npcs import (
|
|
LOCATION_SPAWNS,
|
|
LOCATION_DANGER,
|
|
get_random_npc_for_location,
|
|
get_wandering_enemy_chance
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# Configuration
|
|
SPAWN_CHECK_INTERVAL = 120 # Check every 2 minutes
|
|
ENEMY_LIFETIME = 600 # Enemies live for 10 minutes
|
|
MAX_ENEMIES_PER_LOCATION = {
|
|
0: 0, # Safe zones - no wandering enemies
|
|
1: 1, # Low danger - max 1 enemy
|
|
2: 2, # Medium danger - max 2 enemies
|
|
3: 3, # High danger - max 3 enemies
|
|
4: 4, # Extreme danger - max 4 enemies
|
|
}
|
|
|
|
|
|
def get_danger_level(location_id: str) -> int:
|
|
"""Get danger level for a location."""
|
|
danger_data = LOCATION_DANGER.get(location_id, (0, 0.0, 0.0))
|
|
return danger_data[0]
|
|
|
|
|
|
async def spawn_manager_loop():
|
|
"""
|
|
Main spawn manager loop.
|
|
Runs continuously, checking spawn conditions every SPAWN_CHECK_INTERVAL seconds.
|
|
"""
|
|
logger.info("🎲 Spawn Manager started")
|
|
|
|
while True:
|
|
try:
|
|
await asyncio.sleep(SPAWN_CHECK_INTERVAL)
|
|
|
|
# Clean up expired enemies first
|
|
despawned_count = await database.cleanup_expired_wandering_enemies()
|
|
if despawned_count > 0:
|
|
logger.info(f"🧹 Cleaned up {despawned_count} expired wandering enemies")
|
|
|
|
# Process each location
|
|
spawned_count = 0
|
|
for location_id, spawn_table in LOCATION_SPAWNS.items():
|
|
if not spawn_table:
|
|
continue # Skip locations with no spawns
|
|
|
|
# Get danger level and max enemies for this location
|
|
danger_level = get_danger_level(location_id)
|
|
max_enemies = MAX_ENEMIES_PER_LOCATION.get(danger_level, 0)
|
|
|
|
if max_enemies == 0:
|
|
continue # Skip safe zones
|
|
|
|
# Check current enemy count
|
|
current_count = await database.get_wandering_enemy_count_in_location(location_id)
|
|
|
|
if current_count >= max_enemies:
|
|
continue # Location is at capacity
|
|
|
|
# Calculate spawn chance based on wandering_enemy_chance
|
|
spawn_chance = get_wandering_enemy_chance(location_id)
|
|
|
|
# Attempt to spawn enemies up to max capacity
|
|
for _ in range(max_enemies - current_count):
|
|
if random.random() < spawn_chance:
|
|
# Spawn an enemy
|
|
npc_id = get_random_npc_for_location(location_id)
|
|
if npc_id:
|
|
await database.spawn_wandering_enemy(
|
|
npc_id=npc_id,
|
|
location_id=location_id,
|
|
lifetime_seconds=ENEMY_LIFETIME
|
|
)
|
|
spawned_count += 1
|
|
logger.info(f"👹 Spawned {npc_id} at {location_id} (current: {current_count + 1}/{max_enemies})")
|
|
|
|
if spawned_count > 0:
|
|
logger.info(f"✨ Spawn cycle complete: {spawned_count} enemies spawned")
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Error in spawn manager loop: {e}", exc_info=True)
|
|
# Continue running even if there's an error
|
|
await asyncio.sleep(10)
|
|
|
|
|
|
async def start_spawn_manager():
|
|
"""Start the spawn manager as a background task."""
|
|
asyncio.create_task(spawn_manager_loop())
|
|
logger.info("🎮 Spawn Manager initialized")
|
|
|
|
|
|
async def get_spawn_stats() -> Dict:
|
|
"""Get statistics about current spawns (for debugging/monitoring)."""
|
|
all_enemies = await database.get_all_active_wandering_enemies()
|
|
|
|
# Count by location
|
|
location_counts = {}
|
|
for enemy in all_enemies:
|
|
loc = enemy['location_id']
|
|
location_counts[loc] = location_counts.get(loc, 0) + 1
|
|
|
|
return {
|
|
"total_active": len(all_enemies),
|
|
"by_location": location_counts,
|
|
"enemies": all_enemies
|
|
}
|