342 lines
12 KiB
Python
342 lines
12 KiB
Python
# 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)
|