impulsivefps
/
Lemontropia-Suite
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Lemontropia-Suite/core/nexus_api.py

841 lines
26 KiB
Python
Raw Blame History

"""
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.70"),
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
"""
total_cost_ped = (decay_pec + ammo_pec) / 100
if total_cost_ped == 0:
return Decimal("0")
return damage / total_cost_ped
# =============================================================================
# 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',
]
Reference in New Issue View Git Blame Copy Permalink