""" 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() raw_equipment = await db.get_all_equipment(character_id) enriched_equipment = {} for slot, item_data in raw_equipment.items(): if not item_data or not item_data.get('item_id'): continue inv_item = await db.get_inventory_item_by_id(item_data['item_id']) if not inv_item: continue enriched_item = { 'item_id': inv_item['item_id'], # String ID 'inventory_id': item_data['item_id'] } unique_item_id = inv_item.get('unique_item_id') if unique_item_id: unique_item = await db.get_unique_item(unique_item_id) if unique_item and unique_item.get('unique_stats'): enriched_item['unique_stats'] = unique_item['unique_stats'] enriched_equipment[slot] = enriched_item 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, enriched_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 item_id_str = item_data.get('item_id', '') item_def = ITEMS_MANAGER.get_item(item_id_str) if not item_def: continue # Merge base stats and unique stats merged_stats = {} if item_def.stats: merged_stats.update(item_def.stats) if item_data.get('unique_stats'): merged_stats.update(item_data['unique_stats']) if merged_stats: total_armor += merged_stats.get('armor', 0) weapon_crit += merged_stats.get('crit_chance', 0) max_hp += merged_stats.get('max_hp', 0) max_stamina += merged_stats.get('max_stamina', 0) carry_weight += merged_stats.get('weight_capacity', 0) if slot == 'weapon': weapon_damage_min = merged_stats.get('damage_min', 0) weapon_damage_max = merged_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 # Sync derived max_hp and max_stamina to the database characters table try: derived = await calculate_derived_stats(character_id, redis_mgr) char = await db.get_player_by_id(character_id) if char: new_max_hp = derived.get('max_hp', char['max_hp']) new_max_stamina = derived.get('max_stamina', char['max_stamina']) if new_max_hp != char['max_hp'] or new_max_stamina != char['max_stamina']: new_hp = min(char['hp'], new_max_hp) new_stamina = min(char['stamina'], new_max_stamina) await db.update_player( character_id, max_hp=new_max_hp, max_stamina=new_max_stamina, hp=new_hp, stamina=new_stamina ) except Exception as e: import logging logging.getLogger(__name__).warning(f"Failed to sync derived stats to DB for {character_id}: {e}") 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)))