Lemontropia-Suite/modules/loot_analyzer.py

360 lines
13 KiB
Python

"""
Lemontropia Suite - Enhanced Loot Analyzer
Detailed loot breakdown by mob type, item category, and value ranges.
"""
import json
import logging
from decimal import Decimal
from pathlib import Path
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from datetime import datetime
from collections import defaultdict
logger = logging.getLogger(__name__)
@dataclass
class LootItem:
"""Individual loot item record."""
name: str
quantity: int
value_ped: Decimal
timestamp: datetime
mob_name: Optional[str] = None
session_id: Optional[int] = None
@dataclass
class LootStats:
"""Statistics for a specific item type."""
item_name: str
total_count: int = 0
total_value: Decimal = field(default_factory=lambda: Decimal("0"))
max_value: Decimal = field(default_factory=lambda: Decimal("0"))
min_value: Decimal = field(default_factory=lambda: Decimal("999999"))
avg_value: Decimal = field(default_factory=lambda: Decimal("0"))
def add_loot(self, value: Decimal, count: int = 1):
"""Add loot to statistics."""
self.total_count += count
self.total_value += value
if value > self.max_value:
self.max_value = value
if value < self.min_value:
self.min_value = value
self.avg_value = self.total_value / self.total_count if self.total_count > 0 else Decimal("0")
@dataclass
class MobStats:
"""Statistics for a specific mob type."""
mob_name: str
kill_count: int = 0
total_loot: Decimal = field(default_factory=lambda: Decimal("0"))
item_breakdown: Dict[str, LootStats] = field(default_factory=dict)
def add_kill(self, loot_value: Decimal, items: List[Tuple[str, Decimal]]):
"""Record a kill with its loot."""
self.kill_count += 1
self.total_loot += loot_value
for item_name, value in items:
if item_name not in self.item_breakdown:
self.item_breakdown[item_name] = LootStats(item_name)
self.item_breakdown[item_name].add_loot(value)
class LootAnalyzer:
"""
Advanced loot analysis for hunting sessions.
Features:
- Track loot by mob type
- Item category breakdown
- Value distribution analysis
- Session comparison
"""
def __init__(self, data_dir: Optional[Path] = None):
self.data_dir = data_dir or Path.home() / ".lemontropia" / "loot_analysis"
self.data_dir.mkdir(parents=True, exist_ok=True)
# Current session data
self.current_session_items: List[LootItem] = []
self.mob_stats: Dict[str, MobStats] = defaultdict(lambda: MobStats(""))
self.global_items: List[LootItem] = [] # Track globals separately
# Item categories (expandable)
self.item_categories = {
'shrapnel': ['shrapnel'],
'ores': ['iron stone', 'lysterium stone', 'belkar stone', 'caldorite stone'],
'enmatters': ['muscle oil', 'tir oil', 'crude oil'],
'animal_parts': ['wool', 'hide', 'meat', 'bone'],
'robot_parts': ['robot component', 'robot filter'],
'weapons': ['weapon', 'rifle', 'pistol', 'sword'],
'armor': ['armor', 'harness', 'helmet', 'boots'],
'misc': []
}
self._load_historical_data()
def _load_historical_data(self):
"""Load historical loot data for comparison."""
try:
history_file = self.data_dir / "loot_history.json"
if history_file.exists():
with open(history_file, 'r') as f:
data = json.load(f)
# Could restore previous session stats here
except Exception as e:
logger.error(f"Failed to load loot history: {e}")
def _save_historical_data(self):
"""Save current session data for future comparison."""
try:
history_file = self.data_dir / "loot_history.json"
# Save summary stats
data = {
'last_session': datetime.now().isoformat(),
'total_mobs': sum(m.kill_count for m in self.mob_stats.values()),
'total_loot': str(sum(m.total_loot for m in self.mob_stats.values())),
}
with open(history_file, 'w') as f:
json.dump(data, f, indent=2)
except Exception as e:
logger.error(f"Failed to save loot history: {e}")
def categorize_item(self, item_name: str) -> str:
"""Categorize an item by name."""
name_lower = item_name.lower()
for category, items in self.item_categories.items():
for item_pattern in items:
if item_pattern in name_lower:
return category
return 'misc'
def record_loot(self, item_name: str, quantity: int, value_ped: Decimal,
mob_name: Optional[str] = None, is_global: bool = False) -> None:
"""Record a loot item."""
item = LootItem(
name=item_name,
quantity=quantity,
value_ped=value_ped,
timestamp=datetime.now(),
mob_name=mob_name
)
self.current_session_items.append(item)
if is_global:
self.global_items.append(item)
# Update mob stats if we know the mob
if mob_name:
if mob_name not in self.mob_stats:
self.mob_stats[mob_name] = MobStats(mob_name)
# Note: This is simplified - ideally we batch items per kill
def get_session_summary(self) -> Dict:
"""Get summary of current session."""
total_items = len(self.current_session_items)
total_value = sum(item.value_ped for item in self.current_session_items)
# Category breakdown
category_values = defaultdict(lambda: Decimal("0"))
for item in self.current_session_items:
cat = self.categorize_item(item.name)
category_values[cat] += item.value_ped
return {
'total_items': total_items,
'total_value': total_value,
'globals_count': len(self.global_items),
'category_breakdown': dict(category_values),
'mob_types': len(self.mob_stats),
}
def get_top_loot(self, n: int = 10) -> List[LootItem]:
"""Get top N highest value loot items."""
sorted_items = sorted(
self.current_session_items,
key=lambda x: x.value_ped,
reverse=True
)
return sorted_items[:n]
def get_mob_efficiency(self) -> Dict[str, Decimal]:
"""Get loot per kill for each mob type."""
return {
mob_name: stats.total_loot / stats.kill_count
for mob_name, stats in self.mob_stats.items()
if stats.kill_count > 0
}
def generate_report(self) -> str:
"""Generate text report of loot analysis."""
summary = self.get_session_summary()
report = []
report.append("=" * 50)
report.append("LOOT ANALYSIS REPORT")
report.append("=" * 50)
report.append(f"Total Items: {summary['total_items']}")
report.append(f"Total Value: {summary['total_value']:.2f} PED")
report.append(f"Globals: {summary['globals_count']}")
report.append(f"Mob Types: {summary['mob_types']}")
report.append("")
report.append("CATEGORY BREAKDOWN:")
for cat, value in sorted(summary['category_breakdown'].items(),
key=lambda x: x[1], reverse=True):
pct = (value / summary['total_value'] * 100) if summary['total_value'] > 0 else 0
report.append(f" {cat.capitalize():12} {value:8.2f} PED ({pct:5.1f}%)")
report.append("")
report.append("TOP 5 LOOT ITEMS:")
for i, item in enumerate(self.get_top_loot(5), 1):
report.append(f" {i}. {item.name[:25]:25} {item.value_ped:8.2f} PED")
report.append("")
report.append("MOB EFFICIENCY:")
for mob, efficiency in sorted(self.get_mob_efficiency().items(),
key=lambda x: x[1], reverse=True)[:5]:
stats = self.mob_stats[mob]
report.append(f" {mob[:20]:20} {efficiency:6.2f} PED/kill ({stats.kill_count} kills)")
return "\n".join(report)
def export_to_csv(self, filepath: Path) -> None:
"""Export loot data to CSV."""
import csv
with open(filepath, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(['Timestamp', 'Item', 'Quantity', 'Value (PED)', 'Mob', 'Category'])
for item in self.current_session_items:
writer.writerow([
item.timestamp.isoformat(),
item.name,
item.quantity,
str(item.value_ped),
item.mob_name or 'Unknown',
self.categorize_item(item.name)
])
class DPSCalculator:
"""
Real-time DPS (Damage Per Second) and DPP (Damage Per Pec) calculator.
"""
def __init__(self):
self.damage_events: List[Tuple[datetime, Decimal]] = []
self.total_damage = Decimal("0")
self.total_cost = Decimal("0") # In PEC
def record_damage(self, damage: Decimal, cost_pec: Decimal = Decimal("0")):
"""Record a damage event."""
self.damage_events.append((datetime.now(), damage))
self.total_damage += damage
self.total_cost += cost_pec
# Keep only last 60 seconds for DPS calculation
cutoff = datetime.now() - timedelta(seconds=60)
self.damage_events = [(t, d) for t, d in self.damage_events if t > cutoff]
def get_current_dps(self) -> Decimal:
"""Calculate DPS over last 60 seconds."""
if len(self.damage_events) < 2:
return Decimal("0")
total_damage = sum(d for _, d in self.damage_events)
time_span = (self.damage_events[-1][0] - self.damage_events[0][0]).total_seconds()
if time_span > 0:
return Decimal(str(total_damage)) / Decimal(str(time_span))
return Decimal("0")
def get_dpp(self) -> Decimal:
"""Calculate Damage Per Pec (efficiency metric)."""
if self.total_cost > 0:
return self.total_damage / self.total_cost
return Decimal("0")
def get_session_stats(self) -> Dict:
"""Get overall session stats."""
return {
'total_damage': self.total_damage,
'total_cost_pec': self.total_cost,
'total_cost_ped': self.total_cost / 100,
'dpp': self.get_dpp(),
'current_dps': self.get_current_dps(),
'hit_count': len(self.damage_events),
}
class GlobalAlertSystem:
"""
Alert system for globals and HoFs.
Plays sound, shows notification, auto-screenshots.
"""
def __init__(self, screenshot_dir: Optional[Path] = None):
self.screenshot_dir = screenshot_dir or Path.home() / ".lemontropia" / "screenshots"
self.screenshot_dir.mkdir(parents=True, exist_ok=True)
self.global_sound_enabled = True
self.hof_sound_enabled = True
self.auto_screenshot = True
def on_global(self, value_ped: Decimal, item_name: str):
"""Handle global event."""
logger.info(f"🌟 GLOBAL! {item_name} - {value_ped:.2f} PED")
if self.global_sound_enabled:
self._play_sound("global")
if self.auto_screenshot:
self._take_screenshot(f"global_{datetime.now():%Y%m%d_%H%M%S}")
def on_hof(self, value_ped: Decimal, item_name: str):
"""Handle Hall of Fame event."""
logger.info(f"🏆 HALL OF FAME! {item_name} - {value_ped:.2f} PED")
if self.hof_sound_enabled:
self._play_sound("hof")
if self.auto_screenshot:
self._take_screenshot(f"hof_{datetime.now():%Y%m%d_%H%M%S}")
def _play_sound(self, sound_type: str):
"""Play alert sound."""
try:
if sys.platform == 'win32':
import winsound
if sound_type == "hof":
winsound.MessageBeep(winsound.MB_ICONEXCLAMATION)
else:
winsound.MessageBeep(winsound.MB_OK)
except Exception as e:
logger.error(f"Failed to play sound: {e}")
def _take_screenshot(self, filename: str):
"""Take screenshot of game window."""
try:
# This would integrate with the vision system
# For now, just log it
screenshot_path = self.screenshot_dir / f"{filename}.png"
logger.info(f"Screenshot saved: {screenshot_path}")
except Exception as e:
logger.error(f"Failed to take screenshot: {e}")
# Export main classes
__all__ = ['LootAnalyzer', 'DPSCalculator', 'GlobalAlertSystem', 'LootStats', 'MobStats']