Masonmason
080eb5b887
Major Features: • Advanced topological sorting algorithm with cycle detection and resolution • Intelligent pipeline optimization with parallelization analysis • Critical path analysis and performance metrics calculation • Comprehensive .mflow file converter for seamless UI-to-API integration • Complete modular UI framework with node-based pipeline editor • Enhanced model node properties (scpu_fw_path, ncpu_fw_path) • Professional output formatting without emoji decorations Technical Improvements: • Graph theory algorithms (DFS, BFS, topological sort) • Automatic dependency resolution and conflict prevention • Multi-criteria pipeline optimization • Real-time stage count calculation and validation • Comprehensive configuration validation and error handling • Modular architecture with clean separation of concerns New Components: • MFlow converter with topology analysis (core/functions/mflow_converter.py) • Complete node system with exact property matching • Pipeline editor with visual node connections • Performance estimation and dongle management panels • Comprehensive test suite and demonstration scripts 🤖 Generated with Claude Code (https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
95 lines
2.4 KiB
Python
95 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify the pipeline editor functionality.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
# Set up Qt environment
|
|
os.environ['QT_QPA_PLATFORM'] = 'offscreen'
|
|
|
|
from PyQt5.QtWidgets import QApplication
|
|
from PyQt5.QtCore import QTimer
|
|
|
|
# Create Qt application
|
|
app = QApplication(sys.argv)
|
|
|
|
# Import after Qt setup
|
|
from ui.windows.pipeline_editor import PipelineEditor
|
|
|
|
|
|
def test_pipeline_editor():
|
|
"""Test the pipeline editor functionality."""
|
|
print("Testing Pipeline Editor...")
|
|
|
|
# Create editor
|
|
editor = PipelineEditor()
|
|
|
|
# Test initial state
|
|
initial_count = editor.get_current_stage_count()
|
|
print(f"Initial stage count: {initial_count}")
|
|
assert initial_count == 0, f"Expected 0 stages initially, got {initial_count}"
|
|
|
|
# Test adding nodes (if NodeGraphQt is available)
|
|
if hasattr(editor, 'node_graph') and editor.node_graph:
|
|
print("NodeGraphQt is available, testing node addition...")
|
|
|
|
# Add input node
|
|
editor.add_input_node()
|
|
|
|
# Add model node
|
|
editor.add_model_node()
|
|
|
|
# Add output node
|
|
editor.add_output_node()
|
|
|
|
# Wait for analysis to complete
|
|
QTimer.singleShot(1000, lambda: check_final_count(editor))
|
|
|
|
# Run event loop briefly
|
|
QTimer.singleShot(1500, app.quit)
|
|
app.exec_()
|
|
|
|
else:
|
|
print("NodeGraphQt not available, skipping node addition tests")
|
|
|
|
print("✓ Pipeline editor test completed")
|
|
|
|
|
|
def check_final_count(editor):
|
|
"""Check final stage count after adding nodes."""
|
|
final_count = editor.get_current_stage_count()
|
|
print(f"Final stage count: {final_count}")
|
|
|
|
if final_count == 1:
|
|
print("✓ Stage count correctly updated to 1")
|
|
else:
|
|
print(f"❌ Expected 1 stage, got {final_count}")
|
|
|
|
# Get pipeline summary
|
|
summary = editor.get_pipeline_summary()
|
|
print(f"Pipeline summary: {summary}")
|
|
|
|
|
|
def main():
|
|
"""Run all tests."""
|
|
print("Running Pipeline Editor Tests...")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
test_pipeline_editor()
|
|
|
|
print("\n" + "=" * 50)
|
|
print("All tests completed! ✓")
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ Test failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main() |