cad-Integration/CADDesigner-Code-main/test/demo_context_system.py

#!/usr/bin/env python3
"""
Context system demo script.

Shows practical Context system usage scenarios:
1. Create a conversation context
2. Add messages
3. Persist to file
4. Restore from file
5. Search historical messages
"""

import os
import sys
import asyncio
import tempfile
import shutil
from datetime import datetime

# Add the project root directory to the Python path.
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, project_root)

from context.context_manager import ContextManager
from context.context import RedisFileContextBackend
from context.schemas import Message


async def demo_context_system():
    """Demonstrate the complete Context system functionality."""
    
    print("=" * 60)
    print("Context System Feature Demo")
    print("=" * 60)
    
    # Create a temporary directory.
    demo_dir = tempfile.mkdtemp(prefix="demo_context_")
    
    try:
        # 1. Create a context manager.
        print("\n1. Creating context manager...")
        manager = ContextManager(backend_class=RedisFileContextBackend)
        
        # 2. Create a conversation context.
        print("\n2. Creating conversation context...")
        context_id = "demo_conversation_001"
        context = manager.create_context(
            context_id=context_id,
            max_history_length=10
        )
        print(f"✓ Created context: {context_id}")
        
        # 3. Simulate a conversation.
        print("\n3. Simulating conversation...")
        conversation = [
            ("user", "Hello, I want to learn about Python programming"),
            ("assistant", "Hello! Python is a very popular programming language. What features are you interested in?"),
            ("user", "What are Python's advantages?"),
            ("assistant", "Python's main advantages include:\n1. Concise and readable syntax\n2. Rich libraries and frameworks\n3. Cross-platform support\n4. Beginner friendliness"),
            ("user", "I want to learn machine learning. Is Python suitable?"),
            ("assistant", "Absolutely! Python is very popular in machine learning and has many excellent libraries such as TensorFlow, PyTorch, and scikit-learn."),
            ("user", "Thanks for the introduction"),
            ("assistant", "You're welcome! If you have any Python or machine learning questions, feel free to ask.")
        ]
        
        # Add conversation messages.
        for role, content in conversation:
            success = await manager.add_message(
                context_id=context_id,
                role=role,
                content=content
            )
            if success:
                print(f"✓ Added {role} message: {content[:30]}...")
            else:
                print(f"✗ Failed to add {role} message")
        
        # 4. View conversation history.
        print("\n4. Viewing conversation history...")
        history = manager.get_history(context_id)
        print(f"✓ Conversation history contains {len(history)} messages")
        
        for i, message in enumerate(history[-3:], 1):  # Show the last 3 messages.
            print(f"  {i}. {message.role}: {message.content[:50]}...")
        
        # 5. Search historical messages.
        print("\n5. Searching historical messages...")
        search_results = context.search_messages("Python", limit=5)
        print(f"✓ Found {len(search_results)} messages containing 'Python'")
        
        for i, message in enumerate(search_results, 1):
            print(f"  {i}. {message.role}: {message.content[:50]}...")
        
        # 6. Persist to file.
        print("\n6. Persisting to file...")
        success = await context.persist()
        if success:
            print("✓ Successfully persisted to the file system")
            
            # Check file contents.
            file_path = context.file_path
            if os.path.exists(file_path):
                file_size = os.path.getsize(file_path)
                print(f"✓ File size: {file_size} bytes")
        else:
            print("✗ Persistence failed")
        
        # 7. Verify Redis storage.
        print("\n7. Verifying Redis storage...")
        message_count = context.get_message_count()
        metadata = context.get_metadata()
        print(f"✓ Redis stores {message_count} messages")
        print(f"✓ Metadata contains {len(metadata)} fields")
        
        # 8. Simulate system restart (restore from file).
        print("\n8. Simulating system restart...")
        
        # Create a new context manager to simulate a restart.
        new_manager = ContextManager(backend_class=RedisFileContextBackend)
        new_context = new_manager.get_context(context_id)
        
        if new_context:
            restored_history = new_context.retrieve_messages()
            print(f"✓ Successfully restored conversation history with {len(restored_history)} messages")
            
            # Verify restored data.
            if len(restored_history) == len(history):
                print("✓ Data integrity verification passed")
            else:
                print("✗ Data integrity verification failed")
        else:
            print("✗ Failed to restore conversation history")
        
        # 9. Display system statistics.
        print("\n9. System statistics...")
        contexts = manager.list_contexts()
        print(f"✓ Current system has {len(contexts)} contexts")
        
        for ctx_info in contexts:
            print(f"  - {ctx_info['context_id']}: {ctx_info.get('total_messages', 0)} messages")
        
        print("\n" + "=" * 60)
        print("Demo complete!")
        print("=" * 60)
        
        print("\n🎉 Context system feature verification succeeded:")
        print("  ✓ Message storage and retrieval work correctly")
        print("  ✓ File system persistence works correctly")
        print("  ✓ Message search works correctly")
        print("  ✓ Data recovery after system restart works correctly")
        print("  ✓ Redis storage works correctly")
        
        print(f"\n📁 Demo file location: {demo_dir}")
        print("💡 Inspect the generated JSON file to understand the data format")
        
    except Exception as e:
        print(f"❌ Error during demo: {e}")
        import traceback
        traceback.print_exc()
    
    finally:
        # Clean up demo files.
        if os.path.exists(demo_dir):
            shutil.rmtree(demo_dir)


if __name__ == "__main__":
    asyncio.run(demo_context_system())