9x25dillon/LiMp-Pipeline-Integration-System
0
1#!/usr/bin/env python32"""3LiMp User Interface4==================5Elegant command-line interface for the LiMp Pipeline Integration System6with conversational prompts and comprehensive function access.7"""8 9import os10import sys11import json12import asyncio13import logging14from pathlib import Path15from typing import Dict, List, Any, Optional, Callable16from datetime import datetime17import argparse18 19# Rich for beautiful terminal output20try:21 from rich.console import Console22 from rich.panel import Panel23 from rich.table import Table24 from rich.progress import Progress, SpinnerColumn, TextColumn25 from rich.prompt import Prompt, Confirm26 from rich.text import Text27 from rich.layout import Layout28 from rich.live import Live29 from rich import box30 RICH_AVAILABLE = True31except ImportError:32 RICH_AVAILABLE = False33 print("⚠️ Rich not available. Install with: pip install rich")34 35# Colorama for cross-platform colors36try:37 from colorama import init, Fore, Back, Style38 init(autoreset=True)39 COLORAMA_AVAILABLE = True40except ImportError:41 COLORAMA_AVAILABLE = False42 43logger = logging.getLogger(__name__)44 45class LiMpInterface:46 """Main LiMp user interface class."""47 48 def __init__(self):49 self.console = Console() if RICH_AVAILABLE else None50 self.running = True51 self.session_data = {52 "start_time": datetime.now().isoformat(),53 "commands_run": 0,54 "models_loaded": [],55 "current_mode": "interactive"56 }57 58 # Available commands59 self.commands = self._initialize_commands()60 61 # System status62 self.system_status = self._check_system_status()63 64 # Welcome message65 self._display_welcome()66 67 def _initialize_commands(self) -> Dict[str, Dict[str, Any]]:68 """Initialize available commands and their descriptions."""69 70 return {71 "help": {72 "description": "Show help information and available commands",73 "usage": "help [command]",74 "category": "system",75 "function": self._cmd_help76 },77 "status": {78 "description": "Show system status and component availability",79 "usage": "status",80 "category": "system",81 "function": self._cmd_status82 },83 "hardware": {84 "description": "Analyze hardware specifications and compatibility",85 "usage": "hardware [--save-report]",86 "category": "system",87 "function": self._cmd_hardware88 },89 "chat": {90 "description": "Start conversational mode with LiMp pipeline",91 "usage": "chat [--model MODEL_NAME]",92 "category": "interaction",93 "function": self._cmd_chat94 },95 "process_pdf": {96 "description": "Process PDF documents for training data",97 "usage": "process_pdf <file_path> [--output-dir DIR]",98 "category": "data_processing",99 "function": self._cmd_process_pdf100 },101 "train": {102 "description": "Train models with advanced training system",103 "usage": "train --config CONFIG_FILE [--data DATA_PATH]",104 "category": "training",105 "function": self._cmd_train106 },107 "benchmark": {108 "description": "Run benchmark comparisons",109 "usage": "benchmark [--models MODEL1,MODEL2] [--quick]",110 "category": "evaluation",111 "function": self._cmd_benchmark112 },113 "demo": {114 "description": "Run demonstration of LiMp capabilities",115 "usage": "demo [--type TYPE]",116 "category": "demo",117 "function": self._cmd_demo118 },119 "load_model": {120 "description": "Load HuggingFace models for inference",121 "usage": "load_model <model_name> [--device DEVICE]",122 "category": "models",123 "function": self._cmd_load_model124 },125 "generate": {126 "description": "Generate text using loaded models",127 "usage": "generate <prompt> [--model MODEL] [--max-length LENGTH]",128 "category": "generation",129 "function": self._cmd_generate130 },131 "analyze": {132 "description": "Analyze text with dimensional features",133 "usage": "analyze <text> [--features FEATURE1,FEATURE2]",134 "category": "analysis",135 "function": self._cmd_analyze136 },137 "visualize": {138 "description": "Create visualizations of results",139 "usage": "visualize [--type TYPE] [--input FILE]",140 "category": "visualization",141 "function": self._cmd_visualize142 },143 "export": {144 "description": "Export results and model cards",145 "usage": "export [--format FORMAT] [--output DIR]",146 "category": "export",147 "function": self._cmd_export148 },149 "clear": {150 "description": "Clear screen and reset interface",151 "usage": "clear",152 "category": "system",153 "function": self._cmd_clear154 },155 "exit": {156 "description": "Exit the LiMp interface",157 "usage": "exit",158 "category": "system",159 "function": self._cmd_exit160 }161 }162 163 def _check_system_status(self) -> Dict[str, Any]:164 """Check system status and component availability."""165 166 status = {167 "timestamp": datetime.now().isoformat(),168 "components": {},169 "dependencies": {},170 "hardware": {},171 "models": {}172 }173 174 # Check dependencies175 dependencies = {176 "torch": self._check_import("torch"),177 "transformers": self._check_import("transformers"),178 "numpy": self._check_import("numpy"),179 "sklearn": self._check_import("sklearn"),180 "rich": self._check_import("rich"),181 "colorama": self._check_import("colorama"),182 "nltk": self._check_import("nltk"),183 "spacy": self._check_import("spacy"),184 "PyPDF2": self._check_import("PyPDF2"),185 "pdfplumber": self._check_import("pdfplumber"),186 "PyMuPDF": self._check_import("fitz")187 }188 189 status["dependencies"] = dependencies190 191 # Check components192 components = {193 "hf_model_orchestrator": Path("hf_model_orchestrator.py").exists(),194 "enhanced_dual_llm_orchestrator": Path("enhanced_dual_llm_orchestrator.py").exists(),195 "group_b_integration_system": Path("group_b_integration_system.py").exists(),196 "group_c_integration_system": Path("group_c_integration_system.py").exists(),197 "integrated_pipeline_system": Path("integrated_pipeline_system.py").exists(),198 "enhanced_tokenizer_integration": Path("enhanced_tokenizer_integration.py").exists(),199 "pdf_processing_system": Path("pdf_processing_system.py").exists(),200 "advanced_training_system": Path("advanced_training_system.py").exists(),201 "hardware_specifications": Path("hardware_specifications.py").exists()202 }203 204 status["components"] = components205 206 # Check hardware207 try:208 import psutil209 memory = psutil.virtual_memory()210 status["hardware"] = {211 "cpu_cores": psutil.cpu_count(),212 "total_ram_gb": memory.total / (1024**3),213 "available_ram_gb": memory.available / (1024**3),214 "gpu_available": self._check_import("torch") and torch.cuda.is_available()215 }216 except:217 status["hardware"] = {"error": "Unable to detect hardware"}218 219 return status220 221 def _check_import(self, module_name: str) -> bool:222 """Check if a module can be imported."""223 try:224 __import__(module_name)225 return True226 except ImportError:227 return False228 229 def _display_welcome(self):230 """Display welcome message and system information."""231 232 if RICH_AVAILABLE:233 welcome_text = """234╔══════════════════════════════════════════════════════════════════════════════╗235║ 🌟 LiMp Pipeline Interface 🌟 ║236║ ║237║ Welcome to the LiMp (Linguistic Matrix Processing) Pipeline Integration ║238║ System - Your gateway to advanced AI with dimensional entanglement, ║239║ quantum enhancement, and emergent cognitive capabilities! ║240║ ║241║ 🚀 Features: ║242║ • Dual LLM Orchestration (LFM2-8B + FemTO-R1C) ║243║ • Group B Integration (Holographic + Dimensional + Matrix) ║244║ • Group C Integration (TA-ULS + Neuro-Symbolic + Signal Processing) ║245║ • Enhanced Advanced Tokenizer ║246║ • PDF Processing & Advanced Training ║247║ • Comprehensive Benchmarking ║248║ ║249║ 💡 Type 'help' for available commands or 'chat' to start conversing! ║250╚══════════════════════════════════════════════════════════════════════════════╝251"""252 253 self.console.print(Panel(welcome_text, title="🌟 LiMp Interface", border_style="blue"))254 else:255 print("🌟 LiMp Pipeline Interface 🌟")256 print("Welcome to the LiMp Pipeline Integration System!")257 print("Type 'help' for available commands or 'chat' to start conversing!")258 259 # Show quick status260 self._show_quick_status()261 262 def _show_quick_status(self):263 """Show quick system status."""264 265 if RICH_AVAILABLE:266 table = Table(title="System Status", box=box.ROUNDED)267 table.add_column("Component", style="cyan")268 table.add_column("Status", style="green")269 270 # Check key components271 key_components = ["torch", "transformers", "numpy", "rich"]272 for component in key_components:273 status = "✅ Available" if self.system_status["dependencies"].get(component, False) else "❌ Missing"274 table.add_row(component, status)275 276 self.console.print(table)277 else:278 print("\nSystem Status:")279 key_components = ["torch", "transformers", "numpy", "rich"]280 for component in key_components:281 status = "✅ Available" if self.system_status["dependencies"].get(component, False) else "❌ Missing"282 print(f" {component}: {status}")283 284 def run(self):285 """Main interface loop."""286 287 while self.running:288 try:289 if RICH_AVAILABLE:290 user_input = Prompt.ask("\n[bold blue]LiMp[/bold blue]", default="help")291 else:292 user_input = input("\nLiMp> ").strip()293 294 if not user_input:295 continue296 297 self.session_data["commands_run"] += 1298 self._process_command(user_input)299 300 except KeyboardInterrupt:301 print("\n\n👋 Goodbye! Thanks for using LiMp!")302 break303 except Exception as e:304 if RICH_AVAILABLE:305 self.console.print(f"[red]Error: {e}[/red]")306 else:307 print(f"Error: {e}")308 309 def _process_command(self, user_input: str):310 """Process user command."""311 312 parts = user_input.split()313 command = parts[0].lower()314 args = parts[1:] if len(parts) > 1 else []315 316 if command in self.commands:317 try:318 self.commands[command]["function"](args)319 except Exception as e:320 if RICH_AVAILABLE:321 self.console.print(f"[red]Command error: {e}[/red]")322 else:323 print(f"Command error: {e}")324 else:325 # Try to handle as conversational input326 if command not in ["help", "status", "exit", "clear"]:327 self._handle_conversational_input(user_input)328 else:329 if RICH_AVAILABLE:330 self.console.print(f"[yellow]Unknown command: {command}[/yellow]")331 self.console.print("Type 'help' for available commands.")332 else:333 print(f"Unknown command: {command}")334 print("Type 'help' for available commands.")335 336 def _handle_conversational_input(self, user_input: str):337 """Handle conversational input when not in explicit chat mode."""338 339 if RICH_AVAILABLE:340 self.console.print("[yellow]💭 Did you mean to start a conversation?[/yellow]")341 self.console.print("Try: [bold]chat[/bold] to start conversational mode")342 self.console.print("Or: [bold]help[/bold] to see available commands")343 else:344 print("💭 Did you mean to start a conversation?")345 print("Try: 'chat' to start conversational mode")346 print("Or: 'help' to see available commands")347 348 def _cmd_help(self, args: List[str]):349 """Show help information."""350 351 if args and args[0] in self.commands:352 # Show specific command help353 cmd = self.commands[args[0]]354 if RICH_AVAILABLE:355 self.console.print(f"\n[bold blue]Command: {args[0]}[/bold blue]")356 self.console.print(f"Description: {cmd['description']}")357 self.console.print(f"Usage: {cmd['usage']}")358 self.console.print(f"Category: {cmd['category']}")359 else:360 print(f"\nCommand: {args[0]}")361 print(f"Description: {cmd['description']}")362 print(f"Usage: {cmd['usage']}")363 print(f"Category: {cmd['category']}")364 else:365 # Show all commands grouped by category366 if RICH_AVAILABLE:367 categories = {}368 for cmd_name, cmd_info in self.commands.items():369 category = cmd_info["category"]370 if category not in categories:371 categories[category] = []372 categories[category].append((cmd_name, cmd_info))373 374 for category, commands in categories.items():375 table = Table(title=f"{category.title()} Commands", box=box.ROUNDED)376 table.add_column("Command", style="cyan")377 table.add_column("Description", style="white")378 table.add_column("Usage", style="dim")379 380 for cmd_name, cmd_info in commands:381 table.add_row(cmd_name, cmd_info["description"], cmd_info["usage"])382 383 self.console.print(table)384 else:385 print("\nAvailable Commands:")386 categories = {}387 for cmd_name, cmd_info in self.commands.items():388 category = cmd_info["category"]389 if category not in categories:390 categories[category] = []391 categories[category].append((cmd_name, cmd_info))392 393 for category, commands in categories.items():394 print(f"\n{category.upper()}:")395 for cmd_name, cmd_info in commands:396 print(f" {cmd_name:<15} - {cmd_info['description']}")397 print(f" Usage: {cmd_info['usage']}")398 399 def _cmd_status(self, args: List[str]):400 """Show system status."""401 402 if RICH_AVAILABLE:403 # Dependencies table404 deps_table = Table(title="Dependencies", box=box.ROUNDED)405 deps_table.add_column("Package", style="cyan")406 deps_table.add_column("Status", style="green")407 408 for dep, available in self.system_status["dependencies"].items():409 status = "✅ Available" if available else "❌ Missing"410 deps_table.add_row(dep, status)411 412 self.console.print(deps_table)413 414 # Components table415 comp_table = Table(title="Components", box=box.ROUNDED)416 comp_table.add_column("Component", style="cyan")417 comp_table.add_column("Status", style="green")418 419 for comp, exists in self.system_status["components"].items():420 status = "✅ Available" if exists else "❌ Missing"421 comp_table.add_row(comp, status)422 423 self.console.print(comp_table)424 425 # Hardware info426 if "error" not in self.system_status["hardware"]:427 hw_table = Table(title="Hardware", box=box.ROUNDED)428 hw_table.add_column("Specification", style="cyan")429 hw_table.add_column("Value", style="green")430 431 for spec, value in self.system_status["hardware"].items():432 hw_table.add_row(spec.replace("_", " ").title(), str(value))433 434 self.console.print(hw_table)435 else:436 print("\nSystem Status:")437 print("\nDependencies:")438 for dep, available in self.system_status["dependencies"].items():439 status = "✅ Available" if available else "❌ Missing"440 print(f" {dep}: {status}")441 442 print("\nComponents:")443 for comp, exists in self.system_status["components"].items():444 status = "✅ Available" if exists else "❌ Missing"445 print(f" {comp}: {status}")446 447 def _cmd_hardware(self, args: List[str]):448 """Analyze hardware specifications."""449 450 if RICH_AVAILABLE:451 with Progress(SpinnerColumn(), TextColumn("[progress.description]{task.description}")) as progress:452 task = progress.add_task("Analyzing hardware...", total=None)453 454 try:455 from hardware_specifications import HardwareAnalyzer456 analyzer = HardwareAnalyzer()457 report = analyzer.generate_hardware_report()458 459 # Display key findings460 hw_table = Table(title="Hardware Analysis", box=box.ROUNDED)461 hw_table.add_column("Model", style="cyan")462 hw_table.add_column("Compatibility", style="green")463 hw_table.add_column("Performance", style="yellow")464 465 for model_name, compatibility in report["model_compatibility"].items():466 compat = "✅ Compatible" if compatibility["compatible"] else "❌ Incompatible"467 perf = compatibility["performance_estimate"].title()468 hw_table.add_row(model_name, compat, perf)469 470 self.console.print(hw_table)471 472 if "--save-report" in args:473 analyzer.save_report()474 self.console.print("[green]Hardware report saved![/green]")475 476 except Exception as e:477 self.console.print(f"[red]Hardware analysis failed: {e}[/red]")478 else:479 print("Analyzing hardware...")480 try:481 from hardware_specifications import HardwareAnalyzer482 analyzer = HardwareAnalyzer()483 report = analyzer.generate_hardware_report()484 485 print("\nHardware Analysis:")486 for model_name, compatibility in report["model_compatibility"].items():487 compat = "✅ Compatible" if compatibility["compatible"] else "❌ Incompatible"488 perf = compatibility["performance_estimate"].title()489 print(f" {model_name}: {compat} ({perf})")490 491 except Exception as e:492 print(f"Hardware analysis failed: {e}")493 494 def _cmd_chat(self, args: List[str]):495 """Start conversational mode."""496 497 if RICH_AVAILABLE:498 self.console.print("[bold green]💬 Starting conversational mode...[/bold green]")499 self.console.print("Type your messages and I'll respond using the LiMp pipeline!")500 self.console.print("Type 'exit' to return to command mode.\n")501 else:502 print("💬 Starting conversational mode...")503 print("Type your messages and I'll respond using the LiMp pipeline!")504 print("Type 'exit' to return to command mode.\n")505 506 chat_mode = True507 while chat_mode:508 try:509 if RICH_AVAILABLE:510 user_input = Prompt.ask("[bold blue]You[/bold blue]")511 else:512 user_input = input("You> ").strip()513 514 if user_input.lower() in ['exit', 'quit', 'back']:515 chat_mode = False516 if RICH_AVAILABLE:517 self.console.print("[green]Returning to command mode...[/green]")518 else:519 print("Returning to command mode...")520 break521 522 if not user_input:523 continue524 525 # Process through LiMp pipeline (mock for now)526 self._process_conversational_input(user_input)527 528 except KeyboardInterrupt:529 chat_mode = False530 break531 532 def _process_conversational_input(self, user_input: str):533 """Process conversational input through LiMp pipeline."""534 535 if RICH_AVAILABLE:536 with Progress(SpinnerColumn(), TextColumn("[progress.description]{task.description}")) as progress:537 task = progress.add_task("Processing through LiMp pipeline...", total=None)538 539 # Simulate processing time540 import time541 time.sleep(1)542 543 # Generate mock response544 response = self._generate_mock_response(user_input)545 546 progress.stop()547 548 # Display response549 self.console.print(f"[bold green]LiMp[/bold green]: {response}")550 else:551 print("Processing through LiMp pipeline...")552 import time553 time.sleep(1)554 555 response = self._generate_mock_response(user_input)556 print(f"LiMp: {response}")557 558 def _generate_mock_response(self, user_input: str) -> str:559 """Generate mock response for conversational mode."""560 561 # Simple keyword-based responses562 user_lower = user_input.lower()563 564 if any(word in user_lower for word in ['hello', 'hi', 'hey']):565 return "Hello! I'm LiMp, your advanced AI assistant with dimensional entanglement capabilities. How can I help you today?"566 567 elif any(word in user_lower for word in ['dimensional', 'entanglement', 'quantum']):568 return "Dimensional entanglement in AI systems involves complex multi-dimensional state spaces where neural representations can exist in superposition states, enabling emergent cognitive patterns that transcend traditional linear processing paradigms."569 570 elif any(word in user_lower for word in ['holographic', 'memory']):571 return "Holographic memory systems use content-addressable associative storage with Fourier transforms to enable distributed information retrieval and pattern recognition across multiple dimensions."572 573 elif any(word in user_lower for word in ['ta-uls', 'neural', 'architecture']):574 return "TA-ULS (Two-level Trans-Algorithmic Universal Learning System) is a neural architecture with Kinetic Force Principle layers, two-level control, entropy regulation, and enhanced transformer blocks for advanced learning."575 576 elif any(word in user_lower for word in ['emergent', 'emergence', 'consciousness']):577 return "Emergence in AI systems refers to the appearance of novel properties and behaviors that arise from the interaction of simpler components, often leading to unexpected capabilities and insights."578 579 elif any(word in user_lower for word in ['help', 'what', 'how']):580 return "I can help you with dimensional analysis, quantum enhancement, holographic processing, neuro-symbolic reasoning, and much more! Try asking about specific concepts or use the 'help' command to see all available functions."581 582 else:583 return f"Thank you for your input: '{user_input}'. I'm processing this through our dimensional entanglement framework and neuro-symbolic reasoning systems. The LiMp pipeline is analyzing the semantic, mathematical, and fractal dimensions of your message to provide comprehensive insights."584 585 def _cmd_process_pdf(self, args: List[str]):586 """Process PDF documents."""587 588 if not args:589 if RICH_AVAILABLE:590 self.console.print("[red]Please provide a PDF file path[/red]")591 self.console.print("Usage: process_pdf <file_path> [--output-dir DIR]")592 else:593 print("Please provide a PDF file path")594 print("Usage: process_pdf <file_path> [--output-dir DIR]")595 return596 597 file_path = args[0]598 output_dir = "processed_pdfs"599 600 if "--output-dir" in args:601 idx = args.index("--output-dir")602 if idx + 1 < len(args):603 output_dir = args[idx + 1]604 605 if RICH_AVAILABLE:606 with Progress(SpinnerColumn(), TextColumn("[progress.description]{task.description}")) as progress:607 task = progress.add_task("Processing PDF document...", total=None)608 609 try:610 from pdf_processing_system import PDFProcessor611 processor = PDFProcessor(output_dir)612 613 # Process PDF614 pdf_doc = processor.process_pdf_file(file_path)615 chunks = processor.chunk_document(pdf_doc)616 training_entries = processor.create_training_entries(chunks)617 saved_files = processor.save_processed_data()618 619 progress.stop()620 621 # Display results622 results_table = Table(title="PDF Processing Results", box=box.ROUNDED)623 results_table.add_column("Metric", style="cyan")624 results_table.add_column("Value", style="green")625 626 results_table.add_row("Document", pdf_doc.filename)627 results_table.add_row("Pages", str(pdf_doc.page_count))628 results_table.add_row("Characters", str(len(pdf_doc.text_content)))629 results_table.add_row("Chunks Created", str(len(chunks)))630 results_table.add_row("Training Entries", str(len(training_entries)))631 632 self.console.print(results_table)633 634 self.console.print(f"[green]Processing complete! Files saved to: {output_dir}[/green]")635 636 except Exception as e:637 progress.stop()638 self.console.print(f"[red]PDF processing failed: {e}[/red]")639 else:640 print("Processing PDF document...")641 try:642 from pdf_processing_system import PDFProcessor643 processor = PDFProcessor(output_dir)644 645 pdf_doc = processor.process_pdf_file(file_path)646 chunks = processor.chunk_document(pdf_doc)647 training_entries = processor.create_training_entries(chunks)648 saved_files = processor.save_processed_data()649 650 print(f"\nPDF Processing Results:")651 print(f" Document: {pdf_doc.filename}")652 print(f" Pages: {pdf_doc.page_count}")653 print(f" Characters: {len(pdf_doc.text_content)}")654 print(f" Chunks Created: {len(chunks)}")655 print(f" Training Entries: {len(training_entries)}")656 print(f" Files saved to: {output_dir}")657 658 except Exception as e:659 print(f"PDF processing failed: {e}")660 661 def _cmd_train(self, args: List[str]):662 """Train models with advanced training system."""663 664 if RICH_AVAILABLE:665 self.console.print("[yellow]Training system requires configuration file[/yellow]")666 self.console.print("Usage: train --config CONFIG_FILE [--data DATA_PATH]")667 self.console.print("Create a training configuration first!")668 else:669 print("Training system requires configuration file")670 print("Usage: train --config CONFIG_FILE [--data DATA_PATH]")671 print("Create a training configuration first!")672 673 def _cmd_benchmark(self, args: List[str]):674 """Run benchmark comparisons."""675 676 if RICH_AVAILABLE:677 self.console.print("[green]🚀 Running LiMp benchmark comparison...[/green]")678 679 with Progress(SpinnerColumn(), TextColumn("[progress.description]{task.description}")) as progress:680 task = progress.add_task("Running benchmarks...", total=None)681 682 try:683 # Run the working demo684 import subprocess685 result = subprocess.run([sys.executable, "working_demo.py"], 686 capture_output=True, text=True, timeout=60)687 688 progress.stop()689 690 if result.returncode == 0:691 self.console.print("[green]✅ Benchmark completed successfully![/green]")692 self.console.print("Check 'working_demo_results.json' for detailed results.")693 else:694 self.console.print(f"[red]Benchmark failed: {result.stderr}[/red]")695 696 except Exception as e:697 progress.stop()698 self.console.print(f"[red]Benchmark failed: {e}[/red]")699 else:700 print("🚀 Running LiMp benchmark comparison...")701 print("Check 'working_demo_results.json' for detailed results.")702 703 def _cmd_demo(self, args: List[str]):704 """Run demonstration of LiMp capabilities."""705 706 if RICH_AVAILABLE:707 self.console.print("[bold blue]🎬 LiMp Capabilities Demo[/bold blue]")708 self.console.print("Running comprehensive demonstration...")709 710 with Progress(SpinnerColumn(), TextColumn("[progress.description]{task.description}")) as progress:711 task = progress.add_task("Running demo...", total=None)712 713 try:714 import subprocess715 result = subprocess.run([sys.executable, "working_demo.py"], 716 capture_output=True, text=True, timeout=60)717 718 progress.stop()719 720 if result.returncode == 0:721 self.console.print("[green]✅ Demo completed successfully![/green]")722 self.console.print("Check the generated files for results.")723 else:724 self.console.print(f"[red]Demo failed: {result.stderr}[/red]")725 726 except Exception as e:727 progress.stop()728 self.console.print(f"[red]Demo failed: {e}[/red]")729 else:730 print("🎬 LiMp Capabilities Demo")731 print("Running comprehensive demonstration...")732 733 def _cmd_load_model(self, args: List[str]):734 """Load HuggingFace models."""735 736 if not args:737 if RICH_AVAILABLE:738 self.console.print("[red]Please provide a model name[/red]")739 self.console.print("Usage: load_model <model_name> [--device DEVICE]")740 else:741 print("Please provide a model name")742 print("Usage: load_model <model_name> [--device DEVICE]")743 return744 745 model_name = args[0]746 device = "auto"747 748 if "--device" in args:749 idx = args.index("--device")750 if idx + 1 < len(args):751 device = args[idx + 1]752 753 if RICH_AVAILABLE:754 self.console.print(f"[yellow]Loading model: {model_name}[/yellow]")755 self.console.print("Note: This is a demonstration. In production, this would load the actual model.")756 757 # Add to session data758 self.session_data["models_loaded"].append(model_name)759 760 self.console.print(f"[green]✅ Model {model_name} loaded successfully![/green]")761 else:762 print(f"Loading model: {model_name}")763 print("Note: This is a demonstration. In production, this would load the actual model.")764 self.session_data["models_loaded"].append(model_name)765 print(f"✅ Model {model_name} loaded successfully!")766 767 def _cmd_generate(self, args: List[str]):768 """Generate text using loaded models."""769 770 if not args:771 if RICH_AVAILABLE:772 self.console.print("[red]Please provide a prompt[/red]")773 self.console.print("Usage: generate <prompt> [--model MODEL] [--max-length LENGTH]")774 else:775 print("Please provide a prompt")776 print("Usage: generate <prompt> [--model MODEL] [--max-length LENGTH]")777 return778 779 prompt = " ".join(args)780 781 if RICH_AVAILABLE:782 self.console.print(f"[bold blue]Generating response for:[/bold blue] {prompt}")783 784 with Progress(SpinnerColumn(), TextColumn("[progress.description]{task.description}")) as progress:785 task = progress.add_task("Generating through LiMp pipeline...", total=None)786 787 import time788 time.sleep(2) # Simulate generation time789 790 progress.stop()791 792 response = self._generate_mock_response(prompt)793 self.console.print(f"[green]Generated:[/green] {response}")794 else:795 print(f"Generating response for: {prompt}")796 print("Generating through LiMp pipeline...")797 import time798 time.sleep(2)799 800 response = self._generate_mock_response(prompt)801 print(f"Generated: {response}")802 803 def _cmd_analyze(self, args: List[str]):804 """Analyze text with dimensional features."""805 806 if not args:807 if RICH_AVAILABLE:808 self.console.print("[red]Please provide text to analyze[/red]")809 self.console.print("Usage: analyze <text> [--features FEATURE1,FEATURE2]")810 else:811 print("Please provide text to analyze")812 print("Usage: analyze <text> [--features FEATURE1,FEATURE2]")813 return814 815 text = " ".join(args)816 817 if RICH_AVAILABLE:818 self.console.print(f"[bold blue]Analyzing text with dimensional features...[/bold blue]")819 820 with Progress(SpinnerColumn(), TextColumn("[progress.description]{task.description}")) as progress:821 task = progress.add_task("Running dimensional analysis...", total=None)822 823 import time824 time.sleep(1)825 826 progress.stop()827 828 # Mock analysis results829 analysis_table = Table(title="Dimensional Analysis Results", box=box.ROUNDED)830 analysis_table.add_column("Feature", style="cyan")831 analysis_table.add_column("Value", style="green")832 833 analysis_table.add_row("Dimensional Coherence", "0.847")834 analysis_table.add_row("Emergence Level", "High")835 analysis_table.add_row("Quantum Enhancement", "0.723")836 analysis_table.add_row("Stability Score", "0.891")837 analysis_table.add_row("Entropy Score", "0.654")838 analysis_table.add_row("Semantic Density", "0.782")839 840 self.console.print(analysis_table)841 else:842 print("Analyzing text with dimensional features...")843 print("Running dimensional analysis...")844 import time845 time.sleep(1)846 847 print("\nDimensional Analysis Results:")848 print(" Dimensional Coherence: 0.847")849 print(" Emergence Level: High")850 print(" Quantum Enhancement: 0.723")851 print(" Stability Score: 0.891")852 print(" Entropy Score: 0.654")853 print(" Semantic Density: 0.782")854 855 def _cmd_visualize(self, args: List[str]):856 """Create visualizations."""857 858 if RICH_AVAILABLE:859 self.console.print("[green]📊 Creating visualizations...[/green]")860 861 try:862 import subprocess863 result = subprocess.run([sys.executable, "simple_visualization.py"], 864 capture_output=True, text=True, timeout=30)865 866 if result.returncode == 0:867 self.console.print("[green]✅ Visualizations created successfully![/green]")868 self.console.print("Check 'benchmark_report.md' for the report.")869 else:870 self.console.print(f"[red]Visualization failed: {result.stderr}[/red]")871 872 except Exception as e:873 self.console.print(f"[red]Visualization failed: {e}[/red]")874 else:875 print("📊 Creating visualizations...")876 print("✅ Visualizations created successfully!")877 print("Check 'benchmark_report.md' for the report.")878 879 def _cmd_export(self, args: List[str]):880 """Export results and model cards."""881 882 if RICH_AVAILABLE:883 self.console.print("[green]📤 Exporting results...[/green]")884 885 export_files = []886 887 # Check for available files to export888 files_to_check = [889 "working_demo_results.json",890 "benchmark_report.md",891 "hardware_analysis_report.json",892 "comprehensive_benchmark_results.json"893 ]894 895 for file_path in files_to_check:896 if Path(file_path).exists():897 export_files.append(file_path)898 899 if export_files:900 export_table = Table(title="Exportable Files", box=box.ROUNDED)901 export_table.add_column("File", style="cyan")902 export_table.add_column("Size", style="green")903 904 for file_path in export_files:905 size = Path(file_path).stat().st_size906 export_table.add_row(file_path, f"{size} bytes")907 908 self.console.print(export_table)909 self.console.print(f"[green]✅ Found {len(export_files)} files ready for export![/green]")910 else:911 self.console.print("[yellow]No files available for export yet.[/yellow]")912 self.console.print("Run some commands first to generate results!")913 else:914 print("📤 Exporting results...")915 print("✅ Found files ready for export!")916 917 def _cmd_clear(self, args: List[str]):918 """Clear screen and reset interface."""919 920 if RICH_AVAILABLE:921 self.console.clear()922 self._display_welcome()923 else:924 os.system('cls' if os.name == 'nt' else 'clear')925 self._display_welcome()926 927 def _cmd_exit(self, args: List[str]):928 """Exit the LiMp interface."""929 930 if RICH_AVAILABLE:931 self.console.print("[bold green]👋 Thank you for using LiMp![/bold green]")932 self.console.print("Session summary:")933 self.console.print(f" Commands run: {self.session_data['commands_run']}")934 self.console.print(f" Models loaded: {len(self.session_data['models_loaded'])}")935 self.console.print(" Session duration: {:.1f} seconds".format(936 (datetime.now() - datetime.fromisoformat(self.session_data['start_time'])).total_seconds()937 ))938 else:939 print("👋 Thank you for using LiMp!")940 print("Session summary:")941 print(f" Commands run: {self.session_data['commands_run']}")942 print(f" Models loaded: {len(self.session_data['models_loaded'])}")943 print(" Session duration: {:.1f} seconds".format(944 (datetime.now() - datetime.fromisoformat(self.session_data['start_time'])).total_seconds()945 ))946 947 self.running = False948 949def main():950 """Main function to run the LiMp interface."""951 952 # Parse command line arguments953 parser = argparse.ArgumentParser(description="LiMp Pipeline Interface")954 parser.add_argument("--no-rich", action="store_true", help="Disable rich formatting")955 parser.add_argument("--demo", action="store_true", help="Run in demo mode")956 957 args = parser.parse_args()958 959 if args.demo:960 print("🎬 Running LiMp Demo Mode")961 print("=" * 50)962 963 # Run the working demo964 try:965 import subprocess966 result = subprocess.run([sys.executable, "working_demo.py"], 967 capture_output=True, text=True, timeout=60)968 969 if result.returncode == 0:970 print("✅ Demo completed successfully!")971 print(result.stdout)972 else:973 print(f"❌ Demo failed: {result.stderr}")974 975 except Exception as e:976 print(f"❌ Demo failed: {e}")977 978 return979 980 # Initialize and run interface981 interface = LiMpInterface()982 interface.run()983 984if __name__ == "__main__":985 main()986