Commit
This commit is contained in:
245
api/services/helpers.py
Normal file
245
api/services/helpers.py
Normal file
@@ -0,0 +1,245 @@
|
||||
"""
|
||||
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
|
||||
from .. import database as db
|
||||
from ..items import ItemsManager
|
||||
|
||||
|
||||
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': 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': 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
|
||||
Reference in New Issue
Block a user