149 lines
No EOL
5 KiB
Python
149 lines
No EOL
5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for parallel layout processing implementation
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
from pathlib import Path
|
|
|
|
# Add current directory to path
|
|
sys.path.insert(0, os.getcwd())
|
|
|
|
def test_parallel_processing():
|
|
"""Test the parallel processing implementation"""
|
|
|
|
print("Testing parallel layout processing implementation...")
|
|
|
|
# Test 1: Import all classes
|
|
try:
|
|
from hybrid_detector import HybridImageDetector, InlierAnalysisCoordinator, ProgressTracker
|
|
print("✓ Successfully imported all classes")
|
|
except ImportError as e:
|
|
print(f"✗ Import error: {e}")
|
|
return False
|
|
|
|
# Test 2: Create InlierAnalysisCoordinator
|
|
try:
|
|
from memory_manager import MemoryManager
|
|
memory_manager = MemoryManager()
|
|
coordinator = InlierAnalysisCoordinator(
|
|
local_workers=2,
|
|
memory_manager=memory_manager,
|
|
min_good_matches=10
|
|
)
|
|
print("✓ Successfully created InlierAnalysisCoordinator")
|
|
except Exception as e:
|
|
print(f"✗ Error creating coordinator: {e}")
|
|
return False
|
|
|
|
# Test 3: Create ProgressTracker
|
|
try:
|
|
tracker = ProgressTracker(total_layouts=100)
|
|
info = tracker.get_progress_info()
|
|
print(f"✓ Successfully created ProgressTracker (total: {info['total']})")
|
|
except Exception as e:
|
|
print(f"✗ Error creating progress tracker: {e}")
|
|
return False
|
|
|
|
# Test 4: Create HybridImageDetector with parallel processing
|
|
try:
|
|
detector = HybridImageDetector(
|
|
panel_threshold=2,
|
|
inlier_threshold=0.65,
|
|
parallel_layouts=True,
|
|
layout_workers=2,
|
|
max_concurrent_layouts=2
|
|
)
|
|
print("✓ Successfully created HybridImageDetector with parallel processing")
|
|
print(f" - Parallel layouts: {detector.parallel_layouts}")
|
|
print(f" - Layout workers: {detector.layout_workers}")
|
|
print(f" - Max concurrent layouts: {detector.max_concurrent_layouts}")
|
|
except Exception as e:
|
|
print(f"✗ Error creating detector: {e}")
|
|
return False
|
|
|
|
# Test 5: Test coordinator start/stop
|
|
try:
|
|
coordinator.start()
|
|
print("✓ Successfully started coordinator")
|
|
|
|
# Test queue size
|
|
queue_size = coordinator.get_queue_size()
|
|
print(f" - Queue size: {queue_size}")
|
|
|
|
coordinator.stop()
|
|
print("✓ Successfully stopped coordinator")
|
|
except Exception as e:
|
|
print(f"✗ Error with coordinator lifecycle: {e}")
|
|
return False
|
|
|
|
# Test 6: Test memory monitoring
|
|
try:
|
|
memory_adjusted = detector._monitor_memory_and_adjust_workers()
|
|
print(f"✓ Memory monitoring executed (adjustments made: {memory_adjusted})")
|
|
except Exception as e:
|
|
print(f"✗ Error with memory monitoring: {e}")
|
|
return False
|
|
|
|
# Test 7: Test error handling
|
|
try:
|
|
error_result = detector._handle_worker_failure("test_layout.jpg", Exception("test error"))
|
|
print(f"✓ Error handling executed (result has error: {'error' in error_result})")
|
|
except Exception as e:
|
|
print(f"✗ Error with error handling: {e}")
|
|
return False
|
|
|
|
print("\n🎉 All tests passed! Parallel processing implementation is working correctly.")
|
|
return True
|
|
|
|
def test_cli_integration():
|
|
"""Test CLI integration"""
|
|
|
|
print("\nTesting CLI integration...")
|
|
|
|
# Test parsing with parallel arguments
|
|
try:
|
|
from cli import parse_arguments
|
|
|
|
# Mock sys.argv for testing
|
|
import sys
|
|
original_argv = sys.argv
|
|
|
|
# Test with parallel processing arguments
|
|
sys.argv = ['cli.py', '--test', '--hybrid', '--parallel-layouts', '--layout-workers', '4']
|
|
|
|
try:
|
|
args = parse_arguments()
|
|
print("✓ Successfully parsed parallel processing arguments")
|
|
print(f" - Parallel layouts: {args.parallel_layouts}")
|
|
print(f" - Layout workers: {args.layout_workers}")
|
|
print(f" - Max concurrent layouts: {args.max_concurrent_layouts}")
|
|
except SystemExit:
|
|
# parse_arguments calls sys.exit() if help is requested
|
|
pass
|
|
finally:
|
|
sys.argv = original_argv
|
|
|
|
except Exception as e:
|
|
print(f"✗ Error testing CLI integration: {e}")
|
|
return False
|
|
|
|
print("✓ CLI integration tests passed!")
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
success = test_parallel_processing()
|
|
|
|
if success:
|
|
success = test_cli_integration()
|
|
|
|
if success:
|
|
print("\n🚀 Implementation is ready for production use!")
|
|
print("\nUsage examples:")
|
|
print(" python cli.py --test --hybrid --parallel-layouts")
|
|
print(" python cli.py --limit 10 --hybrid --parallel-layouts --layout-workers 4")
|
|
print(" python cli.py --all --hybrid --parallel-layouts --layout-workers 6")
|
|
else:
|
|
print("\n❌ Implementation needs fixes before production use.")
|
|
sys.exit(1) |