# Description: Session cost tracking using loadout configuration # Tracks actual weapon shots, armor hits, and heals during hunting # Standards: Python 3.11+, type hints, Decimal precision, Observer Pattern import logging from decimal import Decimal from typing import Optional, Dict, Any, Callable from dataclasses import dataclass, field from datetime import datetime from core.loadout_db import LoadoutDatabase from core.database import DatabaseManager logger = logging.getLogger(__name__) @dataclass class SessionCostState: """Real-time cost tracking state for a hunting session.""" # Counters shots_fired: int = 0 hits_taken: int = 0 heals_used: int = 0 # Costs (PED) weapon_cost: Decimal = field(default_factory=lambda: Decimal("0")) armor_cost: Decimal = field(default_factory=lambda: Decimal("0")) healing_cost: Decimal = field(default_factory=lambda: Decimal("0")) enhancer_cost: Decimal = field(default_factory=lambda: Decimal("0")) mindforce_cost: Decimal = field(default_factory=lambda: Decimal("0")) @property def total_cost(self) -> Decimal: """Total cost including all categories.""" return self.weapon_cost + self.armor_cost + self.healing_cost + self.enhancer_cost + self.mindforce_cost class SessionCostTracker: """ Tracks real-time costs during a hunting session based on loadout configuration. Uses per-action costs from loadout: - Cost per shot (weapon + ammo + enhancers + amp) - Cost per hit (armor decay) - Cost per heal (FAP/chip decay) """ def __init__(self, session_id: int, loadout_id: int, db_manager: Optional[DatabaseManager] = None): """ Initialize cost tracker for a session. Args: session_id: Hunting session ID loadout_id: Loadout ID to use for cost calculations db_manager: Optional database manager """ self.session_id = session_id self.loadout_id = loadout_id self.db = db_manager or DatabaseManager() self.loadout_db = LoadoutDatabase(self.db) # Load loadout costs self._load_loadout_costs() # State self.state = SessionCostState() self._callbacks: List[Callable[[SessionCostState], None]] = [] logger.info(f"Cost tracker initialized for session {session_id} with loadout {loadout_id}") def _load_loadout_costs(self): """Load per-action costs from loadout.""" conn = self.db.get_connection() row = conn.execute( "SELECT cost_per_shot_ped, cost_per_hit_ped, cost_per_heal_ped, enhancers_json, mindforce_decay_pec " "FROM loadouts WHERE id = ?", (self.loadout_id,) ).fetchone() if row: self.cost_per_shot = Decimal(str(row['cost_per_shot_ped'] or 0)) self.cost_per_hit = Decimal(str(row['cost_per_hit_ped'] or 0)) self.cost_per_heal = Decimal(str(row['cost_per_heal_ped'] or 0)) self.mindforce_decay_pec = Decimal(str(row['mindforce_decay_pec'] or 0)) # Parse enhancers for additional decay import json self.enhancer_decay_per_shot = Decimal("0") if row['enhancers_json']: try: enhancers = json.loads(row['enhancers_json']) for tier, enh in enhancers.items(): self.enhancer_decay_per_shot += Decimal(str(enh.get('decay', 0))) except json.JSONDecodeError: pass logger.debug(f"Loaded costs: shot={self.cost_per_shot}, hit={self.cost_per_hit}, heal={self.cost_per_heal}") else: logger.error(f"Loadout {self.loadout_id} not found") self.cost_per_shot = Decimal("0") self.cost_per_hit = Decimal("0") self.cost_per_heal = Decimal("0") self.enhancer_decay_per_shot = Decimal("0") self.mindforce_decay_pec = Decimal("0") def register_callback(self, callback: Callable[[SessionCostState], None]): """ Register a callback to be called when costs update. Args: callback: Function that takes SessionCostState """ self._callbacks.append(callback) def unregister_callback(self, callback: Callable[[SessionCostState], None]): """Unregister a callback.""" if callback in self._callbacks: self._callbacks.remove(callback) def _notify(self): """Notify all callbacks of state change.""" for callback in self._callbacks: try: callback(self.state) except Exception as e: logger.error(f"Callback error: {e}") def record_shot(self, count: int = 1): """ Record weapon shots fired. Args: count: Number of shots (default 1) """ self.state.shots_fired += count # Calculate costs weapon_cost = self.cost_per_shot * count enhancer_cost = self.enhancer_decay_per_shot * Decimal("0.01") * count # Convert PEC to PED self.state.weapon_cost += weapon_cost self.state.enhancer_cost += enhancer_cost self._persist_update() self._notify() logger.debug(f"Recorded {count} shot(s), cost: {weapon_cost + enhancer_cost:.4f} PED") def record_hit(self, damage: Optional[Decimal] = None, count: int = 1): """ Record hits taken (armor damage). Args: damage: Damage amount (if None, uses average from loadout) count: Number of hits """ self.state.hits_taken += count # Calculate cost if damage: # Scale cost based on actual damage # Assuming cost_per_hit is for average damage cost = self.cost_per_hit * damage * count else: cost = self.cost_per_hit * count self.state.armor_cost += cost self._persist_update() self._notify() logger.debug(f"Recorded {count} hit(s), cost: {cost:.4f} PED") def record_heal(self, amount: Optional[Decimal] = None, count: int = 1, is_mindforce: bool = False): """ Record healing used. Args: amount: Heal amount (if None, uses average from loadout) count: Number of heals is_mindforce: Whether using mindforce implant """ self.state.heals_used += count if is_mindforce and self.mindforce_decay_pec > 0: # Mindforce heal cost = self.mindforce_decay_pec * Decimal("0.01") * count self.state.mindforce_cost += cost else: # Regular FAP heal if amount: cost = self.cost_per_heal * amount * count else: cost = self.cost_per_heal * count self.state.healing_cost += cost self._persist_update() self._notify() logger.debug(f"Recorded {count} heal(s), cost: {cost:.4f} PED") def _persist_update(self): """Persist current state to database.""" try: conn = self.db.get_connection() conn.execute(""" UPDATE hunting_sessions SET weapon_cost_ped = ?, armor_cost_ped = ?, healing_cost_ped = ?, enhancer_cost_ped = ?, mindforce_cost_ped = ?, total_cost_ped = ?, shots_fired = ?, hits_taken = ?, heals_used = ? WHERE session_id = ? """, ( float(self.state.weapon_cost), float(self.state.armor_cost), float(self.state.healing_cost), float(self.state.enhancer_cost), float(self.state.mindforce_cost), float(self.state.total_cost), self.state.shots_fired, self.state.hits_taken, self.state.heals_used, self.session_id )) conn.commit() except Exception as e: logger.error(f"Failed to persist cost update: {e}") def get_summary(self) -> Dict[str, Any]: """ Get cost summary for display. Returns: Dict with cost breakdown and counters """ return { 'shots_fired': self.state.shots_fired, 'hits_taken': self.state.hits_taken, 'heals_used': self.state.heals_used, 'weapon_cost': self.state.weapon_cost, 'armor_cost': self.state.armor_cost, 'healing_cost': self.state.healing_cost, 'enhancer_cost': self.state.enhancer_cost, 'mindforce_cost': self.state.mindforce_cost, 'total_cost': self.state.total_cost, 'cost_per_shot': self.cost_per_shot, 'cost_per_hit': self.cost_per_hit, 'cost_per_heal': self.cost_per_heal } def reset(self): """Reset all counters and costs.""" self.state = SessionCostState() self._persist_update() self._notify() logger.info(f"Reset cost tracker for session {self.session_id}") class SessionCostIntegration: """ Integrates cost tracking with log watcher events. Connects to log watcher signals and updates cost tracker accordingly. """ def __init__(self, cost_tracker: SessionCostTracker): """ Initialize integration. Args: cost_tracker: SessionCostTracker instance """ self.cost_tracker = cost_tracker self._enabled = False logger.info("Session cost integration initialized") def enable(self, log_watcher: 'LogWatcher'): """ Enable cost tracking by connecting to log watcher signals. Args: log_watcher: LogWatcher instance to connect to """ if self._enabled: return # Connect to log watcher signals # These signals would need to be added to LogWatcher if hasattr(log_watcher, 'shot_fired'): log_watcher.shot_fired.connect(self._on_shot_fired) if hasattr(log_watcher, 'damage_taken'): log_watcher.damage_taken.connect(self._on_damage_taken) if hasattr(log_watcher, 'heal_used'): log_watcher.heal_used.connect(self._on_heal_used) self._enabled = True logger.info("Cost tracking enabled") def disable(self, log_watcher: Optional['LogWatcher'] = None): """ Disable cost tracking. Args: log_watcher: Optional LogWatcher to disconnect from """ if not self._enabled: return if log_watcher: if hasattr(log_watcher, 'shot_fired'): log_watcher.shot_fired.disconnect(self._on_shot_fired) if hasattr(log_watcher, 'damage_taken'): log_watcher.damage_taken.disconnect(self._on_damage_taken) if hasattr(log_watcher, 'heal_used'): log_watcher.heal_used.disconnect(self._on_heal_used) self._enabled = False logger.info("Cost tracking disabled") def _on_shot_fired(self, weapon_name: str): """Handle shot fired event.""" self.cost_tracker.record_shot() def _on_damage_taken(self, damage: Decimal, creature_name: str): """Handle damage taken event.""" self.cost_tracker.record_hit(damage) def _on_heal_used(self, amount: Decimal, tool_name: str): """Handle heal used event.""" # Detect if mindforce based on tool name is_mindforce = 'chip' in tool_name.lower() or 'implant' in tool_name.lower() self.cost_tracker.record_heal(amount, is_mindforce=is_mindforce)