460 lines
19 KiB
Python
460 lines
19 KiB
Python
import asyncio
|
|
import os
|
|
import time
|
|
import traceback
|
|
from typing import List, Tuple, Dict, Any
|
|
from logix_designer_sdk import LogixProject, StdOutEventLogger
|
|
from logix_designer_sdk.exceptions import (
|
|
LogixSdkError,
|
|
OperationFailedError,
|
|
OperationNotPerformedError,
|
|
LoggerFailedError,
|
|
EventLoggerRuntimeError
|
|
)
|
|
|
|
# Configuration
|
|
TARGET_REV = 36
|
|
|
|
class DetailedEventLogger:
|
|
"""Enhanced event logger that captures all SDK messages, errors, and progress"""
|
|
|
|
def __init__(self, filename: str):
|
|
self.filename = filename
|
|
self.start_time = time.time()
|
|
self.messages = []
|
|
self.errors = []
|
|
self.warnings = []
|
|
self.progress_messages = []
|
|
self.sdk_messages = [] # Store messages from StdOutEventLogger
|
|
|
|
def log(self, message: str, level: str = "INFO"):
|
|
"""Log a message with detailed categorization"""
|
|
current_time = time.time()
|
|
elapsed = current_time - self.start_time
|
|
|
|
# Categorize the message
|
|
message_lower = message.lower()
|
|
if any(keyword in message_lower for keyword in ['error', 'failed', 'exception']):
|
|
self.errors.append(message)
|
|
level = "ERROR"
|
|
elif any(keyword in message_lower for keyword in ['warning', 'warn']):
|
|
self.warnings.append(message)
|
|
level = "WARN"
|
|
elif any(keyword in message_lower for keyword in ['progress', 'converting', 'saving', 'loading']):
|
|
self.progress_messages.append(message)
|
|
level = "PROGRESS"
|
|
|
|
self.messages.append({
|
|
'timestamp': elapsed,
|
|
'level': level,
|
|
'message': message
|
|
})
|
|
|
|
# Format and print the message
|
|
timestamp_str = f"[{elapsed:>6.1f}s]"
|
|
level_str = f"[{level:<8}]"
|
|
print(f"[LOG] {timestamp_str} {level_str} {self.filename} | {message}")
|
|
|
|
def capture_sdk_message(self, message: str):
|
|
"""Capture messages from the SDK's StdOutEventLogger"""
|
|
self.sdk_messages.append(message)
|
|
# Also categorize SDK messages
|
|
message_lower = message.lower()
|
|
if any(keyword in message_lower for keyword in ['error', 'failed', 'exception']):
|
|
self.errors.append(f"SDK: {message}")
|
|
elif any(keyword in message_lower for keyword in ['warning', 'warn']):
|
|
self.warnings.append(f"SDK: {message}")
|
|
elif any(keyword in message_lower for keyword in ['convert', 'save', 'load', 'build']):
|
|
self.progress_messages.append(f"SDK: {message}")
|
|
|
|
def get_summary(self) -> Dict[str, Any]:
|
|
"""Get a summary of all captured messages"""
|
|
return {
|
|
'total_messages': len(self.messages),
|
|
'sdk_messages': len(self.sdk_messages),
|
|
'errors': len(self.errors),
|
|
'warnings': len(self.warnings),
|
|
'progress_updates': len(self.progress_messages),
|
|
'duration': time.time() - self.start_time
|
|
}
|
|
|
|
class CustomStdOutEventLogger(StdOutEventLogger):
|
|
"""Custom wrapper around StdOutEventLogger to capture messages"""
|
|
|
|
def __init__(self, capture_logger: DetailedEventLogger):
|
|
super().__init__()
|
|
self.capture_logger = capture_logger
|
|
|
|
def log(self, message):
|
|
# Capture the message for our detailed logger
|
|
self.capture_logger.capture_sdk_message(message)
|
|
# Still output to stdout via parent class
|
|
super().log(message)
|
|
|
|
def categorize_exception(exc: Exception) -> Dict[str, Any]:
|
|
"""Categorize exceptions based on SDK documentation patterns"""
|
|
|
|
exc_info = {
|
|
'type': type(exc).__name__,
|
|
'message': str(exc),
|
|
'category': 'Unknown',
|
|
'severity': 'High',
|
|
'suggested_action': 'Check error details and retry',
|
|
'is_recoverable': False
|
|
}
|
|
|
|
if isinstance(exc, OperationFailedError):
|
|
exc_info.update({
|
|
'category': 'Operation Failed',
|
|
'severity': 'High',
|
|
'suggested_action': 'Check input parameters, file permissions, and target revision compatibility',
|
|
'is_recoverable': True
|
|
})
|
|
elif isinstance(exc, OperationNotPerformedError):
|
|
exc_info.update({
|
|
'category': 'Operation Not Performed',
|
|
'severity': 'Critical',
|
|
'suggested_action': 'Check SDK server connection and project state',
|
|
'is_recoverable': False
|
|
})
|
|
elif isinstance(exc, LoggerFailedError):
|
|
exc_info.update({
|
|
'category': 'Logger Failed',
|
|
'severity': 'Medium',
|
|
'suggested_action': 'Check logging permissions and disk space',
|
|
'is_recoverable': True
|
|
})
|
|
elif isinstance(exc, EventLoggerRuntimeError):
|
|
exc_info.update({
|
|
'category': 'Event Logger Runtime Error',
|
|
'severity': 'Medium',
|
|
'suggested_action': 'Check event logging configuration',
|
|
'is_recoverable': True
|
|
})
|
|
elif isinstance(exc, LogixSdkError):
|
|
exc_info.update({
|
|
'category': 'SDK Error',
|
|
'severity': 'High',
|
|
'suggested_action': 'Check SDK installation and project file integrity',
|
|
'is_recoverable': True
|
|
})
|
|
elif isinstance(exc, FileNotFoundError):
|
|
exc_info.update({
|
|
'category': 'File Not Found',
|
|
'severity': 'High',
|
|
'suggested_action': 'Verify input file path exists and is accessible',
|
|
'is_recoverable': False
|
|
})
|
|
elif isinstance(exc, PermissionError):
|
|
exc_info.update({
|
|
'category': 'Permission Error',
|
|
'severity': 'High',
|
|
'suggested_action': 'Check file permissions and run as administrator if needed',
|
|
'is_recoverable': True
|
|
})
|
|
elif isinstance(exc, TypeError):
|
|
exc_info.update({
|
|
'category': 'Type Error',
|
|
'severity': 'High',
|
|
'suggested_action': 'Check parameter types and SDK API usage',
|
|
'is_recoverable': True
|
|
})
|
|
|
|
return exc_info
|
|
|
|
async def convert_with_comprehensive_error_handling(input_file: str, output_file: str) -> Dict[str, Any]:
|
|
"""Convert L5X to ACD with comprehensive error handling and logging"""
|
|
|
|
filename = os.path.basename(input_file)
|
|
start_time = time.time()
|
|
|
|
print(f"\n[START] Starting conversion: {filename}")
|
|
print("=" * 80)
|
|
|
|
# Create enhanced event logger
|
|
event_logger = DetailedEventLogger(filename)
|
|
|
|
# Initialize stop_heartbeat early to avoid UnboundLocalError
|
|
stop_heartbeat = asyncio.Event()
|
|
|
|
try:
|
|
# Validate input file first
|
|
if not os.path.exists(input_file):
|
|
raise FileNotFoundError(f"Input file not found: {input_file}")
|
|
|
|
if not os.access(input_file, os.R_OK):
|
|
raise PermissionError(f"Cannot read input file: {input_file}")
|
|
|
|
# Check file size
|
|
file_size_mb = os.path.getsize(input_file) / (1024*1024)
|
|
if file_size_mb > 500: # SDK limit
|
|
event_logger.log(f"WARNING: Large file ({file_size_mb:.1f} MB) - may take longer to process", "WARN")
|
|
|
|
event_logger.log(f"Starting conversion of {input_file} (size: {file_size_mb:.2f} MB)")
|
|
event_logger.log(f"Target revision: {TARGET_REV}")
|
|
|
|
async def heartbeat():
|
|
"""Print elapsed time every 2 s until stop_heartbeat is set"""
|
|
while not stop_heartbeat.is_set():
|
|
await asyncio.sleep(2)
|
|
elapsed_hb = time.time() - start_time
|
|
print(f"Elapsed: {elapsed_hb:.1f}s")
|
|
|
|
hb_task = asyncio.create_task(heartbeat())
|
|
|
|
# Create custom event logger that captures SDK messages
|
|
custom_sdk_logger = CustomStdOutEventLogger(event_logger)
|
|
|
|
# Convert with comprehensive event logging
|
|
proj = await LogixProject.convert(input_file, TARGET_REV, custom_sdk_logger)
|
|
|
|
# Stop heartbeat once operations finished
|
|
stop_heartbeat.set()
|
|
await hb_task
|
|
|
|
event_logger.log("[SUCCESS] Conversion completed successfully")
|
|
event_logger.log(f"[SAVING] Saving to {output_file}")
|
|
|
|
# Validate output directory
|
|
output_dir = os.path.dirname(output_file)
|
|
if output_dir and not os.path.exists(output_dir):
|
|
os.makedirs(output_dir)
|
|
event_logger.log(f"Created output directory: {output_dir}")
|
|
|
|
# Save the converted project
|
|
await proj.save_as(output_file, True)
|
|
|
|
# Verify output file was created
|
|
if not os.path.exists(output_file):
|
|
raise OperationFailedError("Output file was not created successfully")
|
|
|
|
# Calculate final results
|
|
elapsed_time = time.time() - start_time
|
|
output_size_mb = os.path.getsize(output_file) / (1024*1024)
|
|
|
|
event_logger.log(f"[SUCCESS] File saved successfully")
|
|
event_logger.log(f"[INFO] Output size: {output_size_mb:.2f} MB")
|
|
event_logger.log(f"[INFO] Total time: {elapsed_time:.1f}s")
|
|
|
|
logger_summary = event_logger.get_summary()
|
|
|
|
print(f"\n[SUCCESS]: {filename}")
|
|
print(f" Input: {file_size_mb:.2f} MB")
|
|
print(f" Output: {output_size_mb:.2f} MB")
|
|
print(f" Duration: {elapsed_time:.1f}s")
|
|
print(f" Messages: {logger_summary['total_messages']} app + {logger_summary['sdk_messages']} SDK")
|
|
if logger_summary['warnings'] > 0:
|
|
print(f" Warnings: {logger_summary['warnings']}")
|
|
print("=" * 80)
|
|
|
|
return {
|
|
'status': 'success',
|
|
'input': input_file,
|
|
'output': output_file,
|
|
'input_size_mb': round(file_size_mb, 2),
|
|
'output_size_mb': round(output_size_mb, 2),
|
|
'duration_seconds': round(elapsed_time, 1),
|
|
'messages_captured': logger_summary['total_messages'],
|
|
'sdk_messages': logger_summary['sdk_messages'],
|
|
'warnings': logger_summary['warnings'],
|
|
'errors_logged': logger_summary['errors']
|
|
}
|
|
|
|
except Exception as e:
|
|
# Stop heartbeat so it doesn't continue after error
|
|
stop_heartbeat.set()
|
|
if 'hb_task' in locals():
|
|
try:
|
|
await hb_task
|
|
except Exception:
|
|
pass
|
|
|
|
elapsed_time = time.time() - start_time
|
|
|
|
# Categorize and analyze the exception
|
|
exc_info = categorize_exception(e)
|
|
logger_summary = event_logger.get_summary()
|
|
|
|
event_logger.log(f"FAILED: Conversion failed: {exc_info['message']}", "ERROR")
|
|
|
|
print(f"\nFAILED: {filename}")
|
|
print(f" Error Type: {exc_info['type']} ({exc_info['category']})")
|
|
print(f" Message: {exc_info['message']}")
|
|
print(f" Severity: {exc_info['severity']}")
|
|
print(f" Suggested Action: {exc_info['suggested_action']}")
|
|
print(f" Recoverable: {'Yes' if exc_info['is_recoverable'] else 'No'}")
|
|
print(f" Failed after: {elapsed_time:.1f}s")
|
|
print(f" Messages captured: {logger_summary['total_messages']} app + {logger_summary['sdk_messages']} SDK")
|
|
|
|
# Print detailed stack trace for debugging
|
|
if logger_summary['errors'] > 0:
|
|
print(f" Errors logged: {logger_summary['errors']}")
|
|
print(" Recent error messages:")
|
|
for error_msg in event_logger.errors[-3:]: # Show last 3 errors
|
|
print(f" • {error_msg}")
|
|
|
|
print("\nFull Stack Trace:")
|
|
print(traceback.format_exc())
|
|
print("=" * 80)
|
|
|
|
return {
|
|
'status': 'failed',
|
|
'input': input_file,
|
|
'output': output_file,
|
|
'error': exc_info['message'],
|
|
'error_type': exc_info['type'],
|
|
'error_category': exc_info['category'],
|
|
'severity': exc_info['severity'],
|
|
'suggested_action': exc_info['suggested_action'],
|
|
'is_recoverable': exc_info['is_recoverable'],
|
|
'duration_seconds': round(elapsed_time, 1),
|
|
'messages_captured': logger_summary['total_messages'],
|
|
'sdk_messages': logger_summary['sdk_messages'],
|
|
'errors_logged': logger_summary['errors'],
|
|
'stack_trace': traceback.format_exc()
|
|
}
|
|
|
|
async def convert_multiple_files_with_error_recovery(file_pairs: List[Tuple[str, str]]) -> List[Dict[str, Any]]:
|
|
"""Convert multiple L5X files with error recovery and detailed reporting"""
|
|
|
|
if not file_pairs:
|
|
print("ERROR: No files to convert")
|
|
return []
|
|
|
|
print(f"Converting {len(file_pairs)} file(s) to ACD format")
|
|
print(f"Target Logix revision: {TARGET_REV}")
|
|
print(f"Using Logix Designer SDK with comprehensive error handling")
|
|
print(f"Error recovery and detailed logging enabled")
|
|
|
|
results = []
|
|
|
|
for i, (input_file, output_file) in enumerate(file_pairs, 1):
|
|
print(f"\nProcessing file {i}/{len(file_pairs)}")
|
|
|
|
# Convert the file with comprehensive error handling
|
|
result = await convert_with_comprehensive_error_handling(input_file, output_file)
|
|
results.append(result)
|
|
|
|
# Add recovery suggestions for failed files
|
|
if result['status'] == 'failed' and result.get('is_recoverable', False):
|
|
print(f"Recovery suggestion: {result['suggested_action']}")
|
|
|
|
# Print comprehensive final summary
|
|
print_comprehensive_summary(results)
|
|
|
|
return results
|
|
|
|
def print_comprehensive_summary(results: List[Dict[str, Any]]):
|
|
"""Print a comprehensive summary with error analysis"""
|
|
|
|
successful = [r for r in results if r['status'] == 'success']
|
|
failed = [r for r in results if r['status'] == 'failed']
|
|
|
|
total_time = sum(r.get('duration_seconds', 0) for r in results)
|
|
total_input_size = sum(r.get('input_size_mb', 0) for r in successful)
|
|
total_output_size = sum(r.get('output_size_mb', 0) for r in successful)
|
|
total_messages = sum(r.get('messages_captured', 0) for r in results)
|
|
total_sdk_messages = sum(r.get('sdk_messages', 0) for r in results)
|
|
total_warnings = sum(r.get('warnings', 0) for r in results)
|
|
|
|
print(f"\n{'COMPREHENSIVE CONVERSION SUMMARY':^80}")
|
|
print("=" * 80)
|
|
print(f"Total files processed: {len(results)}")
|
|
print(f"Successfully converted: {len(successful)}")
|
|
print(f"Failed conversions: {len(failed)}")
|
|
print(f"Total processing time: {total_time:.1f}s")
|
|
print(f"Total messages captured: {total_messages} app + {total_sdk_messages} SDK")
|
|
|
|
if total_warnings > 0:
|
|
print(f"Total warnings: {total_warnings}")
|
|
|
|
if successful:
|
|
print(f"Total input size: {total_input_size:.2f} MB")
|
|
print(f"Total output size: {total_output_size:.2f} MB")
|
|
avg_time = total_time / len(successful) if successful else 0
|
|
print(f"Average time per file: {avg_time:.1f}s")
|
|
|
|
if total_input_size > 0:
|
|
compression_ratio = (total_output_size / total_input_size) * 100
|
|
print(f"Size ratio: {compression_ratio:.1f}% (output/input)")
|
|
|
|
if failed:
|
|
print(f"\nFailed Files Analysis:")
|
|
|
|
# Group failures by category
|
|
failure_categories = {}
|
|
for result in failed:
|
|
category = result.get('error_category', 'Unknown')
|
|
if category not in failure_categories:
|
|
failure_categories[category] = []
|
|
failure_categories[category].append(result)
|
|
|
|
for category, category_failures in failure_categories.items():
|
|
print(f"\n {category} ({len(category_failures)} files):")
|
|
for result in category_failures:
|
|
recovery_status = "Recoverable" if result.get('is_recoverable', False) else "Not recoverable"
|
|
print(f" FAILED: {os.path.basename(result['input'])}")
|
|
print(f" Error: {result.get('error', 'Unknown error')}")
|
|
print(f" Status: {recovery_status}")
|
|
|
|
print("=" * 80)
|
|
|
|
async def main():
|
|
"""Main execution function with comprehensive error handling"""
|
|
|
|
print("Logix Designer SDK L5X to ACD Converter")
|
|
print("Enhanced with comprehensive error handling and progress tracking")
|
|
print("-" * 80)
|
|
|
|
# Check if command-line arguments were provided
|
|
import sys
|
|
if len(sys.argv) > 1:
|
|
# Use command-line argument as input file
|
|
input_file = sys.argv[1]
|
|
output_file = input_file.replace('.L5X', '.ACD').replace('.l5x', '.ACD')
|
|
file_pairs = [(input_file, output_file)]
|
|
else:
|
|
# Define files to convert - add more files here as needed
|
|
file_pairs = [
|
|
(r"MTN6_MCM01_UL1_UL3.L5X", r"MTN6_MCM01_UL1_UL3.ACD"),
|
|
# Add more file pairs here like:
|
|
# (r"Project2.L5X", r"Project2.ACD"),
|
|
# (r"Project3.L5X", r"Project3.ACD"),
|
|
]
|
|
|
|
if not file_pairs:
|
|
print("ERROR: No files defined for conversion!")
|
|
print("Edit the file_pairs list in main() to add files to convert.")
|
|
return 1
|
|
|
|
try:
|
|
# Execute conversions with comprehensive error handling
|
|
results = await convert_multiple_files_with_error_recovery(file_pairs)
|
|
|
|
# Determine exit code based on results
|
|
failed_count = len([r for r in results if r['status'] == 'failed'])
|
|
critical_failures = len([r for r in results if r['status'] == 'failed' and not r.get('is_recoverable', True)])
|
|
|
|
if critical_failures > 0:
|
|
print(f"\nCRITICAL: {critical_failures} non-recoverable failures detected")
|
|
return 2 # Critical failure exit code
|
|
elif failed_count > 0:
|
|
print(f"\nWARNING: {failed_count} recoverable failures detected")
|
|
return 1 # Warning exit code
|
|
else:
|
|
print(f"\nSUCCESS: All {len(results)} files converted successfully!")
|
|
return 0 # Success exit code
|
|
|
|
except Exception as e:
|
|
print(f"\nFATAL ERROR in main execution:")
|
|
print(f" Type: {type(e).__name__}")
|
|
print(f" Message: {str(e)}")
|
|
print(f"\nFull Stack Trace:")
|
|
print(traceback.format_exc())
|
|
return 3 # Fatal error exit code
|
|
|
|
if __name__ == "__main__":
|
|
exit_code = asyncio.run(main())
|
|
print(f"\nProcess completed with exit code: {exit_code}")
|
|
exit(exit_code)
|