481 lines
18 KiB
Python
481 lines
18 KiB
Python
# Description: Main entry point for Lemontropia Suite
|
|
# Provides CLI interface for user testing the Data Capture Engine
|
|
# CORRECTED TERMINOLOGY: Projects = Activities (Hunt/Mine/Craft), Sessions = Gameplay Instances
|
|
|
|
import sys
|
|
import asyncio
|
|
import signal
|
|
import os
|
|
from pathlib import Path
|
|
from decimal import Decimal
|
|
from datetime import datetime
|
|
import logging
|
|
|
|
# Load .env file before anything else
|
|
from dotenv import load_dotenv
|
|
env_path = Path(__file__).parent / ".env"
|
|
load_dotenv(dotenv_path=env_path)
|
|
|
|
# Add core to path
|
|
core_dir = Path(__file__).parent / "core"
|
|
sys.path.insert(0, str(core_dir))
|
|
|
|
from core.database import DatabaseManager
|
|
from core.project_manager import ProjectManager, ProjectData, SessionData, LootEvent
|
|
from core.log_watcher import LogWatcher, MockLogGenerator
|
|
|
|
# Configure logging for user visibility
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s | %(levelname)-8s | %(message)s',
|
|
datefmt='%H:%M:%S'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LemontropiaApp:
|
|
"""
|
|
Main application class for user testing.
|
|
|
|
TERMINOLOGY CLARIFICATION:
|
|
- PROJECT: A long-term activity (e.g., "Argo Hunting", "Calypso Mining")
|
|
- SESSION: A single gameplay instance within a project (e.g., "2-hour hunt")
|
|
- RUN: A complete project lifecycle (multiple sessions)
|
|
|
|
Example:
|
|
Project: "Daily Argo Grind"
|
|
├── Session 1: Morning hunt (45 min, +15 PED)
|
|
├── Session 2: Evening hunt (2 hrs, -5 PED)
|
|
└── Session 3: Weekend marathon (5 hrs, +200 PED)
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.db = DatabaseManager()
|
|
self.pm = ProjectManager(self.db)
|
|
self.watcher = None
|
|
self._running = False
|
|
|
|
# Initialize database
|
|
logger.info("🍋 Initializing Lemontropia Suite...")
|
|
if not self.db.initialize():
|
|
logger.error("Failed to initialize database!")
|
|
sys.exit(1)
|
|
|
|
logger.info("✅ Database ready")
|
|
|
|
# Ensure test data exists
|
|
self._ensure_mock_data()
|
|
|
|
def _ensure_mock_data(self):
|
|
"""Create mock chat.log if it doesn't exist."""
|
|
test_data_dir = Path(__file__).parent / "test-data"
|
|
test_data_dir.mkdir(exist_ok=True)
|
|
|
|
mock_log = test_data_dir / "mock-chat.log"
|
|
if not mock_log.exists():
|
|
logger.info("📝 Generating mock chat.log for testing...")
|
|
MockLogGenerator.create_mock_file(mock_log, lines=50)
|
|
logger.info(f"✅ Mock log created: {mock_log}")
|
|
|
|
def print_header(self):
|
|
"""Print application header."""
|
|
print("\n" + "="*65)
|
|
print(" 🍋 LEMONTROPIA SUITE — User Test Build v0.1.0")
|
|
if os.getenv('USE_MOCK_DATA', 'true').lower() in ('true', '1', 'yes'):
|
|
print(" Core Data Capture Engine — Mock Mode")
|
|
else:
|
|
print(" Core Data Capture Engine — Live Mode")
|
|
print("="*65)
|
|
print("\n 📚 TERMINOLOGY:")
|
|
print(" • PROJECT = Activity type (Hunt/Mine/Craft)")
|
|
print(" • SESSION = Single gameplay instance")
|
|
print(" • RUN = Complete project with multiple sessions")
|
|
|
|
def print_menu(self):
|
|
"""Print main menu."""
|
|
print("\n📋 MAIN MENU")
|
|
print("-" * 45)
|
|
print(" 1. 🎯 Create New Project (Activity)")
|
|
print(" 2. 📂 View All Projects")
|
|
print(" 3. ▶️ Start New Session (Live Tracking)")
|
|
print(" 4. 📊 View Project Statistics")
|
|
print(" 5. 🗄️ Archive/Complete Project")
|
|
print(" 6. 🧹 Reset Database (WARNING)")
|
|
print(" 0. 🚪 Exit")
|
|
print("-" * 45)
|
|
|
|
def create_project(self):
|
|
"""Create a new project (activity definition)."""
|
|
print("\n🎯 CREATE NEW PROJECT")
|
|
print("-" * 50)
|
|
print(" A PROJECT defines your activity type:")
|
|
print(" • 'Daily Argo Grind' (Hunting)")
|
|
print(" • 'Calypso Mining Route' (Mining)")
|
|
print(" • 'Weapon Crafting Batch' (Crafting)")
|
|
print("-" * 50)
|
|
|
|
name = input("Project name: ").strip()
|
|
if not name:
|
|
print("❌ Name required")
|
|
return
|
|
|
|
print("\nActivity type:")
|
|
print(" 1. hunt - Combat & creature looting")
|
|
print(" 2. mine - Resource extraction")
|
|
print(" 3. craft - Manufacturing items")
|
|
print(" 4. inventory - Asset management")
|
|
|
|
type_choice = input("Select type (1-4): ").strip()
|
|
type_map = {'1': 'hunt', '2': 'mine', '3': 'craft', '4': 'inventory'}
|
|
project_type = type_map.get(type_choice, 'hunt')
|
|
|
|
project = self.pm.create_project(name, project_type)
|
|
print(f"\n✅ Created PROJECT: {project.name}")
|
|
print(f" Type: {project_type}")
|
|
print(f" ID: {project.id}")
|
|
print(f"\n Next: Start a SESSION to track gameplay (Option 3)")
|
|
|
|
# Show current projects
|
|
self.list_projects()
|
|
|
|
def list_projects(self):
|
|
"""List all projects (activities)."""
|
|
print("\n📂 PROJECTS (Activities)")
|
|
print("-" * 65)
|
|
print(" These are your defined hunting/mining/crafting activities")
|
|
print("-" * 65)
|
|
|
|
projects = self.pm.list_projects()
|
|
|
|
if not projects:
|
|
print(" No projects found. Create one first! (Option 1)")
|
|
return
|
|
|
|
print(f" {'ID':<5} {'Name':<22} {'Type':<10} {'Status':<10} {'Created'}")
|
|
print(" " + "-" * 63)
|
|
|
|
for p in projects:
|
|
created = p.created_at.strftime("%Y-%m-%d") if p.created_at else "N/A"
|
|
print(f" {p.id:<5} {p.name:<22} {p.type:<10} {p.status:<10} {created}")
|
|
|
|
print(f"\n Total: {len(projects)} project(s)")
|
|
print("\n 💡 Tip: Select a project and start a SESSION to track gameplay")
|
|
|
|
async def start_live_session(self):
|
|
"""Start a live session (single gameplay instance)."""
|
|
print("\n▶️ START NEW SESSION (Live Tracking)")
|
|
print("-" * 65)
|
|
print(" A SESSION is a single gameplay instance within a project.")
|
|
print(" Example: 'Morning Argo Hunt' or 'Mining Run #5'")
|
|
print("-" * 65)
|
|
|
|
# Select project
|
|
projects = self.pm.list_projects(status='active')
|
|
if not projects:
|
|
print("❌ No active projects. Create one first (Option 1).")
|
|
return
|
|
|
|
print("Select PROJECT for this session:")
|
|
for i, p in enumerate(projects, 1):
|
|
print(f" {i}. {p.name} ({p.type})")
|
|
|
|
choice = input(f"\nSelect (1-{len(projects)}): ").strip()
|
|
try:
|
|
project = projects[int(choice) - 1]
|
|
except (ValueError, IndexError):
|
|
print("❌ Invalid selection")
|
|
return
|
|
|
|
print(f"\n📋 Starting SESSION for: {project.name}")
|
|
session_notes = input("Session notes (optional): ").strip()
|
|
|
|
# Start session
|
|
session = self.pm.start_session(project.id, notes=session_notes)
|
|
print(f"✅ SESSION started: ID {session.id}")
|
|
|
|
# Setup log watcher - use real log or mock based on .env
|
|
use_mock = os.getenv('USE_MOCK_DATA', 'true').lower() in ('true', '1', 'yes')
|
|
log_path = os.getenv('EU_CHAT_LOG_PATH', '')
|
|
|
|
if use_mock or not log_path:
|
|
test_data_dir = Path(__file__).parent / "test-data"
|
|
mock_log = test_data_dir / "mock-chat.log"
|
|
self.watcher = LogWatcher(str(mock_log), poll_interval=2.0, mock_mode=True)
|
|
logger.info("Using MOCK data for testing")
|
|
else:
|
|
self.watcher = LogWatcher(log_path, poll_interval=1.0, mock_mode=False)
|
|
logger.info(f"Using REAL log: {log_path}")
|
|
|
|
# Stats tracking
|
|
stats = {'loot': 0, 'globals': 0, 'hofs': 0, 'skills': 0, 'level_ups': 0, 'weapon_tiers': 0, 'damage_dealt': 0, 'damage_taken': 0, 'evades': 0, 'total_ped': Decimal('0.0')}
|
|
|
|
def on_event(event):
|
|
"""Handle log events."""
|
|
if event.event_type == 'loot':
|
|
item_name = event.data.get('item_name', 'Unknown')
|
|
# Skip Universal Ammo - it's converted shrapnel, not loot
|
|
if item_name == 'Universal Ammo':
|
|
return
|
|
loot = LootEvent(
|
|
item_name=item_name,
|
|
quantity=event.data.get('quantity', 1),
|
|
value_ped=event.data.get('value_ped', Decimal('0.0')),
|
|
event_type='regular',
|
|
raw_log_line=event.raw_line
|
|
)
|
|
self.pm.record_loot(loot)
|
|
stats['loot'] += 1
|
|
stats['total_ped'] += loot.value_ped
|
|
print(f" 💰 Loot: {loot.item_name} x{loot.quantity} ({loot.value_ped} PED)")
|
|
|
|
elif event.event_type == 'global':
|
|
stats['globals'] += 1
|
|
print(f" 🌍 GLOBAL: {event.data.get('player_name')} found {event.data.get('value_ped')} PED!")
|
|
|
|
elif event.event_type == 'hof':
|
|
stats['hofs'] += 1
|
|
print(f" 🏆 HALL OF FAME: {event.data.get('value_ped')} PED!")
|
|
|
|
elif event.event_type == 'skill':
|
|
stats['skills'] += 1
|
|
print(f" 📈 Skill: {event.data.get('skill_name')} +{event.data.get('gained')}")
|
|
|
|
elif event.event_type == 'level_up':
|
|
stats['level_ups'] += 1
|
|
print(f" 🎉 LEVEL UP: {event.data.get('skill_name')} reached level {event.data.get('new_level')}!")
|
|
|
|
elif event.event_type == 'weapon_tier':
|
|
stats['weapon_tiers'] += 1
|
|
print(f" ⬆️ TIER UP: {event.data.get('weapon_name')} reached tier {event.data.get('new_tier')}!")
|
|
|
|
elif event.event_type == 'damage_dealt':
|
|
stats['damage_dealt'] += 1
|
|
print(f" 💥 Damage Dealt: {event.data.get('damage')} pts")
|
|
|
|
elif event.event_type == 'critical_hit':
|
|
stats['damage_dealt'] += 1 # Count as damage dealt too
|
|
print(f" 💀 CRITICAL: {event.data.get('damage')} pts")
|
|
|
|
elif event.event_type == 'damage_taken':
|
|
stats['damage_taken'] += 1
|
|
print(f" 🛡️ Damage Taken: {event.data.get('damage')} pts")
|
|
|
|
elif event.event_type == 'evade':
|
|
stats['evades'] += 1
|
|
print(f" ✨ {event.data.get('type', 'Evade')}")
|
|
|
|
# Subscribe to events
|
|
self.watcher.subscribe('loot', on_event)
|
|
self.watcher.subscribe('global', on_event)
|
|
self.watcher.subscribe('hof', on_event)
|
|
self.watcher.subscribe('skill', on_event)
|
|
self.watcher.subscribe('level_up', on_event)
|
|
self.watcher.subscribe('weapon_tier', on_event)
|
|
self.watcher.subscribe('damage_dealt', on_event)
|
|
self.watcher.subscribe('critical_hit', on_event)
|
|
self.watcher.subscribe('damage_taken', on_event)
|
|
self.watcher.subscribe('evade', on_event)
|
|
|
|
print("\n" + "="*50)
|
|
print("🔴 LIVE SESSION RUNNING")
|
|
print("="*50)
|
|
print(f" Project: {project.name}")
|
|
print(f" Session ID: {session.id}")
|
|
print(" Watching chat.log for events...")
|
|
print(" Press Ctrl+C to end session\n")
|
|
|
|
await self.watcher.start()
|
|
|
|
try:
|
|
while self._running:
|
|
await asyncio.sleep(1)
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
await self.watcher.stop()
|
|
|
|
# End session
|
|
self.pm.end_session(session.id)
|
|
|
|
print("\n" + "="*50)
|
|
print("✅ SESSION ENDED")
|
|
print("="*50)
|
|
print(f"\n📊 SESSION SUMMARY:")
|
|
print(f" Loot events: {stats['loot']}")
|
|
print(f" Globals: {stats['globals']}")
|
|
print(f" HoFs: {stats['hofs']}")
|
|
print(f" Skills: {stats['skills']}")
|
|
print(f" Level Ups: {stats['level_ups']}")
|
|
print(f" Weapon Tiers: {stats['weapon_tiers']}")
|
|
print(f" Damage Dealt: {stats['damage_dealt']}")
|
|
print(f" Damage Taken: {stats['damage_taken']}")
|
|
print(f" Evades/Misses: {stats['evades']}")
|
|
print(f" Total Value: {stats['total_ped']} PED")
|
|
print(f"\n View full stats: Option 4 (Project Statistics)")
|
|
|
|
def view_project_stats(self):
|
|
"""View statistics for a project (all sessions combined)."""
|
|
print("\n📊 PROJECT STATISTICS")
|
|
print("-" * 65)
|
|
print(" View combined stats for all sessions in a project")
|
|
print("-" * 65)
|
|
|
|
projects = self.pm.list_projects()
|
|
if not projects:
|
|
print(" No projects found.")
|
|
return
|
|
|
|
print("Select project to analyze:")
|
|
for i, p in enumerate(projects, 1):
|
|
print(f" {i}. {p.name}")
|
|
|
|
choice = input(f"\nSelect (1-{len(projects)}): ").strip()
|
|
try:
|
|
project = projects[int(choice) - 1]
|
|
except (ValueError, IndexError):
|
|
print("❌ Invalid selection")
|
|
return
|
|
|
|
summary = self.pm.get_project_summary(project.id)
|
|
if not summary:
|
|
print("❌ Could not load summary")
|
|
return
|
|
|
|
print(f"\n" + "="*50)
|
|
print(f"📈 STATS FOR: {summary['name']}")
|
|
print("="*50)
|
|
print(f" Activity Type: {summary['type']}")
|
|
print(f" Status: {summary['status']}")
|
|
print(f" Total Sessions: {summary['session_count']}")
|
|
print(f"\n 💰 FINANCIALS:")
|
|
print(f" Total Spent: {summary['total_spent']} PED")
|
|
print(f" Total Return: {summary['total_return']} PED")
|
|
print(f" Net Profit: {summary['net_profit']} PED")
|
|
if summary['total_spent'] > 0:
|
|
roi = (summary['net_profit'] / summary['total_spent']) * 100
|
|
print(f" ROI: {roi:.2f}%")
|
|
print(f"\n 🏆 ACHIEVEMENTS:")
|
|
print(f" Globals: {summary['global_count']}")
|
|
print(f" Hall of Fames: {summary['hof_count']}")
|
|
|
|
def archive_project(self):
|
|
"""Archive a completed project."""
|
|
print("\n🗄️ ARCHIVE PROJECT")
|
|
print("-" * 50)
|
|
print(" Archive a project when you're done with it.")
|
|
print(" Archived projects are kept for historical comparison.")
|
|
print("-" * 50)
|
|
|
|
projects = self.pm.list_projects()
|
|
if not projects:
|
|
print(" No projects found.")
|
|
return
|
|
|
|
print("Select project to archive:")
|
|
for i, p in enumerate(projects, 1):
|
|
status_icon = "🟢" if p.status == 'active' else "⚪"
|
|
print(f" {i}. {status_icon} {p.name} ({p.status})")
|
|
|
|
choice = input(f"\nSelect (1-{len(projects)}): ").strip()
|
|
try:
|
|
project = projects[int(choice) - 1]
|
|
except (ValueError, IndexError):
|
|
print("❌ Invalid selection")
|
|
return
|
|
|
|
print(f"\n⚠️ Archive '{project.name}'?")
|
|
print(" This will mark the project as completed.")
|
|
confirm = input("Type 'archive' to confirm: ").strip().lower()
|
|
|
|
if confirm == 'archive':
|
|
self.pm.archive_project(project.id)
|
|
print(f"✅ Archived: {project.name}")
|
|
else:
|
|
print("Cancelled")
|
|
|
|
def reset_database(self):
|
|
"""Reset database (for testing)."""
|
|
print("\n🧹 RESET DATABASE")
|
|
print("-" * 50)
|
|
print("⚠️ WARNING: This will DELETE all data!")
|
|
print(" All projects, sessions, and loot data will be lost.")
|
|
|
|
confirm = input("\nType 'RESET' to confirm: ").strip()
|
|
if confirm == 'RESET':
|
|
db_path = self.db.db_path
|
|
self.db.close()
|
|
|
|
import os
|
|
if db_path.exists():
|
|
os.remove(db_path)
|
|
print(f"✅ Database deleted: {db_path}")
|
|
|
|
# Reinitialize
|
|
self.db = DatabaseManager()
|
|
self.db.initialize()
|
|
self.pm = ProjectManager(self.db)
|
|
print("✅ Database reinitialized (empty)")
|
|
else:
|
|
print("Cancelled")
|
|
|
|
async def run(self):
|
|
"""Main application loop."""
|
|
self._running = True
|
|
|
|
# Setup signal handlers
|
|
def signal_handler():
|
|
self._running = False
|
|
print("\n\n🛑 Shutting down...")
|
|
|
|
signal.signal(signal.SIGINT, lambda s, f: signal_handler())
|
|
|
|
self.print_header()
|
|
|
|
while self._running:
|
|
self.print_menu()
|
|
choice = input("Select option: ").strip()
|
|
|
|
try:
|
|
if choice == '1':
|
|
self.create_project()
|
|
elif choice == '2':
|
|
self.list_projects()
|
|
elif choice == '3':
|
|
await self.start_live_session()
|
|
elif choice == '4':
|
|
self.view_project_stats()
|
|
elif choice == '5':
|
|
self.archive_project()
|
|
elif choice == '6':
|
|
self.reset_database()
|
|
elif choice == '0':
|
|
self._running = False
|
|
print("\n" + "="*50)
|
|
print("🍋 Thank you for testing Lemontropia Suite!")
|
|
print("="*50)
|
|
else:
|
|
print("❌ Invalid option")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error: {e}")
|
|
print(f"❌ Error: {e}")
|
|
|
|
# Cleanup
|
|
self.db.close()
|
|
|
|
|
|
def main():
|
|
"""Application entry point."""
|
|
app = LemontropiaApp()
|
|
|
|
try:
|
|
asyncio.run(app.run())
|
|
except KeyboardInterrupt:
|
|
print("\n\n🛑 Interrupted by user")
|
|
finally:
|
|
print("\n👋 Goodbye!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|