""" Entropia Nexus API Client Provides async access to weapon, armor, and tool statistics from entropianexus.com. Includes mock data for offline testing and development. 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 functools import wraps import time try: import aiohttp HAS_AIOHTTP = True except ImportError: HAS_AIOHTTP = False try: import urllib.request import urllib.error import urllib.parse HAS_URLLIB = True except ImportError: HAS_URLLIB = False logger = logging.getLogger(__name__) # ============================================================================= # Data Classes # ============================================================================= @dataclass class WeaponStats: """Weapon statistics from Entropia Nexus.""" name: str damage: Decimal decay_pec: Decimal ammo_pec: Decimal dpp: Decimal # Damage Per PEC (calculated) range: int = 0 attacks_per_min: int = 0 total_cost_pec: Decimal = field(default=None) markup_percent: Decimal = Decimal("100.0") item_id: str = "" 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 def calculate_cost_per_hour(self) -> Decimal: """Calculate total cost per hour of use in PED.""" if 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 def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'name': self.name, 'damage': str(self.damage), 'decay_pec': str(self.decay_pec), 'ammo_pec': str(self.ammo_pec), 'dpp': str(self.dpp), 'range': self.range, 'attacks_per_min': self.attacks_per_min, 'total_cost_pec': str(self.total_cost_pec), 'markup_percent': str(self.markup_percent), 'item_id': self.item_id, } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'WeaponStats': """Create from dictionary.""" return cls( name=data['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, markup_percent=Decimal(data.get('markup_percent', 100)), item_id=data.get('item_id', ''), ) @dataclass class ArmorStats: """Armor piece statistics from Entropia Nexus.""" name: str decay_pec: Decimal protection: Dict[str, Decimal] slot: str = "body" # head, body, arms, legs, feet durability: int = 10000 item_id: str = "" def get_total_protection(self) -> Decimal: """Get sum of all protection values.""" 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")) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" 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, 'item_id': self.item_id, } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ArmorStats': """Create from dictionary.""" return cls( name=data['name'], decay_pec=Decimal(data.get('decay_pec', 0)), protection={k: Decimal(v) for k, v in data.get('protection', {}).items()}, slot=data.get('slot', 'body'), durability=data.get('durability', 10000), item_id=data.get('item_id', ''), ) @dataclass class ToolStats: """Mining tool statistics from Entropia Nexus.""" 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 item_id: str = "" 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 def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" 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), 'item_id': self.item_id, } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ToolStats': """Create from dictionary.""" return cls( name=data['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', ''), ) # ============================================================================= # Mock Data # ============================================================================= MOCK_WEAPONS = { "sollomate_opalo": WeaponStats( name="Sollomate Opalo", damage=Decimal("4.0"), decay_pec=Decimal("0.13"), ammo_pec=Decimal("1.07"), dpp=Decimal("3.33"), range=26, attacks_per_min=56, item_id="sollomate_opalo" ), "omegaton_m2100": WeaponStats( name="Omegaton M2100", damage=Decimal("5.0"), decay_pec=Decimal("0.15"), ammo_pec=Decimal("1.35"), dpp=Decimal("3.33"), range=28, attacks_per_min=54, item_id="omegaton_m2100" ), "breer_m1a": WeaponStats( name="Breer M1a", damage=Decimal("6.0"), decay_pec=Decimal("0.18"), ammo_pec=Decimal("1.62"), dpp=Decimal("3.33"), range=30, attacks_per_min=52, item_id="breer_m1a" ), "castorian_enforcer_se": WeaponStats( name="Castorian Enforcer SE", damage=Decimal("14.0"), decay_pec=Decimal("1.42"), ammo_pec=Decimal("6.08"), dpp=Decimal("1.87"), range=26, attacks_per_min=44, item_id="castorian_enforcer_se" ), "isis_lr1": WeaponStats( name="ISIS LR1", damage=Decimal("8.0"), decay_pec=Decimal("0.32"), ammo_pec=Decimal("2.88"), dpp=Decimal("2.50"), range=60, attacks_per_min=38, item_id="isis_lr1" ), "sollomate_ony": WeaponStats( name="Sollomate Ony", damage=Decimal("12.0"), decay_pec=Decimal("0.89"), ammo_pec=Decimal("5.11"), dpp=Decimal("2.00"), range=24, attacks_per_min=48, item_id="sollomate_ony" ), } MOCK_ARMORS = { "pixie": ArmorStats( name="Pixie", decay_pec=Decimal("0.135"), protection={ "impact": Decimal("3.0"), "cut": Decimal("3.0"), "stab": Decimal("3.0"), "burn": Decimal("2.0"), "cold": Decimal("1.0"), }, slot="body", durability=2800, item_id="pixie" ), "shogun": ArmorStats( name="Shogun", decay_pec=Decimal("0.60"), protection={ "impact": Decimal("10.0"), "cut": Decimal("10.0"), "stab": Decimal("10.0"), "burn": Decimal("6.0"), "cold": Decimal("6.0"), "acid": Decimal("4.0"), }, slot="body", durability=4200, item_id="shogun" ), "ghost": ArmorStats( name="Ghost", decay_pec=Decimal("0.62"), protection={ "impact": Decimal("12.0"), "cut": Decimal("12.0"), "stab": Decimal("12.0"), "burn": Decimal("8.0"), "cold": Decimal("8.0"), "acid": Decimal("6.0"), }, slot="body", durability=4800, item_id="ghost" ), "vigilante": ArmorStats( name="Vigilante", decay_pec=Decimal("0.45"), protection={ "impact": Decimal("8.0"), "cut": Decimal("8.0"), "stab": Decimal("8.0"), "burn": Decimal("5.0"), "cold": Decimal("5.0"), }, slot="body", 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 = { "ziplex_z1": ToolStats( name="Ziplex Z1 Seeker", depth=Decimal("219.5"), radius=Decimal("22.0"), decay_pec=Decimal("0.20"), tool_type="finder", probe_cost=Decimal("0.5"), item_id="ziplex_z1" ), "vrtx_1000": ToolStats( name="VRTX 1000", depth=Decimal("250.0"), radius=Decimal("25.0"), decay_pec=Decimal("0.35"), tool_type="finder", probe_cost=Decimal("0.5"), item_id="vrtx_1000" ), "ziplex_p15": ToolStats( name="Ziplex P15", depth=Decimal("310.0"), radius=Decimal("30.0"), decay_pec=Decimal("0.50"), tool_type="finder", probe_cost=Decimal("0.5"), item_id="ziplex_p15" ), "ore_extractor_md1": ToolStats( name="Ore Extractor MD-1", depth=Decimal("0"), radius=Decimal("0"), decay_pec=Decimal("0.15"), tool_type="extractor", item_id="ore_extractor_md1" ), } # ============================================================================= # 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 get_mock_weapons() -> Dict[str, WeaponStats]: """Get all mock weapons.""" return MOCK_WEAPONS.copy() def get_mock_armors() -> Dict[str, ArmorStats]: """Get all mock armors.""" return MOCK_ARMORS.copy() def get_mock_tools() -> Dict[str, ToolStats]: """Get all mock tools.""" 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 # ============================================================================= # Module Exports # ============================================================================= __all__ = [ 'WeaponStats', 'ArmorStats', 'ToolStats', 'EntropiaNexusAPI', 'SimpleCache', 'MOCK_WEAPONS', 'MOCK_ARMORS', 'MOCK_TOOLS', 'get_mock_weapons', 'get_mock_armors', 'get_mock_tools', 'calculate_dpp', ]