feat(api): implement combat loot integration with hybrid static/procedural system

Add CombatLootService that orchestrates loot generation from combat,
supporting both static item drops (consumables, materials) and procedural
equipment generation (weapons, armor with affixes).

Key changes:
- Extend LootEntry model with LootType enum (STATIC/PROCEDURAL)
- Create StaticItemLoader service for consumables/materials from YAML
- Create CombatLootService with full rarity formula incorporating:
  - Party average level
  - Enemy difficulty tier (EASY: +0%, MEDIUM: +5%, HARD: +15%, BOSS: +30%)
  - Character luck stat
  - Per-entry rarity bonus
- Integrate with CombatService._calculate_rewards() for automatic loot gen
- Add boss guaranteed drops via generate_boss_loot()

New enemy variants (goblin family proof-of-concept):
- goblin_scout (Easy) - static drops only
- goblin_warrior (Medium) - static + procedural weapon drops
- goblin_chieftain (Hard) - static + procedural weapon/armor drops

Static items added:
- consumables.yaml: health/mana potions, elixirs, food
- materials.yaml: trophy items, crafting materials

Tests: 59 new tests across 3 test files (all passing)
This commit is contained in:
2025-11-27 00:01:17 -06:00
parent a38906b445
commit fdd48034e4
14 changed files with 2257 additions and 26 deletions

View File

@@ -0,0 +1,359 @@
"""
Combat Loot Service - Orchestrates loot generation from combat encounters.
This service bridges the EnemyTemplate loot tables with both the StaticItemLoader
(for consumables and materials) and ItemGenerator (for procedural equipment).
The service calculates effective rarity based on:
- Party average level
- Enemy difficulty tier
- Character luck stat
- Optional loot bonus modifiers (from abilities, buffs, etc.)
"""
import random
from dataclasses import dataclass
from typing import List, Optional
from app.models.enemy import EnemyTemplate, LootEntry, LootType, EnemyDifficulty
from app.models.items import Item
from app.services.item_generator import get_item_generator, ItemGenerator
from app.services.static_item_loader import get_static_item_loader, StaticItemLoader
from app.utils.logging import get_logger
logger = get_logger(__file__)
# Difficulty tier rarity bonuses (converted to effective luck points)
# Higher difficulty enemies have better chances of dropping rare items
DIFFICULTY_RARITY_BONUS = {
EnemyDifficulty.EASY: 0.0,
EnemyDifficulty.MEDIUM: 0.05,
EnemyDifficulty.HARD: 0.15,
EnemyDifficulty.BOSS: 0.30,
}
# Multiplier for converting rarity bonus to effective luck points
# Each 0.05 bonus translates to +1 effective luck
LUCK_CONVERSION_FACTOR = 20
@dataclass
class LootContext:
"""
Context for loot generation calculations.
Provides all the factors that influence loot quality and rarity.
Attributes:
party_average_level: Average level of player characters in the encounter
enemy_difficulty: Difficulty tier of the enemy being looted
luck_stat: Party's luck stat (typically average or leader's luck)
loot_bonus: Additional bonus from abilities, buffs, or modifiers (0.0 to 1.0)
"""
party_average_level: int = 1
enemy_difficulty: EnemyDifficulty = EnemyDifficulty.EASY
luck_stat: int = 8
loot_bonus: float = 0.0
class CombatLootService:
"""
Service for generating combat loot drops.
Supports two types of loot:
- STATIC: Predefined items loaded from YAML (consumables, materials)
- PROCEDURAL: Generated equipment with affixes (weapons, armor)
The service handles:
- Rolling for drops based on drop_chance
- Loading static items via StaticItemLoader
- Generating procedural items via ItemGenerator
- Calculating effective rarity based on context
"""
def __init__(
self,
item_generator: Optional[ItemGenerator] = None,
static_loader: Optional[StaticItemLoader] = None
):
"""
Initialize the combat loot service.
Args:
item_generator: ItemGenerator instance (uses global singleton if None)
static_loader: StaticItemLoader instance (uses global singleton if None)
"""
self.item_generator = item_generator or get_item_generator()
self.static_loader = static_loader or get_static_item_loader()
logger.info("CombatLootService initialized")
def generate_loot_from_enemy(
self,
enemy: EnemyTemplate,
context: LootContext
) -> List[Item]:
"""
Generate all loot drops from a defeated enemy.
Iterates through the enemy's loot table, rolling for each entry
and generating appropriate items based on loot type.
Args:
enemy: The defeated enemy template
context: Loot generation context (party level, luck, etc.)
Returns:
List of Item objects to add to player inventory
"""
items = []
for entry in enemy.loot_table:
# Roll for drop chance
if random.random() >= entry.drop_chance:
continue
# Determine quantity
quantity = random.randint(entry.quantity_min, entry.quantity_max)
if entry.loot_type == LootType.STATIC:
# Static item: load from predefined templates
static_items = self._generate_static_items(entry, quantity)
items.extend(static_items)
elif entry.loot_type == LootType.PROCEDURAL:
# Procedural equipment: generate with ItemGenerator
procedural_items = self._generate_procedural_items(
entry, quantity, context
)
items.extend(procedural_items)
logger.info(
"Loot generated from enemy",
enemy_id=enemy.enemy_id,
enemy_difficulty=enemy.difficulty.value,
item_count=len(items),
party_level=context.party_average_level,
luck=context.luck_stat
)
return items
def _generate_static_items(
self,
entry: LootEntry,
quantity: int
) -> List[Item]:
"""
Generate static items from a loot entry.
Args:
entry: The loot table entry
quantity: Number of items to generate
Returns:
List of Item instances
"""
items = []
if not entry.item_id:
logger.warning(
"Static loot entry missing item_id",
entry=entry.to_dict()
)
return items
for _ in range(quantity):
item = self.static_loader.get_item(entry.item_id)
if item:
items.append(item)
else:
logger.warning(
"Failed to load static item",
item_id=entry.item_id
)
return items
def _generate_procedural_items(
self,
entry: LootEntry,
quantity: int,
context: LootContext
) -> List[Item]:
"""
Generate procedural items from a loot entry.
Calculates effective luck based on:
- Base luck stat
- Entry-specific rarity bonus
- Difficulty bonus
- Loot bonus from abilities/buffs
Args:
entry: The loot table entry
quantity: Number of items to generate
context: Loot generation context
Returns:
List of generated Item instances
"""
items = []
if not entry.item_type:
logger.warning(
"Procedural loot entry missing item_type",
entry=entry.to_dict()
)
return items
# Calculate effective luck for rarity roll
effective_luck = self._calculate_effective_luck(entry, context)
for _ in range(quantity):
item = self.item_generator.generate_loot_drop(
character_level=context.party_average_level,
luck_stat=effective_luck,
item_type=entry.item_type
)
if item:
items.append(item)
else:
logger.warning(
"Failed to generate procedural item",
item_type=entry.item_type,
level=context.party_average_level
)
return items
def _calculate_effective_luck(
self,
entry: LootEntry,
context: LootContext
) -> int:
"""
Calculate effective luck for rarity rolling.
Combines multiple factors:
- Base luck stat from party
- Entry-specific rarity bonus (defined per loot entry)
- Difficulty bonus (based on enemy tier)
- Loot bonus (from abilities, buffs, etc.)
The formula:
effective_luck = base_luck + (entry_bonus + difficulty_bonus + loot_bonus) * FACTOR
Args:
entry: The loot table entry
context: Loot generation context
Returns:
Effective luck stat for rarity calculations
"""
# Get difficulty bonus
difficulty_bonus = DIFFICULTY_RARITY_BONUS.get(
context.enemy_difficulty, 0.0
)
# Sum all bonuses
total_bonus = (
entry.rarity_bonus +
difficulty_bonus +
context.loot_bonus
)
# Convert bonus to effective luck points
bonus_luck = int(total_bonus * LUCK_CONVERSION_FACTOR)
effective_luck = context.luck_stat + bonus_luck
logger.debug(
"Effective luck calculated",
base_luck=context.luck_stat,
entry_bonus=entry.rarity_bonus,
difficulty_bonus=difficulty_bonus,
loot_bonus=context.loot_bonus,
total_bonus=total_bonus,
effective_luck=effective_luck
)
return effective_luck
def generate_boss_loot(
self,
enemy: EnemyTemplate,
context: LootContext,
guaranteed_drops: int = 1
) -> List[Item]:
"""
Generate loot from a boss enemy with guaranteed drops.
Boss enemies are guaranteed to drop at least one piece of equipment
in addition to their normal loot table rolls.
Args:
enemy: The boss enemy template
context: Loot generation context
guaranteed_drops: Number of guaranteed equipment drops
Returns:
List of Item objects including guaranteed drops
"""
# Generate normal loot first
items = self.generate_loot_from_enemy(enemy, context)
# Add guaranteed procedural drops for bosses
if enemy.is_boss():
context_for_boss = LootContext(
party_average_level=context.party_average_level,
enemy_difficulty=EnemyDifficulty.BOSS,
luck_stat=context.luck_stat,
loot_bonus=context.loot_bonus + 0.1 # Extra bonus for bosses
)
for _ in range(guaranteed_drops):
# Alternate between weapon and armor
item_type = random.choice(["weapon", "armor"])
effective_luck = self._calculate_effective_luck(
LootEntry(
loot_type=LootType.PROCEDURAL,
item_type=item_type,
rarity_bonus=0.15 # Boss-tier bonus
),
context_for_boss
)
item = self.item_generator.generate_loot_drop(
character_level=context.party_average_level,
luck_stat=effective_luck,
item_type=item_type
)
if item:
items.append(item)
logger.info(
"Boss loot generated",
enemy_id=enemy.enemy_id,
guaranteed_drops=guaranteed_drops,
total_items=len(items)
)
return items
# Global singleton
_service_instance: Optional[CombatLootService] = None
def get_combat_loot_service() -> CombatLootService:
"""
Get the global CombatLootService instance.
Returns:
Singleton CombatLootService instance
"""
global _service_instance
if _service_instance is None:
_service_instance = CombatLootService()
return _service_instance

View File

@@ -15,7 +15,7 @@ from uuid import uuid4
from app.models.combat import Combatant, CombatEncounter
from app.models.character import Character
from app.models.enemy import EnemyTemplate
from app.models.enemy import EnemyTemplate, EnemyDifficulty
from app.models.stats import Stats
from app.models.abilities import Ability, AbilityLoader
from app.models.effects import Effect
@@ -25,6 +25,11 @@ from app.services.damage_calculator import DamageCalculator, DamageResult
from app.services.enemy_loader import EnemyLoader, get_enemy_loader
from app.services.session_service import get_session_service
from app.services.character_service import get_character_service
from app.services.combat_loot_service import (
get_combat_loot_service,
CombatLootService,
LootContext
)
from app.utils.logging import get_logger
logger = get_logger(__file__)
@@ -197,6 +202,7 @@ class CombatService:
self.character_service = get_character_service()
self.enemy_loader = get_enemy_loader()
self.ability_loader = AbilityLoader()
self.loot_service = get_combat_loot_service()
logger.info("CombatService initialized")
@@ -898,6 +904,9 @@ class CombatService:
"""
Calculate and distribute rewards after victory.
Uses CombatLootService for loot generation, supporting both
static items (consumables) and procedural equipment.
Args:
encounter: Completed combat encounter
session: Game session
@@ -908,6 +917,9 @@ class CombatService:
"""
rewards = CombatRewards()
# Build loot context from encounter
loot_context = self._build_loot_context(encounter)
# Sum up rewards from defeated enemies
for combatant in encounter.combatants:
if not combatant.is_player and combatant.is_dead():
@@ -919,9 +931,28 @@ class CombatService:
rewards.experience += enemy.experience_reward
rewards.gold += enemy.get_gold_reward()
# Roll for loot
loot = enemy.roll_loot()
rewards.items.extend(loot)
# Generate loot using the loot service
# Update context with this specific enemy's difficulty
enemy_context = LootContext(
party_average_level=loot_context.party_average_level,
enemy_difficulty=enemy.difficulty,
luck_stat=loot_context.luck_stat,
loot_bonus=loot_context.loot_bonus
)
# Use boss loot for boss enemies
if enemy.is_boss():
loot_items = self.loot_service.generate_boss_loot(
enemy, enemy_context
)
else:
loot_items = self.loot_service.generate_loot_from_enemy(
enemy, enemy_context
)
# Convert Item objects to dicts for serialization
for item in loot_items:
rewards.items.append(item.to_dict())
# Distribute rewards to player characters
player_combatants = [c for c in encounter.combatants if c.is_player]
@@ -964,6 +995,49 @@ class CombatService:
return rewards
def _build_loot_context(self, encounter: CombatEncounter) -> LootContext:
"""
Build loot generation context from a combat encounter.
Calculates:
- Party average level
- Party average luck stat
- Default difficulty (uses EASY, specific enemies override)
Args:
encounter: Combat encounter with player combatants
Returns:
LootContext for loot generation
"""
player_combatants = [c for c in encounter.combatants if c.is_player]
# Calculate party average level
if player_combatants:
# Use combatant's level if available, otherwise default to 1
levels = []
for p in player_combatants:
# Try to get level from stats or combatant
level = getattr(p, 'level', 1)
levels.append(level)
avg_level = sum(levels) // len(levels) if levels else 1
else:
avg_level = 1
# Calculate party average luck
if player_combatants:
luck_values = [p.stats.luck for p in player_combatants]
avg_luck = sum(luck_values) // len(luck_values) if luck_values else 8
else:
avg_luck = 8
return LootContext(
party_average_level=avg_level,
enemy_difficulty=EnemyDifficulty.EASY, # Default; overridden per-enemy
luck_stat=avg_luck,
loot_bonus=0.0 # Future: add buffs/abilities bonus
)
# =========================================================================
# Helper Methods
# =========================================================================

View File

@@ -0,0 +1,276 @@
"""
Static Item Loader Service - YAML-based static item loading.
This service loads predefined item definitions (consumables, materials, quest items)
from YAML files, providing a way to reference specific items by ID in loot tables.
Static items differ from procedurally generated items in that they have fixed
properties defined in YAML rather than randomly generated affixes.
"""
from pathlib import Path
from typing import Dict, List, Optional
import uuid
import yaml
from app.models.items import Item
from app.models.effects import Effect
from app.models.enums import ItemType, ItemRarity, EffectType
from app.utils.logging import get_logger
logger = get_logger(__file__)
class StaticItemLoader:
"""
Loads and manages static item definitions from YAML configuration files.
Static items are predefined items (consumables, materials, quest items)
that can be referenced by item_id in enemy loot tables.
Items are loaded from:
- api/app/data/static_items/consumables.yaml
- api/app/data/static_items/materials.yaml
Each call to get_item() creates a new Item instance with a unique ID,
so multiple drops of the same item_id become distinct inventory items.
"""
def __init__(self, data_dir: Optional[str] = None):
"""
Initialize the static item loader.
Args:
data_dir: Path to directory containing static item YAML files.
Defaults to /app/data/static_items/
"""
if data_dir is None:
# Default to app/data/static_items relative to this file
current_file = Path(__file__)
app_dir = current_file.parent.parent # Go up to /app
data_dir = str(app_dir / "data" / "static_items")
self.data_dir = Path(data_dir)
self._cache: Dict[str, dict] = {}
self._loaded = False
logger.info("StaticItemLoader initialized", data_dir=str(self.data_dir))
def _ensure_loaded(self) -> None:
"""Ensure items are loaded before any operation."""
if not self._loaded:
self._load_all()
def _load_all(self) -> None:
"""Load all static item YAML files."""
if not self.data_dir.exists():
logger.warning(
"Static items directory not found",
path=str(self.data_dir)
)
self._loaded = True
return
# Load all YAML files in the directory
for yaml_file in self.data_dir.glob("*.yaml"):
self._load_file(yaml_file)
self._loaded = True
logger.info("Static items loaded", count=len(self._cache))
def _load_file(self, yaml_file: Path) -> None:
"""
Load items from a single YAML file.
Args:
yaml_file: Path to the YAML file
"""
try:
with open(yaml_file, 'r') as f:
data = yaml.safe_load(f)
if data is None:
logger.warning("Empty YAML file", file=str(yaml_file))
return
items = data.get("items", {})
for item_id, item_data in items.items():
# Store the template data with its ID
item_data["_item_id"] = item_id
self._cache[item_id] = item_data
logger.debug(
"Static items loaded from file",
file=str(yaml_file),
count=len(items)
)
except Exception as e:
logger.error(
"Failed to load static items file",
file=str(yaml_file),
error=str(e)
)
def get_item(self, item_id: str, quantity: int = 1) -> Optional[Item]:
"""
Get an item instance by ID.
Creates a new Item instance with a unique ID for each call,
so multiple drops become distinct inventory items.
Args:
item_id: The static item ID (e.g., "health_potion_small")
quantity: Requested quantity (not used for individual item,
but available for future stackable item support)
Returns:
Item instance or None if item_id not found
"""
self._ensure_loaded()
template = self._cache.get(item_id)
if template is None:
logger.warning("Static item not found", item_id=item_id)
return None
# Create new instance with unique ID
instance_id = f"{item_id}_{uuid.uuid4().hex[:8]}"
# Parse item type
item_type_str = template.get("item_type", "quest_item")
try:
item_type = ItemType(item_type_str)
except ValueError:
logger.warning(
"Unknown item type, defaulting to quest_item",
item_type=item_type_str,
item_id=item_id
)
item_type = ItemType.QUEST_ITEM
# Parse rarity
rarity_str = template.get("rarity", "common")
try:
rarity = ItemRarity(rarity_str)
except ValueError:
logger.warning(
"Unknown rarity, defaulting to common",
rarity=rarity_str,
item_id=item_id
)
rarity = ItemRarity.COMMON
# Parse effects if present
effects = []
for effect_data in template.get("effects_on_use", []):
try:
effect = self._parse_effect(effect_data)
if effect:
effects.append(effect)
except Exception as e:
logger.warning(
"Failed to parse effect",
item_id=item_id,
error=str(e)
)
# Parse stat bonuses if present
stat_bonuses = template.get("stat_bonuses", {})
return Item(
item_id=instance_id,
name=template.get("name", item_id),
item_type=item_type,
rarity=rarity,
description=template.get("description", ""),
value=template.get("value", 1),
is_tradeable=template.get("is_tradeable", True),
stat_bonuses=stat_bonuses,
effects_on_use=effects,
)
def _parse_effect(self, effect_data: Dict) -> Optional[Effect]:
"""
Parse an effect from YAML data.
Supports simplified YAML format where effect_type is a string.
Args:
effect_data: Effect definition from YAML
Returns:
Effect instance or None if parsing fails
"""
# Parse effect type
effect_type_str = effect_data.get("effect_type", "buff")
try:
effect_type = EffectType(effect_type_str)
except ValueError:
logger.warning(
"Unknown effect type",
effect_type=effect_type_str
)
return None
# Generate effect ID if not provided
effect_id = effect_data.get(
"effect_id",
f"effect_{uuid.uuid4().hex[:8]}"
)
return Effect(
effect_id=effect_id,
name=effect_data.get("name", "Unknown Effect"),
effect_type=effect_type,
duration=effect_data.get("duration", 1),
power=effect_data.get("power", 0),
stacks=effect_data.get("stacks", 1),
max_stacks=effect_data.get("max_stacks", 5),
)
def get_all_item_ids(self) -> List[str]:
"""
Get list of all available static item IDs.
Returns:
List of item_id strings
"""
self._ensure_loaded()
return list(self._cache.keys())
def has_item(self, item_id: str) -> bool:
"""
Check if an item ID exists.
Args:
item_id: The item ID to check
Returns:
True if item exists in cache
"""
self._ensure_loaded()
return item_id in self._cache
def clear_cache(self) -> None:
"""Clear the item cache, forcing reload on next access."""
self._cache.clear()
self._loaded = False
logger.debug("Static item cache cleared")
# Global instance for convenience
_loader_instance: Optional[StaticItemLoader] = None
def get_static_item_loader() -> StaticItemLoader:
"""
Get the global StaticItemLoader instance.
Returns:
Singleton StaticItemLoader instance
"""
global _loader_instance
if _loader_instance is None:
_loader_instance = StaticItemLoader()
return _loader_instance