""" Helper utilities for game calculations and common operations. Contains distance calculations, stamina costs, capacity calculations, etc. """ import math from typing import Tuple, List, Dict, Any, Union from .. import database as db from ..items import ItemsManager def get_locale_string(value: Union[str, Dict[str, str]], lang: str = 'en') -> str: """Helper to safely get string from i18n object or string.""" if isinstance(value, dict): return value.get(lang) or value.get('en') or str(value) return str(value) def calculate_distance(x1: float, y1: float, x2: float, y2: float) -> float: """ Calculate distance between two points using Euclidean distance. Coordinate system: 1 unit = 100 meters (so distance(0,0 to 1,1) = 141.4m) """ coord_distance = math.sqrt((x2 - x1)**2 + (y2 - y1)**2) distance_meters = coord_distance * 100 return distance_meters def calculate_stamina_cost( distance: float, weight: float, agility: int, max_weight: float = 10.0, volume: float = 0.0, max_volume: float = 10.0 ) -> int: """ Calculate stamina cost based on distance, weight, volume, capacity, and agility. - Base cost: distance / 50 (so 50m = 1 stamina, 100m = 2 stamina) - Weight penalty: +1 stamina per 10kg - Agility reduction: -1 stamina per 3 agility points - Over-capacity penalty: 50-200% extra if over weight OR volume limits - Minimum: 1 stamina """ base_cost = max(1, round(distance / 50)) weight_penalty = int(weight / 10) agility_reduction = int(agility / 3) # Add over-capacity penalty over_capacity_penalty = 0 if weight > max_weight or volume > max_volume: weight_excess_ratio = max(0, (weight - max_weight) / max_weight) if max_weight > 0 else 0 volume_excess_ratio = max(0, (volume - max_volume) / max_volume) if max_volume > 0 else 0 excess_ratio = max(weight_excess_ratio, volume_excess_ratio) over_capacity_penalty = int((base_cost + weight_penalty) * (0.5 + min(1.5, excess_ratio))) total_cost = max(1, base_cost + weight_penalty + over_capacity_penalty - agility_reduction) return total_cost def calculate_crafting_stamina_cost(tier: int, action_type: str = 'craft') -> int: """ Calculate stamina cost for workbench actions. Args: tier: Item tier (1-5) action_type: 'craft', 'repair', or 'uncraft' Returns: Stamina cost """ if action_type == 'craft': # Crafting: max(5, tier * 3) -> T1=5, T5=15 return max(5, tier * 3) elif action_type == 'repair': # Repairing: max(3, tier * 2) -> T1=3, T5=10 return max(3, tier * 2) elif action_type == 'uncraft': # Salvaging: max(2, tier * 1) -> T1=2, T5=5 return max(2, tier * 1) return 1 async def calculate_player_capacity(inventory: List[Dict[str, Any]], items_manager: ItemsManager) -> Tuple[float, float, float, float]: """ Calculate player's current and max weight/volume capacity. Uses unique_stats for equipped items with unique_item_id. Args: inventory: List of inventory items (from db.get_inventory) items_manager: ItemsManager instance Returns: (current_weight, max_weight, current_volume, max_volume) """ current_weight = 0.0 current_volume = 0.0 max_weight = 10.0 # Base capacity max_volume = 10.0 # Base capacity # Collect all unique_item_ids for equipped items equipped_unique_item_ids = [ inv_item['unique_item_id'] for inv_item in inventory if inv_item.get('is_equipped') and inv_item.get('unique_item_id') ] # Batch fetch all unique items in one query unique_items_map = {} if equipped_unique_item_ids: unique_items_map = await db.get_unique_items_batch(equipped_unique_item_ids) for inv_item in inventory: item_def = items_manager.get_item(inv_item['item_id']) if item_def: current_weight += item_def.weight * inv_item['quantity'] current_volume += item_def.volume * inv_item['quantity'] # Check for equipped bags/containers that increase capacity if inv_item['is_equipped']: # Use unique_stats if this is a unique item, otherwise fall back to default stats if inv_item.get('unique_item_id'): unique_item = unique_items_map.get(inv_item['unique_item_id']) if unique_item and unique_item.get('unique_stats'): max_weight += unique_item['unique_stats'].get('weight_capacity', 0) max_volume += unique_item['unique_stats'].get('volume_capacity', 0) elif item_def.stats: # Fallback to default stats if no unique_item_id max_weight += item_def.stats.get('weight_capacity', 0) max_volume += item_def.stats.get('volume_capacity', 0) return current_weight, max_weight, current_volume, max_volume async def reduce_armor_durability(player_id: int, damage_taken: int, items_manager: ItemsManager) -> Tuple[int, List[Dict[str, Any]]]: """ Reduce durability of equipped armor pieces when taking damage. Returns: (armor_damage_absorbed, broken_armor_pieces) """ equipment = await db.get_all_equipment(player_id) armor_pieces = ['head', 'torso', 'legs', 'feet'] total_armor = 0 equipped_armor = [] # Collect all equipped armor for slot in armor_pieces: if equipment.get(slot) and equipment[slot]: armor_slot = equipment[slot] inv_item = await db.get_inventory_item_by_id(armor_slot['item_id']) if inv_item and inv_item.get('unique_item_id'): item_def = items_manager.get_item(inv_item['item_id']) if item_def and item_def.stats and 'armor' in item_def.stats: armor_value = item_def.stats['armor'] total_armor += armor_value equipped_armor.append({ 'slot': slot, 'inv_item_id': armor_slot['item_id'], 'unique_item_id': inv_item['unique_item_id'], 'item_id': inv_item['item_id'], 'item_def': item_def, 'armor_value': armor_value }) if not equipped_armor: return 0, [] # Calculate damage absorbed by armor armor_absorbed = min(damage_taken // 2, total_armor) # Calculate durability loss for each armor piece base_reduction_rate = 0.1 broken_armor = [] for armor in equipped_armor: proportion = armor['armor_value'] / total_armor if total_armor > 0 else 0 durability_loss = max(1, int((damage_taken * proportion / max(armor['armor_value'], 1)) * base_reduction_rate * 10)) # Get current durability unique_item = await db.get_unique_item(armor['unique_item_id']) if unique_item: current_durability = unique_item.get('durability', 0) new_durability = max(0, current_durability - durability_loss) # If armor is about to break, unequip it first if new_durability <= 0: await db.unequip_item(player_id, armor['slot']) # We don't need to manually update inventory is_equipped or remove_from_inventory # because decrease_unique_item_durability will delete the unique item, # which cascades to the inventory row. broken_armor.append({ 'name': get_locale_string(armor['item_def'].name), 'emoji': getattr(armor['item_def'], 'emoji', '🛡️') }) # Decrease durability (handles deletion if <= 0) await db.decrease_unique_item_durability(armor['unique_item_id'], durability_loss) return armor_absorbed, broken_armor async def consume_tool_durability(user_id: int, tools: list, inventory: list, items_manager: ItemsManager) -> Tuple[bool, str, list]: """ Consume durability from required tools. Returns: (success, error_message, consumed_tools_info) """ consumed_tools = [] tools_map = {} # Build map of available tools with durability for inv_item in inventory: item_def = items_manager.get_item(inv_item['item_id']) if item_def and item_def.tool_type and inv_item.get('unique_item_id'): unique_item = await db.get_unique_item(inv_item['unique_item_id']) if unique_item and unique_item.get('durability', 0) > 0: tool_type = item_def.tool_type if tool_type not in tools_map: tools_map[tool_type] = [] tools_map[tool_type].append({ 'inv_item_id': inv_item['id'], 'unique_item_id': inv_item['unique_item_id'], 'item_id': inv_item['item_id'], 'durability': unique_item['durability'], 'name': get_locale_string(item_def.name), 'emoji': getattr(item_def, 'emoji', '🔧') }) # Check and consume tools for tool_req in tools: tool_type = tool_req['type'] durability_cost = tool_req.get('durability_cost', 1) if tool_type not in tools_map or not tools_map[tool_type]: return False, f"Missing required tool: {tool_type}", [] # Use first available tool of this type tool = tools_map[tool_type][0] new_durability = tool['durability'] - durability_cost if new_durability <= 0: # Tool breaks - unequip first await db.unequip_item(user_id, 'weapon') # Assuming tools are equipped as weapons consumed_tools.append(f"{tool['emoji']} {tool['name']} (broke)") tools_map[tool_type].pop(0) else: consumed_tools.append(f"{tool['emoji']} {tool['name']} (-{durability_cost} durability)") # Decrease durability (handles deletion if <= 0) await db.decrease_unique_item_durability(tool['unique_item_id'], durability_cost) return True, "", consumed_tools