FCI_WaterMaker_API/demo_dts_test.py

#!/usr/bin/env python3
"""
Demo script to show DTS API Test Suite capabilities without running a full DTS sequence.
This demonstrates the monitoring and reporting features.
"""

import sys
import time
from dts_api_test_suite import DTSAPITester, TestConfig

def demo_api_connectivity():
    """Demo API connectivity testing"""
    print("🔍 Demo: API Connectivity Test")
    print("=" * 50)
    
    tester = DTSAPITester()
    
    print("Testing API connection...")
    if tester.check_system_status():
        print("✅ API is accessible and PLC is connected")
        
        # Test getting current step progress
        print("\nTesting current step progress endpoint...")
        progress = tester.get_current_step_progress()
        if progress:
            current_mode = progress.get("current_mode", "unknown")
            timer_based = progress.get("timer_based_progress", False)
            print(f"📊 Current system mode: {current_mode}")
            print(f"⏱️  Timer-based progress available: {timer_based}")
            
            if timer_based:
                timer_addr = progress.get("timer_address")
                timer_val = progress.get("current_timer_value")
                progress_pct = progress.get("progress_percent", 0)
                print(f"   Timer R{timer_addr}: {timer_val} ({progress_pct}% progress)")
        else:
            print("⚠️  Could not get current step progress")
        
        return True
    else:
        print("❌ API connectivity failed")
        return False

def demo_dts_status_monitoring():
    """Demo DTS status monitoring without starting DTS"""
    print("\n🔍 Demo: DTS Status Monitoring")
    print("=" * 50)
    
    tester = DTSAPITester()
    
    # Check if there are any existing DTS tasks
    print("Checking for existing DTS tasks...")
    response, _ = tester._make_api_request("GET", "/dts/status")
    
    if response:
        latest_task = response.get("latest_task")
        total_tasks = response.get("total_tasks", 0)
        active_tasks = response.get("active_tasks", [])
        
        print(f"📋 Total DTS tasks in history: {total_tasks}")
        print(f"🔄 Active tasks: {len(active_tasks)}")
        
        if latest_task:
            task_id = latest_task.get("task_id", "unknown")
            status = latest_task.get("status", "unknown")
            current_step = latest_task.get("current_step", "unknown")
            progress = latest_task.get("progress_percent", 0)
            
            print(f"📊 Latest task: {task_id}")
            print(f"   Status: {status}")
            print(f"   Current step: {current_step}")
            print(f"   Progress: {progress}%")
            
            if latest_task.get("timer_info"):
                timer_info = latest_task["timer_info"]
                print(f"   Timer info: Mode {timer_info.get('current_mode')}, "
                      f"Timer R{timer_info.get('timer_address', 'N/A')}")
        else:
            print("ℹ️  No previous DTS tasks found")
    else:
        print("❌ Could not retrieve DTS status")

def demo_configuration():
    """Demo configuration options"""
    print("\n🔍 Demo: Configuration Options")
    print("=" * 50)
    
    # Show default configuration
    config = TestConfig()
    print("Default configuration:")
    print(f"  API URL: {config.API_BASE_URL}")
    print(f"  Polling interval: {config.POLLING_INTERVAL}s")
    print(f"  Request timeout: {config.REQUEST_TIMEOUT}s")
    print(f"  Log file enabled: {config.LOG_FILE_ENABLED}")
    print(f"  CSV export enabled: {config.CSV_EXPORT_ENABLED}")
    
    # Show expected screen durations
    print("\nExpected screen durations:")
    for mode, duration in config.EXPECTED_SCREEN_DURATIONS.items():
        screen_names = {5: "Priming", 6: "Init", 8: "Fresh Water Flush"}
        screen_name = screen_names.get(int(mode), f"Mode {mode}")
        print(f"  {screen_name}: {duration}s")

def demo_api_endpoints():
    """Demo available API endpoints"""
    print("\n🔍 Demo: Available API Endpoints")
    print("=" * 50)
    
    tester = DTSAPITester()
    
    # Test config endpoint
    print("Testing API configuration endpoint...")
    response, _ = tester._make_api_request("GET", "/config")
    
    if response:
        api_version = response.get("api_version", "unknown")
        endpoints = response.get("endpoints", {})
        
        print(f"📡 API Version: {api_version}")
        print("📋 Available endpoints:")
        
        # Show DTS-related endpoints
        dts_endpoints = {k: v for k, v in endpoints.items() if "dts" in k.lower()}
        for endpoint, description in dts_endpoints.items():
            print(f"  {endpoint}: {description}")
    else:
        print("❌ Could not retrieve API configuration")

def main():
    """Run all demos"""
    print("🚀 DTS API Test Suite - Demo Mode")
    print("=" * 60)
    print("This demo shows the test suite capabilities without starting DTS")
    print("=" * 60)
    
    # Run demos
    success = True
    
    try:
        # Demo 1: API Connectivity
        if not demo_api_connectivity():
            success = False
        
        # Demo 2: DTS Status Monitoring
        demo_dts_status_monitoring()
        
        # Demo 3: Configuration
        demo_configuration()
        
        # Demo 4: API Endpoints
        demo_api_endpoints()
        
        print("\n" + "=" * 60)
        if success:
            print("✅ Demo completed successfully!")
            print("\nTo run actual DTS monitoring:")
            print("  python run_dts_test.py basic")
            print("  python dts_api_test_suite.py --verbose")
        else:
            print("❌ Demo completed with issues")
            print("Check API server status and PLC connection")
        print("=" * 60)
        
    except KeyboardInterrupt:
        print("\n👋 Demo interrupted by user")
    except Exception as e:
        print(f"\n❌ Demo error: {e}")
        success = False
    
    return 0 if success else 1

if __name__ == "__main__":
    sys.exit(main())