Lemontropia-Suite/core/nexus_api.py

879 lines
29 KiB
Python

"""
Entropia Nexus API Client - Full Implementation
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/
"""
import asyncio
import json
import logging
from dataclasses import dataclass, field, asdict
from decimal import Decimal, InvalidOperation
from typing import Dict, List, Optional, Any, Union
from functools import wraps
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
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
range: int = 0
attacks_per_min: int = 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
# 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 not self.attacks_per_min or self.attacks_per_min <= 0:
return Decimal("0")
shots_per_hour = self.attacks_per_min * 60
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."""
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) 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.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=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 set statistics."""
name: str
decay_pec: Decimal
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:
"""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.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."""
return {
'name': self.name,
'decay_pec': str(self.decay_pec),
'protection': {k: str(v) for k, v in self.protection.items()},
'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.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'),
item_id=data.get('item_id', ''),
markup_percent=Decimal(data.get('markup_percent', 100)),
)
@dataclass
class ToolStats:
"""Mining/scanning tool statistics."""
name: str
depth: Decimal
radius: Decimal
decay_pec: Decimal
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 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."""
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) 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.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: Dict[str, WeaponStats] = {
"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",
item_class="Laser Rifle"
),
"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",
item_class="Laser Carbine"
),
"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",
item_class="BLP Rifle"
),
"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",
item_class="BLP Pistol"
),
"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",
item_class="Laser Sniper"
),
"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",
item_class="Laser Pistol"
),
}
MOCK_ARMORS: Dict[str, ArmorStats] = {
"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=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.55"),
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"
),
}
MOCK_TOOLS: Dict[str, ToolStats] = {
"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"
),
}
# =============================================================================
# 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()
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 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
# =============================================================================
# Module Exports
# =============================================================================
__all__ = [
'WeaponStats',
'ArmorStats',
'ArmorPiece',
'ToolStats',
'GearLoadout',
'EntropiaNexusAPI',
'SimpleCache',
'MOCK_WEAPONS',
'MOCK_ARMORS',
'MOCK_TOOLS',
'calculate_dpp',
'get_mock_weapons',
'get_mock_armors',
'get_mock_tools',
'get_weapon_by_name',
]