292 lines
12 KiB
Python
292 lines
12 KiB
Python
"""
|
|
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)
|
|
|
|
|
|
# Translation maps for backend messages
|
|
DIRECTION_TRANSLATIONS = {
|
|
'north': {'en': 'north', 'es': 'norte'},
|
|
'south': {'en': 'south', 'es': 'sur'},
|
|
'east': {'en': 'east', 'es': 'este'},
|
|
'west': {'en': 'west', 'es': 'oeste'},
|
|
'northeast': {'en': 'northeast', 'es': 'noreste'},
|
|
'northwest': {'en': 'northwest', 'es': 'noroeste'},
|
|
'southeast': {'en': 'southeast', 'es': 'sureste'},
|
|
'southwest': {'en': 'southwest', 'es': 'suroeste'},
|
|
}
|
|
|
|
def translate_travel_message(direction: str, location_name: str, lang: str = 'en') -> str:
|
|
"""Translate a travel message to the user's language."""
|
|
dir_translated = DIRECTION_TRANSLATIONS.get(direction, {}).get(lang, direction)
|
|
|
|
if lang == 'es':
|
|
return f"Viajas al {dir_translated} hacia {location_name}."
|
|
else:
|
|
return f"You travel {dir_translated} to {location_name}."
|
|
|
|
|
|
import json
|
|
|
|
def create_combat_message(message_type: str, **data) -> str:
|
|
"""Create a structured combat message with type and data.
|
|
|
|
Args:
|
|
message_type: Type of combat message (combat_start, player_attack, etc.)
|
|
**data: Dynamic data for the message (damage, npc_name, etc.)
|
|
|
|
Returns:
|
|
Dictionary with 'type' and 'data' fields
|
|
"""
|
|
return json.dumps({
|
|
"type": message_type,
|
|
"data": data
|
|
})
|
|
|
|
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
|