""" Central stat calculation service. All derived stats are computed here from base attributes + equipment + buffs. Results are cached in Redis and invalidated on any change. """ import json from typing import Dict, Any, Optional, List from .. import database as db from ..items import items_manager as ITEMS_MANAGER # ─── Game Config (raise per expansion) ─── STAT_CAP = 50 # Max base stat points per attribute MAX_LEVEL = 60 # Max character level POINTS_PER_LEVEL = 1 # Stat points granted per level async def calculate_derived_stats(character_id: int, redis_mgr=None) -> Dict[str, Any]: """ Calculate all derived stats for a character. Checks Redis cache first; if miss, computes from DB and caches result. Returns dict with all derived stat values. """ # 1. Check Redis cache if redis_mgr and redis_mgr.redis_client: try: cached = await redis_mgr.redis_client.get(f"stats:{character_id}") if cached: return json.loads(cached) except Exception: pass # Graceful degradation — recalculate if Redis fails # 2. Fetch data from DB char = await db.get_player_by_id(character_id) if not char: return _empty_stats() equipment = await db.get_all_equipment(character_id) effects = await db.get_player_effects(character_id) # 3. Fetch owned perks owned_perks = await db.get_character_perks(character_id) owned_perk_ids = [row['perk_id'] for row in owned_perks] # 4. Compute derived stats stats = _compute_stats(char, equipment, effects, owned_perk_ids) # 5. Cache in Redis (5 min TTL) if redis_mgr and redis_mgr.redis_client: try: await redis_mgr.redis_client.setex( f"stats:{character_id}", 300, json.dumps(stats) ) except Exception: pass return stats def _compute_stats(char: Dict[str, Any], equipment: Dict[str, Any], effects: List[Dict], perk_ids: List[str] = None) -> Dict[str, Any]: """Pure computation of derived stats from base data.""" strength = char.get('strength', 0) agility = char.get('agility', 0) endurance = char.get('endurance', 0) intellect = char.get('intellect', 0) level = char.get('level', 1) if perk_ids is None: perk_ids = [] # ─── Base derived stats from attributes ─── attack_power = 5 + int(strength * 1.5) + level crit_chance = 0.05 + (agility * 0.005) crit_damage = 1.5 + (strength * 0.01) dodge_chance = min(0.25, 0.02 + (agility * 0.005)) # Cap 25% flee_chance_base = 0.4 + (agility * 0.01) max_hp = 30 + (endurance * 5) + (level * 3) max_stamina = 20 + (endurance * 2) + level status_resistance = endurance * 0.01 block_chance = 0.0 item_effectiveness = 1.0 + (intellect * 0.02) xp_bonus = 1.0 + (intellect * 0.01) loot_quality = 1.0 + (intellect * 0.005) crafting_bonus = intellect * 0.01 carry_weight = 10.0 + (strength * 0.5) # ─── Equipment bonuses ─── total_armor = 0 weapon_crit = 0.0 weapon_damage_min = 0 weapon_damage_max = 0 has_shield = False for slot, item_data in equipment.items(): if not item_data or not item_data.get('item_id'): continue # Get inventory item to find the item definition inv_item_sync = item_data # equipment dict already has item_id reference item_def = ITEMS_MANAGER.get_item(inv_item_sync.get('item_id', '')) # Try to get item_id from the inventory item if the direct lookup failed if not item_def: continue if item_def.stats: total_armor += item_def.stats.get('armor', 0) weapon_crit += item_def.stats.get('crit_chance', 0) if slot == 'weapon': weapon_damage_min = item_def.stats.get('damage_min', 0) weapon_damage_max = item_def.stats.get('damage_max', 0) if slot == 'offhand': has_shield = True # Apply equipment to derived stats crit_chance += weapon_crit armor_reduction = total_armor / (total_armor + 50) if total_armor > 0 else 0.0 if has_shield: block_chance = strength * 0.003 # ─── Buff effects ─── for effect in effects: effect_name = effect.get('effect_name', '') value = effect.get('value', 0) # Future: apply buff modifiers here # ─── Perk passive bonuses ─── if 'thick_skin' in perk_ids: max_hp = int(max_hp * 1.10) # +10% max HP if 'lucky_strike' in perk_ids: crit_chance += 0.05 # +5% crit chance if 'quick_learner' in perk_ids: xp_bonus *= 1.15 # +15% XP if 'glass_cannon' in perk_ids: attack_power = int(attack_power * 1.30) # +30% attack max_hp = int(max_hp * 0.80) # -20% HP if 'survivor' in perk_ids: max_hp = int(max_hp * 1.02) # Small HP boost from regen perk if 'scavenger' in perk_ids: loot_quality *= 1.10 # +10% loot quality if 'fleet_footed' in perk_ids: # Travel stamina reduction tracked for movement system pass stats = { # Core combat "attack_power": attack_power, "crit_chance": round(crit_chance, 4), "crit_damage": round(crit_damage, 2), "dodge_chance": round(dodge_chance, 4), "flee_chance_base": round(flee_chance_base, 2), # Vitals "max_hp": max_hp, "max_stamina": max_stamina, # Defense "total_armor": total_armor, "armor_reduction": round(armor_reduction, 4), "block_chance": round(block_chance, 4), "status_resistance": round(status_resistance, 4), # Utility "item_effectiveness": round(item_effectiveness, 2), "xp_bonus": round(xp_bonus, 2), "loot_quality": round(loot_quality, 3), "crafting_bonus": round(crafting_bonus, 2), "carry_weight": round(carry_weight, 1), # Weapon info "weapon_damage_min": weapon_damage_min, "weapon_damage_max": weapon_damage_max, "has_shield": has_shield, # Perk flags "has_last_stand": 'last_stand' in perk_ids, "has_resilient": 'resilient' in perk_ids, "has_iron_fist": 'iron_fist' in perk_ids, "has_heavy_hitter": 'heavy_hitter' in perk_ids, } return stats def _empty_stats() -> Dict[str, Any]: """Default stats for error cases.""" return { "attack_power": 5, "crit_chance": 0.05, "crit_damage": 1.5, "dodge_chance": 0.02, "flee_chance_base": 0.4, "max_hp": 30, "max_stamina": 20, "total_armor": 0, "armor_reduction": 0.0, "block_chance": 0.0, "status_resistance": 0.0, "item_effectiveness": 1.0, "xp_bonus": 1.0, "loot_quality": 1.0, "crafting_bonus": 0.0, "carry_weight": 10.0, "weapon_damage_min": 0, "weapon_damage_max": 0, "has_shield": False, } async def invalidate_stats_cache(character_id: int, redis_mgr=None): """ Delete cached stats for a character. Call this whenever: - Equipment changes (equip/unequip/break) - Stat points allocated - Level up - Buff applied/expired """ if redis_mgr and redis_mgr.redis_client: try: await redis_mgr.redis_client.delete(f"stats:{character_id}") except Exception: pass def get_flee_chance(flee_chance_base: float, enemy_level: int) -> float: """Calculate actual flee chance against a specific enemy.""" return max(0.1, min(0.9, flee_chance_base - (enemy_level * 0.02)))