# 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, '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': loot = LootEvent( item_name=event.data.get('item_name', 'Unknown'), 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 == 'damage_dealt': stats['damage_dealt'] += 1 print(f" 💥 Damage Dealt: {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('damage_dealt', 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" 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()