226 lines
7.7 KiB
Python
226 lines
7.7 KiB
Python
"""
|
|
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)))
|