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)