diff --git a/backend/script_groups/ObtainIOFromProjectTia/x4.py b/backend/script_groups/ObtainIOFromProjectTia/x4.py index cd3a0be..a6cfcb4 100644 --- a/backend/script_groups/ObtainIOFromProjectTia/x4.py +++ b/backend/script_groups/ObtainIOFromProjectTia/x4.py @@ -19,20 +19,19 @@ from backend.script_utils import load_configuration # --- Configuration --- # Supported TIA Portal versions mapping (extension -> version) -SUPPORTED_TIA_VERSIONS = { - ".ap18": "18.0", - ".ap19": "19.0", - ".ap20": "20.0" -} +SUPPORTED_TIA_VERSIONS = {".ap18": "18.0", ".ap19": "19.0", ".ap20": "20.0"} # Filter for cross-references. Based on documentation: # 1: 'AllObjects', 2: 'ObjectsWithReferences', 3: 'ObjectsWithoutReferences', 4: 'UnusedObjects' # Using 1 to export all. 0 might also work as a default in some API versions. CROSS_REF_FILTER = 1 -MAX_REOPEN_ATTEMPTS = 5 # Número máximo de re-aperturas permitidas para evitar bucles infinitos +MAX_REOPEN_ATTEMPTS = ( + 5 # Número máximo de re-aperturas permitidas para evitar bucles infinitos +) BLOCK_TIMEOUT_SECONDS = 120 # Referencia de tiempo esperado para el procesamiento de cada bloque (para logging) + class PortalDisposedException(Exception): """Excepción lanzada cuando TIA Portal se ha cerrado inesperadamente o un objeto ha sido descartado.""" @@ -53,6 +52,7 @@ def _is_disposed_exception(exc: Exception) -> bool: ) ) + # --- TIA Scripting Import Handling --- if os.getenv("TIA_SCRIPTING"): sys.path.append(os.getenv("TIA_SCRIPTING")) @@ -82,41 +82,48 @@ except Exception as e: # --- Functions --- + def get_supported_filetypes(): """Returns the supported file types for TIA Portal projects.""" filetypes = [] for ext, version in SUPPORTED_TIA_VERSIONS.items(): - version_major = version.split('.')[0] + version_major = version.split(".")[0] filetypes.append((f"TIA Portal V{version_major} Projects", f"*{ext}")) - + # Add option to show all supported files all_extensions = " ".join([f"*{ext}" for ext in SUPPORTED_TIA_VERSIONS.keys()]) filetypes.insert(0, ("All TIA Portal Projects", all_extensions)) - + return filetypes + def detect_tia_version(project_file_path): """Detects TIA Portal version based on file extension.""" file_path = Path(project_file_path) file_extension = file_path.suffix.lower() - + if file_extension in SUPPORTED_TIA_VERSIONS: detected_version = SUPPORTED_TIA_VERSIONS[file_extension] - print(f"Versión de TIA Portal detectada: {detected_version} (de la extensión {file_extension})") + print( + f"Versión de TIA Portal detectada: {detected_version} (de la extensión {file_extension})" + ) return detected_version else: - print(f"ADVERTENCIA: Extensión de archivo no reconocida '{file_extension}'. Extensiones soportadas: {list(SUPPORTED_TIA_VERSIONS.keys())}") + print( + f"ADVERTENCIA: Extensión de archivo no reconocida '{file_extension}'. Extensiones soportadas: {list(SUPPORTED_TIA_VERSIONS.keys())}" + ) # Default to version 18.0 for backward compatibility print("Usando por defecto TIA Portal V18.0") return "18.0" + def select_project_file(): """Opens a dialog to select a TIA Portal project file.""" root = tk.Tk() root.withdraw() file_path = filedialog.askopenfilename( title="Seleccionar archivo de proyecto TIA Portal", - filetypes=get_supported_filetypes() + filetypes=get_supported_filetypes(), ) root.destroy() if not file_path: @@ -124,43 +131,52 @@ def select_project_file(): sys.exit(0) return file_path + def _normalize_name(name: str) -> str: """Normaliza un nombre quitando espacios laterales y convirtiendo a minúsculas.""" return name.strip().lower() -def _export_block_with_timeout(block, blocks_cr_path, block_name, timeout_seconds=BLOCK_TIMEOUT_SECONDS): + +def _export_block_with_timeout( + block, blocks_cr_path, block_name, timeout_seconds=BLOCK_TIMEOUT_SECONDS +): """ Exporta las referencias cruzadas de un bloque con monitoreo de tiempo. - - Note: TIA Portal Openness no permite operaciones multi-hilo, por lo que + + Note: TIA Portal Openness no permite operaciones multi-hilo, por lo que implementamos un timeout conceptual que al menos registra cuánto tiempo toma. - + Returns: bool: True si se exportó exitosamente """ start_time = time.time() - + try: # Realizar la exportación de forma directa (sin hilos debido a restricciones de TIA) block.export_cross_references( target_directorypath=str(blocks_cr_path), filter=CROSS_REF_FILTER, ) - + elapsed_time = time.time() - start_time - + # Verificar si excedió el tiempo esperado (aunque ya terminó) if elapsed_time > timeout_seconds: - print(f" ADVERTENCIA: El bloque tardó {elapsed_time:.2f}s (>{timeout_seconds}s esperado)") - + print( + f" ADVERTENCIA: El bloque tardó {elapsed_time:.2f}s (>{timeout_seconds}s esperado)" + ) + return True - + except Exception as e: elapsed_time = time.time() - start_time print(f" Tiempo transcurrido antes del error: {elapsed_time:.2f} segundos") raise e -def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, problematic_blocks=None): + +def export_plc_cross_references( + plc, export_base_dir, exported_blocks=None, problematic_blocks=None +): """Exports cross-references for various elements from a given PLC. Parámetros ---------- @@ -185,7 +201,9 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob blocks_cr_exported = 0 blocks_cr_skipped = 0 current_block_name = None # Track current block being processed - print(f"\n[PLC: {plc_name}] Exportando referencias cruzadas de bloques de programa...") + print( + f"\n[PLC: {plc_name}] Exportando referencias cruzadas de bloques de programa..." + ) blocks_cr_path = plc_export_dir / "ProgramBlocks_CR" blocks_cr_path.mkdir(exist_ok=True) print(f" Destino: {blocks_cr_path}") @@ -193,7 +211,7 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob try: program_blocks = plc.get_program_blocks() print(f" Se encontraron {len(program_blocks)} bloques de programa.") - + # Show which blocks will be skipped from the start if problematic_blocks: skipped_names = [] @@ -201,14 +219,18 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob if _normalize_name(block.get_name()) in problematic_blocks: skipped_names.append(block.get_name()) if skipped_names: - print(f" Bloques que serán omitidos (problemáticos previos): {', '.join(skipped_names)}") - + print( + f" Bloques que serán omitidos (problemáticos previos): {', '.join(skipped_names)}" + ) + for block in program_blocks: block_name = block.get_name() current_block_name = block_name # Update current block being processed norm_block = _normalize_name(block_name) if norm_block in problematic_blocks: - print(f" Omitiendo bloque problemático previamente detectado: {block_name}") + print( + f" Omitiendo bloque problemático previamente detectado: {block_name}" + ) blocks_cr_skipped += 1 continue if norm_block in exported_blocks: @@ -219,10 +241,10 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob try: print(f" Exportando referencias cruzadas para {block_name}...") start_time = time.time() - + # Usar la función con monitoreo de tiempo _export_block_with_timeout(block, blocks_cr_path, block_name) - + elapsed_time = time.time() - start_time print(f" Exportación completada en {elapsed_time:.2f} segundos") blocks_cr_exported += 1 @@ -251,7 +273,9 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob " Error de atributo: No se pudo encontrar 'get_program_blocks' en el objeto PLC. Omitiendo bloques de programa." ) except Exception as e: - print(f" ERROR al acceder a los bloques de programa para exportar referencias cruzadas: {e}") + print( + f" ERROR al acceder a los bloques de programa para exportar referencias cruzadas: {e}" + ) traceback.print_exc() # If we know which block was being processed, mark it as problematic if current_block_name: @@ -263,7 +287,9 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob # --- Export PLC Tag Table Cross-References --- tags_cr_exported = 0 tags_cr_skipped = 0 - print(f"\n[PLC: {plc_name}] Exportando referencias cruzadas de tablas de variables...") + print( + f"\n[PLC: {plc_name}] Exportando referencias cruzadas de tablas de variables..." + ) tags_cr_path = plc_export_dir / "PlcTags_CR" tags_cr_path.mkdir(exist_ok=True) print(f" Destino: {tags_cr_path}") @@ -277,8 +303,7 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob try: print(f" Exportando referencias cruzadas para {table_name}...") table.export_cross_references( - target_directorypath=str(tags_cr_path), - filter=CROSS_REF_FILTER + target_directorypath=str(tags_cr_path), filter=CROSS_REF_FILTER ) tags_cr_exported += 1 except RuntimeError as table_ex: @@ -300,13 +325,17 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob " Error de atributo: No se pudo encontrar 'get_plc_tag_tables' en el objeto PLC. Omitiendo tablas de variables." ) except Exception as e: - print(f" ERROR al acceder a las tablas de variables para exportar referencias cruzadas: {e}") + print( + f" ERROR al acceder a las tablas de variables para exportar referencias cruzadas: {e}" + ) traceback.print_exc() # --- Export PLC Data Type (UDT) Cross-References --- udts_cr_exported = 0 udts_cr_skipped = 0 - print(f"\n[PLC: {plc_name}] Exportando referencias cruzadas de tipos de datos PLC (UDTs)...") + print( + f"\n[PLC: {plc_name}] Exportando referencias cruzadas de tipos de datos PLC (UDTs)..." + ) udts_cr_path = plc_export_dir / "PlcDataTypes_CR" udts_cr_path.mkdir(exist_ok=True) print(f" Destino: {udts_cr_path}") @@ -320,8 +349,7 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob try: print(f" Exportando referencias cruzadas para {udt_name}...") udt.export_cross_references( - target_directorypath=str(udts_cr_path), - filter=CROSS_REF_FILTER + target_directorypath=str(udts_cr_path), filter=CROSS_REF_FILTER ) udts_cr_exported += 1 except RuntimeError as udt_ex: @@ -349,7 +377,9 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob # --- Export System Block Cross-References --- sys_blocks_cr_exported = 0 sys_blocks_cr_skipped = 0 - print(f"\n[PLC: {plc_name}] Intentando exportar referencias cruzadas de bloques de sistema...") + print( + f"\n[PLC: {plc_name}] Intentando exportar referencias cruzadas de bloques de sistema..." + ) sys_blocks_cr_path = plc_export_dir / "SystemBlocks_CR" sys_blocks_cr_path.mkdir(exist_ok=True) print(f" Destino: {sys_blocks_cr_path}") @@ -357,14 +387,14 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob try: if hasattr(plc, "get_system_blocks"): system_blocks = plc.get_system_blocks() - print( - f" Se encontraron {len(system_blocks)} bloques de sistema." - ) + print(f" Se encontraron {len(system_blocks)} bloques de sistema.") for sys_block in system_blocks: sys_block_name = sys_block.get_name() print(f" Procesando bloque de sistema: {sys_block_name}...") try: - print(f" Exportando referencias cruzadas para {sys_block_name}...") + print( + f" Exportando referencias cruzadas para {sys_block_name}..." + ) sys_block.export_cross_references( target_directorypath=str(sys_blocks_cr_path), filter=CROSS_REF_FILTER, @@ -403,7 +433,9 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob # --- Export Software Unit Cross-References --- sw_units_cr_exported = 0 sw_units_cr_skipped = 0 - print(f"\n[PLC: {plc_name}] Intentando exportar referencias cruzadas de unidades de software...") + print( + f"\n[PLC: {plc_name}] Intentando exportar referencias cruzadas de unidades de software..." + ) sw_units_cr_path = plc_export_dir / "SoftwareUnits_CR" sw_units_cr_path.mkdir(exist_ok=True) print(f" Destino: {sw_units_cr_path}") @@ -453,6 +485,7 @@ def export_plc_cross_references(plc, export_base_dir, exported_blocks=None, prob print(f"\n--- Finalizado el procesamiento del PLC: {plc_name} ---") + def open_portal_and_project(tia_version: str, project_file_path: str): """Abre TIA Portal y el proyecto indicado, devolviendo el portal y el objeto proyecto.""" print(f"\nConectando a TIA Portal V{tia_version}...") @@ -467,9 +500,12 @@ def open_portal_and_project(tia_version: str, project_file_path: str): if project_obj is None: project_obj = portal.get_project() if project_obj is None: - raise Exception("No se pudo abrir u obtener el proyecto especificado tras la reapertura.") + raise Exception( + "No se pudo abrir u obtener el proyecto especificado tras la reapertura." + ) return portal, project_obj + # --- Main Script --- if __name__ == "__main__": @@ -478,7 +514,9 @@ if __name__ == "__main__": print("--- Exportador de Referencias Cruzadas de TIA Portal ---") print(f"Configuración:") - print(f" - Tiempo esperado por bloque: {BLOCK_TIMEOUT_SECONDS} segundos (para logging)") + print( + f" - Tiempo esperado por bloque: {BLOCK_TIMEOUT_SECONDS} segundos (para logging)" + ) print(f" - Máximo intentos de reapertura: {MAX_REOPEN_ATTEMPTS}") print(f" - Filtro de referencias cruzadas: {CROSS_REF_FILTER}") print("") @@ -486,7 +524,9 @@ if __name__ == "__main__": # Validate working directory if not working_directory or not os.path.isdir(working_directory): print("ERROR: Directorio de trabajo no configurado o inválido.") - print("Por favor configure el directorio de trabajo usando la aplicación principal.") + print( + "Por favor configure el directorio de trabajo usando la aplicación principal." + ) sys.exit(1) # 1. Select Project File @@ -502,7 +542,9 @@ if __name__ == "__main__": print(f"\nProyecto seleccionado: {project_file}") print(f"Usando directorio base de exportación: {export_base_dir.resolve()}") except Exception as e: - print(f"ERROR: No se pudo crear el directorio de exportación '{export_base_dir}'. Error: {e}") + print( + f"ERROR: No se pudo crear el directorio de exportación '{export_base_dir}'. Error: {e}" + ) sys.exit(1) portal_instance = None @@ -510,7 +552,9 @@ if __name__ == "__main__": try: # 4. Connect to TIA Portal with detected version - portal_instance, project_object = open_portal_and_project(tia_version, project_file) + portal_instance, project_object = open_portal_and_project( + tia_version, project_file + ) # 5. Get PLCs plcs = project_object.get_plcs() @@ -547,7 +591,9 @@ if __name__ == "__main__": skipped_blocks_report.append(failed_block) print(f"Marcando bloque problemático: {failed_block}") else: - print("Error general detectado sin bloque específico identificado") + print( + "Error general detectado sin bloque específico identificado" + ) if reopen_attempts > MAX_REOPEN_ATTEMPTS: print( @@ -563,8 +609,12 @@ if __name__ == "__main__": pass # Re-abrir portal y proyecto - print(f"Re-abriendo TIA Portal (intento {reopen_attempts}/{MAX_REOPEN_ATTEMPTS})...") - portal_instance, project_object = open_portal_and_project(tia_version, project_file) + print( + f"Re-abriendo TIA Portal (intento {reopen_attempts}/{MAX_REOPEN_ATTEMPTS})..." + ) + portal_instance, project_object = open_portal_and_project( + tia_version, project_file + ) # Buscar de nuevo el PLC por nombre plc_device = None @@ -581,8 +631,12 @@ if __name__ == "__main__": continue if skipped_blocks_report: - print(f"\nBloques problemáticos para el PLC '{plc_name}': {', '.join(set(skipped_blocks_report))}") - print(f"Total de bloques problemáticos registrados: {len(problematic_blocks)}") + print( + f"\nBloques problemáticos para el PLC '{plc_name}': {', '.join(set(skipped_blocks_report))}" + ) + print( + f"Total de bloques problemáticos registrados: {len(problematic_blocks)}" + ) print("\nProceso de exportación de referencias cruzadas completado.")