diff --git a/core/nexus_api.py b/core/nexus_api.py index 45f4acc..9a36de5 100644 --- a/core/nexus_api.py +++ b/core/nexus_api.py @@ -1,8 +1,9 @@ """ -Entropia Nexus API Client +Entropia Nexus API Client - Full Implementation -Provides async access to weapon, armor, and tool statistics from entropianexus.com. -Includes mock data for offline testing and development. +This module provides a complete API client for Entropia Nexus. +Currently uses mock data as the public API endpoints are not available. +When the API becomes available, update BASE_URL and remove mock_mode. API Documentation: https://api.entropianexus.com/docs/ """ @@ -10,17 +11,20 @@ API Documentation: https://api.entropianexus.com/docs/ import asyncio import json import logging -from dataclasses import dataclass, field -from decimal import Decimal -from typing import Dict, List, Optional, Any +from dataclasses import dataclass, field, asdict +from decimal import Decimal, InvalidOperation +from typing import Dict, List, Optional, Any, Union from functools import wraps -import time +from datetime import datetime, timedelta +from pathlib import Path +# Optional HTTP libraries try: import aiohttp HAS_AIOHTTP = True except ImportError: HAS_AIOHTTP = False + aiohttp = None try: import urllib.request @@ -44,26 +48,36 @@ class WeaponStats: damage: Decimal decay_pec: Decimal ammo_pec: Decimal - dpp: Decimal # Damage Per PEC (calculated) + dpp: Decimal range: int = 0 attacks_per_min: int = 0 - total_cost_pec: Decimal = field(default=None) - markup_percent: Decimal = Decimal("100.0") + total_cost_pec: Optional[Decimal] = None + markup_percent: Decimal = field(default_factory=lambda: Decimal("100.0")) item_id: str = "" + item_class: str = "" # e.g., "Laser rifle", "BLP pistol" + weight: Decimal = field(default_factory=lambda: Decimal("0.0")) + power_cost: Decimal = field(default_factory=lambda: Decimal("0.0")) def __post_init__(self): if self.total_cost_pec is None: self.total_cost_pec = self.decay_pec + self.ammo_pec - # Ensure dpp is calculated if not provided - if self.dpp == Decimal("0") and self.total_cost_pec > 0: - self.dpp = self.damage / (self.total_cost_pec / 100) # Convert PEC to PED + # Validate DPP calculation + if self.dpp == Decimal("0") and self.total_cost_pec and self.total_cost_pec > 0: + self.dpp = self.damage / self.total_cost_pec def calculate_cost_per_hour(self) -> Decimal: """Calculate total cost per hour of use in PED.""" - if self.attacks_per_min <= 0: + if not self.attacks_per_min or self.attacks_per_min <= 0: return Decimal("0") shots_per_hour = self.attacks_per_min * 60 - return (self.total_cost_pec * shots_per_hour) / 100 # Convert PEC to PED + total_pec = self.total_cost_pec * shots_per_hour if self.total_cost_pec else Decimal("0") + return total_pec / 100 # Convert PEC to PED + + def calculate_dpp(self) -> Decimal: + """Calculate Damage Per PEC.""" + if not self.total_cost_pec or self.total_cost_pec == 0: + return Decimal("0") + return self.damage / self.total_cost_pec def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" @@ -75,116 +89,554 @@ class WeaponStats: 'dpp': str(self.dpp), 'range': self.range, 'attacks_per_min': self.attacks_per_min, - 'total_cost_pec': str(self.total_cost_pec), + 'total_cost_pec': str(self.total_cost_pec) if self.total_cost_pec else "0", 'markup_percent': str(self.markup_percent), 'item_id': self.item_id, + 'item_class': self.item_class, + 'weight': str(self.weight), + 'power_cost': str(self.power_cost), } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'WeaponStats': """Create from dictionary.""" return cls( - name=data['name'], + name=data.get('name', ''), damage=Decimal(data.get('damage', 0)), decay_pec=Decimal(data.get('decay_pec', 0)), ammo_pec=Decimal(data.get('ammo_pec', 0)), dpp=Decimal(data.get('dpp', 0)), - range=data.get('range', 0), - attacks_per_min=data.get('attacks_per_min', 0), - total_cost_pec=Decimal(data.get('total_cost_pec', data.get('decay_pec', 0))) if 'total_cost_pec' in data else None, + range=int(data.get('range', 0)), + attacks_per_min=int(data.get('attacks_per_min', 0)), markup_percent=Decimal(data.get('markup_percent', 100)), item_id=data.get('item_id', ''), + item_class=data.get('item_class', ''), + weight=Decimal(data.get('weight', 0)), + power_cost=Decimal(data.get('power_cost', 0)), ) +@dataclass +class ArmorPiece: + """Single armor piece stats.""" + slot: str # head, body, arms, hands, thighs, shins, feet + protection: Dict[str, Decimal] = field(default_factory=dict) + decay_pec: Decimal = field(default_factory=lambda: Decimal("0.0")) + weight: Decimal = field(default_factory=lambda: Decimal("0.0")) + + @dataclass class ArmorStats: - """Armor piece statistics from Entropia Nexus.""" + """Armor set statistics.""" name: str decay_pec: Decimal - protection: Dict[str, Decimal] - slot: str = "body" # head, body, arms, legs, feet - durability: int = 10000 + protection: Dict[str, Decimal] = field(default_factory=dict) + pieces: Dict[str, ArmorPiece] = field(default_factory=dict) + durability: int = 0 + slot: str = "body" item_id: str = "" + markup_percent: Decimal = field(default_factory=lambda: Decimal("100.0")) def get_total_protection(self) -> Decimal: - """Get sum of all protection values.""" + """Calculate total protection across all damage types.""" return sum(self.protection.values(), Decimal("0")) def get_protection(self, damage_type: str) -> Decimal: """Get protection value for specific damage type.""" - return self.protection.get(damage_type, Decimal("0")) + return self.protection.get(damage_type.lower(), Decimal("0")) + + def get_durability_hours(self) -> Decimal: + """Estimate durability in hours of use.""" + if self.durability <= 0: + return Decimal("0") + # Rough estimate: 1000 durability ≈ 10 hours + return Decimal(str(self.durability)) / Decimal("100") def to_dict(self) -> Dict[str, Any]: - """Convert to dictionary for serialization.""" + """Convert to dictionary.""" return { 'name': self.name, 'decay_pec': str(self.decay_pec), 'protection': {k: str(v) for k, v in self.protection.items()}, - 'slot': self.slot, 'durability': self.durability, + 'slot': self.slot, 'item_id': self.item_id, + 'markup_percent': str(self.markup_percent), } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ArmorStats': """Create from dictionary.""" return cls( - name=data['name'], + name=data.get('name', ''), decay_pec=Decimal(data.get('decay_pec', 0)), protection={k: Decimal(v) for k, v in data.get('protection', {}).items()}, + durability=int(data.get('durability', 0)), slot=data.get('slot', 'body'), - durability=data.get('durability', 10000), item_id=data.get('item_id', ''), + markup_percent=Decimal(data.get('markup_percent', 100)), ) -@dataclass +@dataclass class ToolStats: - """Mining tool statistics from Entropia Nexus.""" + """Mining/scanning tool statistics.""" name: str depth: Decimal radius: Decimal decay_pec: Decimal - tool_type: str = "finder" # finder or extractor - probe_cost: Decimal = Decimal("0.5") # PED per drop + tool_type: str = "finder" # finder, extractor, scanner + probe_cost: Optional[Decimal] = None item_id: str = "" + markup_percent: Decimal = field(default_factory=lambda: Decimal("100.0")) def calculate_cost_per_drop(self) -> Decimal: - """Calculate total cost per mining drop in PED.""" - return (self.decay_pec / 100) + self.probe_cost # PEC to PED + probe + """Calculate cost per mining drop in PED.""" + total_pec = self.decay_pec + if self.probe_cost: + total_pec += self.probe_cost + return total_pec / 100 # Convert to PED def to_dict(self) -> Dict[str, Any]: - """Convert to dictionary for serialization.""" + """Convert to dictionary.""" return { 'name': self.name, 'depth': str(self.depth), 'radius': str(self.radius), 'decay_pec': str(self.decay_pec), 'tool_type': self.tool_type, - 'probe_cost': str(self.probe_cost), + 'probe_cost': str(self.probe_cost) if self.probe_cost else "0.5", 'item_id': self.item_id, + 'markup_percent': str(self.markup_percent), } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ToolStats': """Create from dictionary.""" return cls( - name=data['name'], + name=data.get('name', ''), depth=Decimal(data.get('depth', 0)), radius=Decimal(data.get('radius', 0)), decay_pec=Decimal(data.get('decay_pec', 0)), tool_type=data.get('tool_type', 'finder'), probe_cost=Decimal(data.get('probe_cost', 0.5)), item_id=data.get('item_id', ''), + markup_percent=Decimal(data.get('markup_percent', 100)), ) +# ============================================================================= +# Gear Loadout +# ============================================================================= + +@dataclass +class GearLoadout: + """Complete hunting/mining/crafting loadout.""" + name: str + weapon: Optional[WeaponStats] = None + armor: Optional[ArmorStats] = None + tool: Optional[ToolStats] = None + amplifier: Optional[Dict[str, Any]] = None + scope: Optional[Dict[str, Any]] = None + + def get_total_cost_per_hour(self) -> Decimal: + """Calculate total cost per hour for this loadout.""" + total = Decimal("0") + if self.weapon: + total += self.weapon.calculate_cost_per_hour() + # Armor decay is much slower, estimate based on hits taken + if self.armor: + # Rough estimate: 100 hits per hour + armor_cost_ped = self.armor.decay_pec / 100 + total += armor_cost_ped * Decimal("100") # 100 hits + return total + + def get_total_dpp(self) -> Optional[Decimal]: + """Get DPP of primary weapon.""" + if self.weapon: + return self.weapon.dpp + return None + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary.""" + return { + 'name': self.name, + 'weapon': self.weapon.to_dict() if self.weapon else None, + 'armor': self.armor.to_dict() if self.armor else None, + 'tool': self.tool.to_dict() if self.tool else None, + 'total_cost_per_hour': str(self.get_total_cost_per_hour()), + 'dpp': str(self.get_total_dpp()) if self.get_total_dpp() else None, + } + + +# ============================================================================= +# Cache Implementation +# ============================================================================= + +class SimpleCache: + """Simple in-memory cache with TTL.""" + + def __init__(self, default_ttl: int = 3600): + self._cache: Dict[str, Any] = {} + self._timestamps: Dict[str, float] = {} + self._default_ttl = default_ttl + + def get(self, key: str) -> Optional[Any]: + """Get cached value if not expired.""" + if key not in self._cache: + return None + + timestamp = self._timestamps.get(key, 0) + if time.time() - timestamp > self._default_ttl: + self.delete(key) + return None + + return self._cache[key] + + def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None: + """Set cache value.""" + self._cache[key] = value + self._timestamps[key] = time.time() + + def delete(self, key: str) -> None: + """Delete cache entry.""" + self._cache.pop(key, None) + self._timestamps.pop(key, None) + + def clear(self) -> None: + """Clear all cache.""" + self._cache.clear() + self._timestamps.clear() + + +# ============================================================================= +# API Client +# ============================================================================= + +class EntropiaNexusAPI: + """ + Client for Entropia Nexus API. + + Currently operates in mock mode as public API endpoints are not available. + When API becomes available, set mock_mode=False and provide base_url. + """ + + # TODO: Update with real API endpoints when available + BASE_URL = "https://api.entropianexus.com" + API_VERSION = "v1" + + def __init__(self, api_key: Optional[str] = None, + mock_mode: bool = True, + cache_ttl: int = 3600, + base_url: Optional[str] = None): + """ + Initialize API client. + + Args: + api_key: API key for authentication (if required) + mock_mode: Use mock data instead of API calls + cache_ttl: Cache time-to-live in seconds + base_url: Override base URL for API + """ + self.api_key = api_key + self.mock_mode = mock_mode + self._cache = SimpleCache(cache_ttl) + self._session: Optional[Any] = None + self._base_url = base_url or self.BASE_URL + + logger.info(f"EntropiaNexusAPI initialized (mock_mode={mock_mode})") + + async def __aenter__(self): + """Async context manager entry.""" + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + """Async context manager exit.""" + await self.close() + return False + + async def close(self): + """Close API client and cleanup.""" + if self._session and HAS_AIOHTTP: + await self._session.close() + self._session = None + + async def _get_session(self) -> Optional[Any]: + """Get or create HTTP session.""" + if not HAS_AIOHTTP or self.mock_mode: + return None + + if self._session is None or self._session.closed: + self._session = aiohttp.ClientSession( + headers={ + 'Accept': 'application/json', + 'User-Agent': 'Lemontropia-Suite/0.2.0' + }, + timeout=aiohttp.ClientTimeout(total=15) + ) + return self._session + + async def _make_request(self, endpoint: str, + params: Optional[Dict] = None) -> Optional[Dict]: + """ + Make API request. + + Args: + endpoint: API endpoint path + params: Query parameters + + Returns: + JSON response or None on error + """ + if self.mock_mode: + return None + + url = f"{self._base_url}/{self.API_VERSION}/{endpoint}" + + try: + session = await self._get_session() + if not session: + return None + + headers = {} + if self.api_key: + headers['Authorization'] = f'Bearer {self.api_key}' + + async with session.get(url, params=params, headers=headers) as resp: + if resp.status == 200: + return await resp.json() + else: + logger.warning(f"API error: {resp.status}") + return None + + except Exception as e: + logger.error(f"API request failed: {e}") + return None + + # ======================================================================== + # Search Methods + # ======================================================================== + + async def search_items(self, query: str, + item_type: Optional[str] = None) -> List[Dict]: + """ + Search for items. + + Args: + query: Search term + item_type: Filter by type ('weapon', 'armor', 'tool') + + Returns: + List of item summaries + """ + cache_key = f"search:{query}:{item_type}" + cached = self._cache.get(cache_key) + if cached: + return cached + + # Try API + params = {'q': query} + if item_type: + params['type'] = item_type + + data = await self._make_request('items/search', params) + if data: + self._cache.set(cache_key, data) + return data + + # Fallback to mock + if self.mock_mode: + return self._mock_search(query, item_type) + + return [] + + def _mock_search(self, query: str, + item_type: Optional[str]) -> List[Dict]: + """Mock search implementation.""" + results = [] + query_lower = query.lower() + + if not item_type or item_type == 'weapon': + for item_id, weapon in MOCK_WEAPONS.items(): + if query_lower in weapon.name.lower(): + results.append({ + 'id': item_id, + 'name': weapon.name, + 'type': 'weapon', + 'category': weapon.item_class or 'Weapon' + }) + + if not item_type or item_type == 'armor': + for item_id, armor in MOCK_ARMORS.items(): + if query_lower in armor.name.lower(): + results.append({ + 'id': item_id, + 'name': armor.name, + 'type': 'armor', + 'category': 'Armor' + }) + + if not item_type or item_type == 'tool': + for item_id, tool in MOCK_TOOLS.items(): + if query_lower in tool.name.lower(): + results.append({ + 'id': item_id, + 'name': tool.name, + 'type': 'tool', + 'category': tool.tool_type.title() + }) + + return results + + # ======================================================================== + # Weapon Methods + # ======================================================================== + + async def get_weapon(self, item_id: str) -> Optional[WeaponStats]: + """Get weapon details by ID.""" + cache_key = f"weapon:{item_id}" + cached = self._cache.get(cache_key) + if cached: + return WeaponStats.from_dict(cached) + + # Try API + data = await self._make_request(f'items/{item_id}') + if data and data.get('type') == 'weapon': + weapon = self._parse_weapon_data(data) + self._cache.set(cache_key, weapon.to_dict()) + return weapon + + # Fallback to mock + if self.mock_mode and item_id in MOCK_WEAPONS: + return MOCK_WEAPONS[item_id] + + return None + + def _parse_weapon_data(self, data: Dict) -> WeaponStats: + """Parse API weapon data.""" + stats = data.get('stats', {}) + return WeaponStats( + name=data.get('name', 'Unknown'), + damage=Decimal(str(stats.get('damage', 0))), + decay_pec=Decimal(str(stats.get('decay', 0))), + ammo_pec=Decimal(str(stats.get('ammo', 0))), + dpp=Decimal(str(stats.get('dpp', 0))), + range=stats.get('range', 0), + attacks_per_min=stats.get('attacks_per_min', 0), + item_id=str(data.get('id', '')), + item_class=stats.get('class', ''), + ) + + async def get_all_weapons(self) -> List[WeaponStats]: + """Get all weapons.""" + return list(MOCK_WEAPONS.values()) + + # ======================================================================== + # Armor Methods + # ======================================================================== + + async def get_armor(self, item_id: str) -> Optional[ArmorStats]: + """Get armor details by ID.""" + cache_key = f"armor:{item_id}" + cached = self._cache.get(cache_key) + if cached: + return ArmorStats.from_dict(cached) + + data = await self._make_request(f'items/{item_id}') + if data and data.get('type') == 'armor': + armor = self._parse_armor_data(data) + self._cache.set(cache_key, armor.to_dict()) + return armor + + if self.mock_mode and item_id in MOCK_ARMORS: + return MOCK_ARMORS[item_id] + + return None + + def _parse_armor_data(self, data: Dict) -> ArmorStats: + """Parse API armor data.""" + stats = data.get('stats', {}) + protection = {} + for dmg_type, value in stats.get('protection', {}).items(): + protection[dmg_type] = Decimal(str(value)) + + return ArmorStats( + name=data.get('name', 'Unknown'), + decay_pec=Decimal(str(stats.get('decay', 0))), + protection=protection, + durability=stats.get('durability', 0), + item_id=str(data.get('id', '')), + ) + + async def get_all_armors(self) -> List[ArmorStats]: + """Get all armors.""" + return list(MOCK_ARMORS.values()) + + # ======================================================================== + # Tool Methods + # ======================================================================== + + async def get_tool(self, item_id: str) -> Optional[ToolStats]: + """Get tool details by ID.""" + cache_key = f"tool:{item_id}" + cached = self._cache.get(cache_key) + if cached: + return ToolStats.from_dict(cached) + + data = await self._make_request(f'items/{item_id}') + if data and data.get('type') == 'tool': + tool = self._parse_tool_data(data) + self._cache.set(cache_key, tool.to_dict()) + return tool + + if self.mock_mode and item_id in MOCK_TOOLS: + return MOCK_TOOLS[item_id] + + return None + + def _parse_tool_data(self, data: Dict) -> ToolStats: + """Parse API tool data.""" + stats = data.get('stats', {}) + return ToolStats( + name=data.get('name', 'Unknown'), + depth=Decimal(str(stats.get('depth', 0))), + radius=Decimal(str(stats.get('radius', 0))), + decay_pec=Decimal(str(stats.get('decay', 0))), + tool_type=stats.get('tool_type', 'finder'), + item_id=str(data.get('id', '')), + ) + + async def get_all_tools(self) -> List[ToolStats]: + """Get all tools.""" + return list(MOCK_TOOLS.values()) + + # ======================================================================== + # Loadout Methods + # ======================================================================== + + async def calculate_loadout(self, weapon_id: Optional[str] = None, + armor_id: Optional[str] = None, + tool_id: Optional[str] = None) -> Optional[GearLoadout]: + """Calculate stats for a complete loadout.""" + loadout = GearLoadout(name="Custom Loadout") + + if weapon_id: + loadout.weapon = await self.get_weapon(weapon_id) + + if armor_id: + loadout.armor = await self.get_armor(armor_id) + + if tool_id: + loadout.tool = await self.get_tool(tool_id) + + return loadout + + # ============================================================================= # Mock Data # ============================================================================= -MOCK_WEAPONS = { +MOCK_WEAPONS: Dict[str, WeaponStats] = { "sollomate_opalo": WeaponStats( name="Sollomate Opalo", damage=Decimal("4.0"), @@ -193,7 +645,8 @@ MOCK_WEAPONS = { dpp=Decimal("3.33"), range=26, attacks_per_min=56, - item_id="sollomate_opalo" + item_id="sollomate_opalo", + item_class="Laser Rifle" ), "omegaton_m2100": WeaponStats( name="Omegaton M2100", @@ -203,7 +656,8 @@ MOCK_WEAPONS = { dpp=Decimal("3.33"), range=28, attacks_per_min=54, - item_id="omegaton_m2100" + item_id="omegaton_m2100", + item_class="Laser Carbine" ), "breer_m1a": WeaponStats( name="Breer M1a", @@ -213,7 +667,8 @@ MOCK_WEAPONS = { dpp=Decimal("3.33"), range=30, attacks_per_min=52, - item_id="breer_m1a" + item_id="breer_m1a", + item_class="BLP Rifle" ), "castorian_enforcer_se": WeaponStats( name="Castorian Enforcer SE", @@ -223,7 +678,8 @@ MOCK_WEAPONS = { dpp=Decimal("1.87"), range=26, attacks_per_min=44, - item_id="castorian_enforcer_se" + item_id="castorian_enforcer_se", + item_class="BLP Pistol" ), "isis_lr1": WeaponStats( name="ISIS LR1", @@ -233,7 +689,8 @@ MOCK_WEAPONS = { dpp=Decimal("2.50"), range=60, attacks_per_min=38, - item_id="isis_lr1" + item_id="isis_lr1", + item_class="Laser Sniper" ), "sollomate_ony": WeaponStats( name="Sollomate Ony", @@ -243,11 +700,12 @@ MOCK_WEAPONS = { dpp=Decimal("2.00"), range=24, attacks_per_min=48, - item_id="sollomate_ony" + item_id="sollomate_ony", + item_class="Laser Pistol" ), } -MOCK_ARMORS = { +MOCK_ARMORS: Dict[str, ArmorStats] = { "pixie": ArmorStats( name="Pixie", decay_pec=Decimal("0.135"), @@ -259,12 +717,27 @@ MOCK_ARMORS = { "cold": Decimal("1.0"), }, slot="body", - durability=2800, + durability=2400, item_id="pixie" ), + "goblin": ArmorStats( + name="Goblin", + decay_pec=Decimal("0.20"), + protection={ + "impact": Decimal("4.0"), + "cut": Decimal("4.0"), + "stab": Decimal("4.0"), + "burn": Decimal("3.0"), + "cold": Decimal("2.0"), + "acid": Decimal("1.0"), + }, + slot="body", + durability=2800, + item_id="goblin" + ), "shogun": ArmorStats( name="Shogun", - decay_pec=Decimal("0.60"), + decay_pec=Decimal("0.55"), protection={ "impact": Decimal("10.0"), "cut": Decimal("10.0"), @@ -306,24 +779,9 @@ MOCK_ARMORS = { durability=3600, item_id="vigilante" ), - "necconu": ArmorStats( - name="Necconu", - decay_pec=Decimal("0.38"), - protection={ - "impact": Decimal("7.0"), - "cut": Decimal("7.0"), - "stab": Decimal("7.0"), - "burn": Decimal("4.0"), - "cold": Decimal("4.0"), - "electric": Decimal("3.0"), - }, - slot="body", - durability=3200, - item_id="necconu" - ), } -MOCK_TOOLS = { +MOCK_TOOLS: Dict[str, ToolStats] = { "ziplex_z1": ToolStats( name="Ziplex Z1 Seeker", depth=Decimal("219.5"), @@ -362,428 +820,18 @@ MOCK_TOOLS = { } -# ============================================================================= -# Cache Implementation -# ============================================================================= - -class SimpleCache: - """Simple in-memory cache with TTL support.""" - - def __init__(self, default_ttl: int = 3600): - self._cache: Dict[str, Any] = {} - self._timestamps: Dict[str, float] = {} - self._default_ttl = default_ttl - - def get(self, key: str) -> Optional[Any]: - """Get cached value if not expired.""" - if key not in self._cache: - return None - - timestamp = self._timestamps.get(key, 0) - if time.time() - timestamp > self._default_ttl: - self.delete(key) - return None - - return self._cache[key] - - def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None: - """Cache a value with optional TTL.""" - self._cache[key] = value - self._timestamps[key] = time.time() - - def delete(self, key: str) -> None: - """Delete a cached value.""" - self._cache.pop(key, None) - self._timestamps.pop(key, None) - - def clear(self) -> None: - """Clear all cached values.""" - self._cache.clear() - self._timestamps.clear() - - def keys(self) -> List[str]: - """Get all cache keys.""" - return list(self._cache.keys()) - - -# ============================================================================= -# API Client -# ============================================================================= - -class EntropiaNexusAPI: - """ - Async client for Entropia Nexus API. - - Provides access to weapon, armor, and tool statistics with: - - Automatic caching of results - - Mock data fallback for offline testing - - Graceful error handling - """ - - BASE_URL = "https://api.entropianexus.com/v1" - - def __init__(self, api_key: Optional[str] = None, use_mock_fallback: bool = True, - cache_ttl: int = 3600, mock_mode: bool = False): - """ - Initialize API client. - - Args: - api_key: Optional API key for higher rate limits - use_mock_fallback: Whether to fall back to mock data on API failure - cache_ttl: Cache time-to-live in seconds (default: 1 hour) - mock_mode: If True, always use mock data (no API calls) - """ - self.api_key = api_key - self.use_mock_fallback = use_mock_fallback - self.mock_mode = mock_mode - self._cache = SimpleCache(default_ttl=cache_ttl) - self._session: Optional[Any] = None - - logger.info(f"EntropiaNexusAPI initialized (mock_mode={mock_mode})") - - async def _get_session(self) -> Optional[Any]: - """Get or create aiohttp session.""" - if not HAS_AIOHTTP: - return None - - if self._session is None or self._session.closed: - self._session = aiohttp.ClientSession( - headers={ - 'Accept': 'application/json', - 'User-Agent': 'Lemontropia-Suite/0.1.0' - }, - timeout=aiohttp.ClientTimeout(total=10) - ) - return self._session - - async def _make_request(self, endpoint: str, params: Optional[Dict] = None) -> Optional[Dict]: - """ - Make async API request to Entropia Nexus. - - Args: - endpoint: API endpoint path (without base URL) - params: Query parameters - - Returns: - JSON response as dict, or None on error - """ - if self.mock_mode: - return None - - url = f"{self.BASE_URL}/{endpoint}" - - try: - session = await self._get_session() - if session is None: - logger.warning("aiohttp not available, falling back to sync request") - return self._make_sync_request(endpoint, params) - - async with session.get(url, params=params) as response: - if response.status == 200: - return await response.json() - else: - logger.warning(f"API returned status {response.status}") - return None - - except Exception as e: - logger.error(f"API request failed: {e}") - return None - - def _make_sync_request(self, endpoint: str, params: Optional[Dict] = None) -> Optional[Dict]: - """Synchronous fallback request using urllib.""" - if not HAS_URLLIB or self.mock_mode: - return None - - url = f"{self.BASE_URL}/{endpoint}" - - if params: - query_string = '&'.join(f"{k}={urllib.parse.quote(str(v))}" for k, v in params.items()) - url = f"{url}?{query_string}" - - headers = { - 'Accept': 'application/json', - 'User-Agent': 'Lemontropia-Suite/0.1.0' - } - - if self.api_key: - headers['Authorization'] = f'Bearer {self.api_key}' - - try: - req = urllib.request.Request(url, headers=headers) - with urllib.request.urlopen(req, timeout=10) as response: - return json.loads(response.read().decode('utf-8')) - except Exception as e: - logger.error(f"Sync API request failed: {e}") - return None - - async def search_items(self, query: str, item_type: Optional[str] = None) -> List[Dict]: - """ - Search for items by name. - - Args: - query: Item name search term - item_type: Optional filter ('weapon', 'armor', 'tool') - - Returns: - List of item data dicts with 'id', 'name', 'type' keys - """ - cache_key = f"search:{query}:{item_type}" - cached = self._cache.get(cache_key) - if cached: - return cached - - # Try API first - params = {'q': query} - if item_type: - params['type'] = item_type - - data = await self._make_request('items/search', params) - - if data and 'results' in data: - results = data['results'] - self._cache.set(cache_key, results) - return results - - # Fallback to mock data - if self.use_mock_fallback: - return self._search_mock_items(query, item_type) - - return [] - - def _search_mock_items(self, query: str, item_type: Optional[str] = None) -> List[Dict]: - """Search mock items by query string.""" - query_lower = query.lower() - results = [] - - if not item_type or item_type == 'weapon': - for item_id, weapon in MOCK_WEAPONS.items(): - if query_lower in weapon.name.lower(): - results.append({ - 'id': item_id, - 'name': weapon.name, - 'type': 'weapon' - }) - - if not item_type or item_type == 'armor': - for item_id, armor in MOCK_ARMORS.items(): - if query_lower in armor.name.lower(): - results.append({ - 'id': item_id, - 'name': armor.name, - 'type': 'armor' - }) - - if not item_type or item_type == 'tool': - for item_id, tool in MOCK_TOOLS.items(): - if query_lower in tool.name.lower(): - results.append({ - 'id': item_id, - 'name': tool.name, - 'type': 'tool' - }) - - return results - - async def get_weapon_stats(self, item_id: str) -> Optional[WeaponStats]: - """ - Get detailed weapon statistics. - - Args: - item_id: Weapon item ID or name - - Returns: - WeaponStats object or None if not found - """ - cache_key = f"weapon:{item_id}" - cached = self._cache.get(cache_key) - if cached: - return cached - - # Try API first - data = await self._make_request(f'weapons/{item_id}') - - if data: - try: - weapon = WeaponStats( - name=data['name'], - damage=Decimal(str(data.get('damage', 0))), - decay_pec=Decimal(str(data.get('decay_pec', 0))), - ammo_pec=Decimal(str(data.get('ammo_pec', 0))), - dpp=Decimal(str(data.get('dpp', 0))), - range=data.get('range', 0), - attacks_per_min=data.get('attacks_per_min', 0), - markup_percent=Decimal(str(data.get('markup', 100))), - item_id=item_id - ) - self._cache.set(cache_key, weapon) - return weapon - except (KeyError, ValueError) as e: - logger.error(f"Failed to parse weapon data: {e}") - - # Fallback to mock data - if self.use_mock_fallback: - normalized_id = item_id.lower().replace(' ', '_') - if normalized_id in MOCK_WEAPONS: - weapon = MOCK_WEAPONS[normalized_id] - self._cache.set(cache_key, weapon) - return weapon - - # Try partial match - for mock_id, weapon in MOCK_WEAPONS.items(): - if normalized_id in mock_id or normalized_id in weapon.name.lower(): - self._cache.set(cache_key, weapon) - return weapon - - return None - - async def get_armor_stats(self, item_id: str) -> Optional[ArmorStats]: - """ - Get detailed armor statistics. - - Args: - item_id: Armor item ID or name - - Returns: - ArmorStats object or None if not found - """ - cache_key = f"armor:{item_id}" - cached = self._cache.get(cache_key) - if cached: - return cached - - # Try API first - data = await self._make_request(f'armor/{item_id}') - - if data: - try: - protection = data.get('protection', {}) - armor = ArmorStats( - name=data['name'], - decay_pec=Decimal(str(data.get('decay_pec', 0))), - protection={k: Decimal(str(v)) for k, v in protection.items()}, - slot=data.get('slot', 'body'), - durability=data.get('durability', 10000), - item_id=item_id - ) - self._cache.set(cache_key, armor) - return armor - except (KeyError, ValueError) as e: - logger.error(f"Failed to parse armor data: {e}") - - # Fallback to mock data - if self.use_mock_fallback: - normalized_id = item_id.lower().replace(' ', '_') - if normalized_id in MOCK_ARMORS: - armor = MOCK_ARMORS[normalized_id] - self._cache.set(cache_key, armor) - return armor - - # Try partial match - for mock_id, armor in MOCK_ARMORS.items(): - if normalized_id in mock_id or normalized_id in armor.name.lower(): - self._cache.set(cache_key, armor) - return armor - - return None - - async def get_tool_stats(self, item_id: str) -> Optional[ToolStats]: - """ - Get detailed tool statistics. - - Args: - item_id: Tool item ID or name - - Returns: - ToolStats object or None if not found - """ - cache_key = f"tool:{item_id}" - cached = self._cache.get(cache_key) - if cached: - return cached - - # Try API first - data = await self._make_request(f'tools/{item_id}') - - if data: - try: - tool = ToolStats( - name=data['name'], - depth=Decimal(str(data.get('depth', 0))), - radius=Decimal(str(data.get('radius', 0))), - decay_pec=Decimal(str(data.get('decay_pec', 0))), - tool_type=data.get('type', 'finder'), - probe_cost=Decimal(str(data.get('probe_cost', 0.5))), - item_id=item_id - ) - self._cache.set(cache_key, tool) - return tool - except (KeyError, ValueError) as e: - logger.error(f"Failed to parse tool data: {e}") - - # Fallback to mock data - if self.use_mock_fallback: - normalized_id = item_id.lower().replace(' ', '_') - if normalized_id in MOCK_TOOLS: - tool = MOCK_TOOLS[normalized_id] - self._cache.set(cache_key, tool) - return tool - - # Try partial match - for mock_id, tool in MOCK_TOOLS.items(): - if normalized_id in mock_id or normalized_id in tool.name.lower(): - self._cache.set(cache_key, tool) - return tool - - return None - - async def get_market_markup(self, item_id: str) -> Optional[Decimal]: - """ - Get current market markup percentage for an item. - - Args: - item_id: Item ID or name - - Returns: - Markup percentage (e.g., 105.5 for 105.5%) or None - """ - cache_key = f"markup:{item_id}" - cached = self._cache.get(cache_key) - if cached: - return cached - - data = await self._make_request(f'markup/{item_id}') - - if data and 'markup' in data: - markup = Decimal(str(data['markup'])) - self._cache.set(cache_key, markup, ttl=300) # 5 min cache for market data - return markup - - return None - - def clear_cache(self) -> None: - """Clear all cached data.""" - self._cache.clear() - logger.info("Cache cleared") - - async def close(self) -> None: - """Close the API client and cleanup resources.""" - if self._session and not self._session.closed: - await self._session.close() - self._session = None - - async def __aenter__(self): - """Async context manager entry.""" - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - """Async context manager exit.""" - await self.close() - - # ============================================================================= # Utility Functions # ============================================================================= +def calculate_dpp(damage: Decimal, decay_pec: Decimal, ammo_pec: Decimal) -> Decimal: + """Calculate Damage Per PEC.""" + total_cost = decay_pec + ammo_pec + if total_cost == 0: + return Decimal("0") + return damage / total_cost + + def get_mock_weapons() -> Dict[str, WeaponStats]: """Get all mock weapons.""" return MOCK_WEAPONS.copy() @@ -799,25 +847,13 @@ def get_mock_tools() -> Dict[str, ToolStats]: return MOCK_TOOLS.copy() -def calculate_dpp(damage: Decimal, decay_pec: Decimal, ammo_pec: Decimal) -> Decimal: - """ - Calculate Damage Per PEC (DPP). - - DPP is a key efficiency metric in Entropia Universe. - Higher DPP means more damage per PEC spent. - - Args: - damage: Weapon damage - decay_pec: Decay per shot in PEC - ammo_pec: Ammo cost per shot in PEC - - Returns: - DPP value (damage per PEC spent) - """ - total_cost_pec = decay_pec + ammo_pec - if total_cost_pec == 0: - return Decimal("0") - return damage / total_cost_pec +def get_weapon_by_name(name: str) -> Optional[WeaponStats]: + """Find weapon by name (case insensitive).""" + name_lower = name.lower() + for weapon in MOCK_WEAPONS.values(): + if name_lower in weapon.name.lower(): + return weapon + return None # ============================================================================= @@ -826,15 +862,18 @@ def calculate_dpp(damage: Decimal, decay_pec: Decimal, ammo_pec: Decimal) -> Dec __all__ = [ 'WeaponStats', - 'ArmorStats', + 'ArmorStats', + 'ArmorPiece', 'ToolStats', + 'GearLoadout', 'EntropiaNexusAPI', 'SimpleCache', 'MOCK_WEAPONS', 'MOCK_ARMORS', 'MOCK_TOOLS', + 'calculate_dpp', 'get_mock_weapons', 'get_mock_armors', 'get_mock_tools', - 'calculate_dpp', -] + 'get_weapon_by_name', +] \ No newline at end of file