cluster4npu/tests/test_final_fix.py

#!/usr/bin/env python3
"""
Final test to verify all fixes are working correctly
"""
import sys
import os
import json

# Add paths
current_dir = os.path.dirname(os.path.abspath(__file__))
parent_dir = os.path.dirname(current_dir)
sys.path.append(os.path.join(parent_dir, 'core', 'functions'))

def test_converter_with_postprocessing():
    """Test the mflow converter with postprocessing fixes"""
    
    print("=" * 60)
    print("Testing MFlow Converter with Postprocessing Fixes")
    print("=" * 60)
    
    try:
        from mflow_converter import MFlowConverter
        
        # Test with the fixed mflow file
        mflow_file = 'multi_series_yolov5_fixed.mflow'
        
        if not os.path.exists(mflow_file):
            print(f"❌ Test file not found: {mflow_file}")
            return False
        
        print(f"✓ Loading {mflow_file}...")
        
        converter = MFlowConverter()
        config = converter.load_and_convert(mflow_file)
        
        print(f"✓ Conversion successful!")
        print(f"  - Pipeline name: {config.pipeline_name}")
        print(f"  - Total stages: {len(config.stage_configs)}")
        
        # Check each stage for postprocessor
        for i, stage_config in enumerate(config.stage_configs, 1):
            print(f"\n  Stage {i}: {stage_config.stage_id}")
            
            if stage_config.stage_postprocessor:
                options = stage_config.stage_postprocessor.options
                print(f"    ✅ Postprocessor found!")
                print(f"       Type: {options.postprocess_type.value}")
                print(f"       Threshold: {options.threshold}")
                print(f"       Classes: {len(options.class_names)} ({options.class_names[:3]}...)")
                print(f"       NMS Threshold: {options.nms_threshold}")
                
                if options.postprocess_type.value == 'yolo_v5':
                    print(f"    🎉 YOLOv5 postprocessing correctly configured!")
                else:
                    print(f"    ⚠ Postprocessing type: {options.postprocess_type.value}")
            else:
                print(f"    ❌ No postprocessor found")
        
        return True
        
    except Exception as e:
        print(f"❌ Converter test failed: {e}")
        import traceback
        traceback.print_exc()
        return False

def test_multidongle_postprocessing():
    """Test MultiDongle postprocessing directly"""
    
    print(f"\n" + "=" * 60)
    print("Testing MultiDongle Postprocessing")
    print("=" * 60)
    
    try:
        from Multidongle import MultiDongle, PostProcessorOptions, PostProcessType
        
        # Create YOLOv5 postprocessor options
        options = PostProcessorOptions(
            postprocess_type=PostProcessType.YOLO_V5,
            threshold=0.3,
            class_names=["person", "bicycle", "car", "motorbike", "aeroplane"],
            nms_threshold=0.5,
            max_detections_per_class=50
        )
        
        print(f"✓ Created PostProcessorOptions:")
        print(f"  - Type: {options.postprocess_type.value}")
        print(f"  - Threshold: {options.threshold}")
        print(f"  - Classes: {len(options.class_names)}")
        
        # Test with dummy MultiDongle
        multidongle = MultiDongle(
            port_id=[1],  # Dummy port
            postprocess_options=options
        )
        
        print(f"✓ Created MultiDongle with postprocessing")
        print(f"  - Postprocess type: {multidongle.postprocess_options.postprocess_type.value}")
        
        # Test set_postprocess_options method
        new_options = PostProcessorOptions(
            postprocess_type=PostProcessType.YOLO_V5,
            threshold=0.25,
            class_names=["person", "car", "truck"],
            nms_threshold=0.4
        )
        
        multidongle.set_postprocess_options(new_options)
        print(f"✓ Updated postprocess options:")
        print(f"  - New threshold: {multidongle.postprocess_options.threshold}")
        print(f"  - New classes: {len(multidongle.postprocess_options.class_names)}")
        
        return True
        
    except Exception as e:
        print(f"❌ MultiDongle test failed: {e}")
        import traceback
        traceback.print_exc()
        return False

def show_fix_summary():
    """Show comprehensive fix summary"""
    
    print(f"\n" + "=" * 60)
    print("COMPREHENSIVE FIX SUMMARY")
    print("=" * 60)
    
    print(f"\n🔧 Applied Fixes:")
    print(f"1. ✅ Fixed ui/windows/dashboard.py:")
    print(f"   - Added missing 'postprocess_type' in fallback logic")
    print(f"   - Added all postprocessing properties")
    print(f"   - Lines: 1203-1213")
    
    print(f"\n2. ✅ Enhanced core/functions/Multidongle.py:")
    print(f"   - Improved YOLOv5 postprocessing implementation")
    print(f"   - Added proper NMS (Non-Maximum Suppression)")
    print(f"   - Enhanced live view display with corner markers")
    print(f"   - Better result string generation")
    
    print(f"\n3. ✅ Fixed core/functions/mflow_converter.py:")
    print(f"   - Added connection mapping for postprocessing nodes")
    print(f"   - Extract postprocessing config from ExactPostprocessNode")
    print(f"   - Create PostProcessor instances for each stage")
    print(f"   - Attach stage_postprocessor to StageConfig")
    
    print(f"\n4. ✅ Enhanced core/functions/InferencePipeline.py:")
    print(f"   - Apply stage_postprocessor during initialization")
    print(f"   - Set postprocessor options to MultiDongle")
    print(f"   - Debug logging for postprocessor application")
    
    print(f"\n5. ✅ Updated .mflow configurations:")
    print(f"   - multi_series_example.mflow: enable_postprocessing = true")
    print(f"   - multi_series_yolov5_fixed.mflow: Complete YOLOv5 setup")
    print(f"   - Proper node connections: Input → Model → Postprocess → Output")
    
    print(f"\n🎯 Expected Results:")
    print(f"   ❌ 'No Fire (Prob: -0.39)' → ✅ 'person detected (Conf: 0.85)'")
    print(f"   ❌ Negative probabilities → ✅ Positive probabilities (0.0-1.0)")
    print(f"   ❌ Fire detection output → ✅ COCO object detection")
    print(f"   ❌ No bounding boxes → ✅ Enhanced bounding boxes in live view")
    print(f"   ❌ Simple terminal output → ✅ Detailed object statistics")
    
    print(f"\n🚀 How the Fix Works:")
    print(f"   1. UI loads .mflow file with yolo_v5 postprocess_type")
    print(f"   2. dashboard.py now includes postprocess_type in properties")
    print(f"   3. mflow_converter.py extracts postprocessing config")
    print(f"   4. Creates PostProcessor with YOLOv5 options")
    print(f"   5. InferencePipeline applies postprocessor to MultiDongle")
    print(f"   6. MultiDongle processes with correct YOLOv5 settings")
    print(f"   7. Enhanced live view shows proper object detection")

def main():
    print("Final Fix Verification Test")
    print("=" * 60)
    
    # Run tests
    converter_ok = test_converter_with_postprocessing()
    multidongle_ok = test_multidongle_postprocessing()
    
    # Show summary
    show_fix_summary()
    
    if converter_ok and multidongle_ok:
        print(f"\n🎉 ALL TESTS PASSED!")
        print(f"   The YOLOv5 postprocessing fix should now work correctly.")
        print(f"   Run: python main.py")
        print(f"   Load: multi_series_yolov5_fixed.mflow")
        print(f"   Deploy and check for positive probabilities!")
    else:
        print(f"\n❌ Some tests failed. Please check the output above.")

if __name__ == "__main__":
    main()