392 lines
19 KiB
Python
392 lines
19 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
|
|
GAME_MESSAGES = {
|
|
# Pickup
|
|
'picked_up': {'en': 'Picked up', 'es': 'Has cogido'},
|
|
'inventory_full': {'en': 'Inventory full', 'es': 'Inventario lleno'},
|
|
'dropped_to_ground': {'en': 'Dropped to ground', 'es': 'Tirado al suelo'},
|
|
'item_too_heavy': {
|
|
'en': "⚠️ Item too heavy! {emoji} {name} x{qty} ({weight:.1f}kg) would exceed capacity. Current: {current:.1f}/{max:.1f}kg",
|
|
'es': "⚠️ ¡Objeto muy pesado! {emoji} {name} x{qty} ({weight:.1f}kg) excedería la capacidad. Actual: {current:.1f}/{max:.1f}kg"
|
|
},
|
|
'item_too_large': {
|
|
'en': "⚠️ Item too large! {emoji} {name} x{qty} ({volume:.1f}L) would exceed capacity. Current: {current:.1f}/{max:.1f}L",
|
|
'es': "⚠️ ¡Objeto muy grande! {emoji} {name} x{qty} ({volume:.1f}L) excedería la capacidad. Actual: {current:.1f}/{max:.1f}L"
|
|
},
|
|
'item_not_found_ground': {'en': "Item not found on ground", 'es': "Objeto no encontrado en el suelo"},
|
|
'invalid_quantity': {'en': "Invalid quantity", 'es': "Cantidad inválida"},
|
|
'dropped_item_success': {'en': 'Dropped {emoji} {name} x{qty}', 'es': 'Has tirado {emoji} {name} x{qty}'},
|
|
|
|
# Movement
|
|
'cannot_go_direction': {'en': "You cannot go {direction} from here.", 'es': "No puedes ir al {direction} desde aquí."},
|
|
'exhausted_move': {'en': "You're too exhausted to move. Wait for your stamina to regenerate.", 'es': "Estás demasiado exhausto para moverte. Espera a recuperar stamina."},
|
|
'move_cooldown': {'en': 'You must wait {seconds} seconds before moving again.', 'es': 'Debes esperar {seconds} segundos antes de moverte de nuevo.'},
|
|
'enemy_ambush': {'en': '⚠️ An enemy ambushes you upon arrival!', 'es': '⚠️ ¡Un enemigo te tiende una emboscada al llegar!'},
|
|
'player_left': {'en': '{player_name} left the area', 'es': '{player_name} abandonó el área'},
|
|
'player_arrived': {'en': '{player_name} arrived', 'es': '{player_name} ha llegado'},
|
|
'player_defeated_broadcast': {'en': '{player_name} was defeated in combat', 'es': '{player_name} fue derrotado en combate'},
|
|
'player_defeated_enemy_broadcast': {'en': '{player_name} defeated {npc_name}', 'es': '{player_name} derrotó a {npc_name}'},
|
|
'player_fled_broadcast': {'en': '{player_name} fled from combat', 'es': '{player_name} huyó del combate'},
|
|
'player_entered_combat': {'en': '{player_name} entered combat with {npc_name}', 'es': '{player_name} entró en combate con {npc_name}'},
|
|
'player_returned_pvp': {'en': '{player_name} has returned from PvP combat.', 'es': '{player_name} ha regresado del combate PvP.'},
|
|
|
|
|
|
'pvp_defeat_broadcast': {'en': '{opponent} was defeated by {winner} in PvP combat', 'es': '{opponent} fue derrotado por {winner} en combate PvP'},
|
|
'pvp_initiated_attacker': {'en': "You have initiated combat with {defender}! They get the first turn.", 'es': "¡Has iniciado combate con {defender}! Tiene el primer turno."},
|
|
'pvp_challenged_defender': {'en': "{attacker} has challenged you to PvP combat! It's your turn.", 'es': "¡{attacker} te ha desafiado a combate PvP! Es tu turno."},
|
|
'pvp_combat_started_broadcast': {'en': "⚔️ PvP combat started between {attacker} and {defender}!", 'es': "⚔️ ¡Combate PvP iniciado entre {attacker} y {defender}!"},
|
|
'flee_success_text': {'en': "{name} fled from combat!", 'es': "¡{name} huyó del combate!"},
|
|
'flee_fail_text': {'en': "{name} tried to flee but failed!", 'es': "¡{name} intentó huir pero falló!"},
|
|
|
|
# Loot
|
|
'corpse_name_npc': {'en': "{name} Corpse", 'es': "Cadáver de {name}"},
|
|
'corpse_name_player': {'en': "{name}'s Corpse", 'es': "Cadáver de {name}"},
|
|
'looted_items_start': {'en': "Looted: ", 'es': "Saqueado: "},
|
|
'backpack_full_drop': {'en': "⚠️ Backpack full! Dropped on ground: ", 'es': "⚠️ ¡Mochila llena! Tirado al suelo: "},
|
|
'nothing_looted': {'en': "Nothing could be looted", 'es': "No se pudo saquear nada"},
|
|
'items_require_tools': {'en': "{count} item(s) require tools to extract", 'es': "{count} objeto(s) requieren herramientas"},
|
|
'full_loot_broadcast': {'en': "{player_name} fully looted an NPC corpse", 'es': "{player_name} saqueó completamente un cadáver de NPC"},
|
|
'player_corpse_emptied_broadcast': {'en': "{player_name} fully looted {corpse_name}'s corpse", 'es': "{player_name} vació el cadáver de {corpse_name}"},
|
|
'player_corpse_looted_broadcast': {'en': "{player_name} looted from {corpse_name}'s corpse", 'es': "{player_name} saqueó del cadáver de {corpse_name}"},
|
|
|
|
# Equipment
|
|
'unequip_equip': {'en': "Unequipped {old}, equipped {new}", 'es': "Desequipado {old}, equipado {new}"},
|
|
'equipped': {'en': "Equipped {item}", 'es': "Equipado {item}"},
|
|
'unequipped': {'en': "Unequipped {item}", 'es': "Desequipado {item}"},
|
|
'unequip_dropped': {'en': "Unequipped {item} (dropped to ground - inventory full)", 'es': "Desequipado {item} (tirado al suelo - inventario lleno)"},
|
|
'repaired_success': {'en': "Repaired {item}! Restored {amount} durability.", 'es': "¡Reparado {item}! Restaurados {amount} puntos de durabilidad."},
|
|
|
|
# Characters/Auth
|
|
'character_created': {'en': "Character created successfully", 'es': "Personaje creado con éxito"},
|
|
'character_deleted': {'en': "Character '{name}' deleted successfully", 'es': "Personaje '{name}' eliminado con éxito"},
|
|
'email_updated': {'en': "Email updated successfully", 'es': "Email actualizado con éxito"},
|
|
'password_updated': {'en': "Password updated successfully", 'es': "Contraseña actualizada con éxito"},
|
|
|
|
# Inspection
|
|
'exhausted_inspect': {'en': "You're too exhausted to inspect the area thoroughly. Wait for your stamina to regenerate.", 'es': "Estás demasiado exhausto para inspeccionar. Espera a recuperar stamina."},
|
|
'inspecting_title': {'en': "🔍 **Inspecting {name}**\n", 'es': "🔍 **Inspeccionando {name}**\n"},
|
|
'interactables_title': {'en': "**Interactables:**", 'es': "**Objetos interactuables:**"},
|
|
'npcs_title': {'en': "**NPCs:**", 'es': "**NPCs:**"},
|
|
'items_ground_title': {'en': "**Items on ground:**", 'es': "**Objetos en el suelo:**"},
|
|
|
|
# Interaction
|
|
'not_enough_stamina': {'en': "Not enough stamina. Need {cost}, have {current}.", 'es': "No tienes suficiente stamina. Necesitas {cost}, tienes {current}."},
|
|
'cooldown_wait': {'en': "This action is still on cooldown. Wait {seconds} seconds.", 'es': "Esta acción está en enfriamiento. Espera {seconds} segundos."},
|
|
'object_not_found': {'en': "Object not found", 'es': "Objeto no encontrado"},
|
|
'action_not_found': {'en': "Action not found", 'es': "Acción no encontrada"},
|
|
'action_no_outcomes': {'en': "Action has no defined outcomes", 'es': "La acción no tiene resultados definidos"},
|
|
|
|
# Item Usage
|
|
'item_used': {'en': "Used {name}", 'es': "Usado {name}"},
|
|
'no_item': {'en': "You don't have this item", 'es': "No tienes este objeto"},
|
|
'cannot_use': {'en': "This item cannot be used", 'es': "Este objeto no se puede usar"},
|
|
'cured': {'en': "Cured", 'es': "Curado"},
|
|
|
|
# Status Effects
|
|
'statusDamage': {'en': "You took {damage} damage from status effects", 'es': "Has recibido {damage} de daño por efectos de estado"},
|
|
'statusHeal': {'en': "You recovered {heal} HP from status effects", 'es': "Has recuperado {heal} vida por efectos de estado"},
|
|
'diedFromStatus': {'en': "You died from status effects", 'es': "Has muerto por efectos de estado"},
|
|
}
|
|
|
|
def get_game_message(key: str, lang: str = 'en', **kwargs) -> str:
|
|
"""Get and format a localized game message."""
|
|
messages = GAME_MESSAGES.get(key, {})
|
|
template = messages.get(lang) or messages.get('en') or key
|
|
try:
|
|
return template.format(**kwargs)
|
|
except KeyError:
|
|
return template
|
|
|
|
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, origin: str = "neutral", **data) -> dict:
|
|
"""Create a structured combat message object.
|
|
|
|
Args:
|
|
message_type: Type of combat message (combat_start, player_attack, etc.)
|
|
origin: Origin of the event - "player", "enemy", or "neutral"
|
|
**data: Dynamic data for the message (damage, npc_name, etc.)
|
|
|
|
Returns:
|
|
Dictionary with 'type', 'origin', and 'data' fields
|
|
"""
|
|
return {
|
|
"type": message_type,
|
|
"origin": origin,
|
|
"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
|