Exportando DBs tambien
This commit is contained in:
parent
5213d09bcf
commit
6c604176a1
|
|
@ -0,0 +1,273 @@
|
|||
# x3_generate_scl.py
|
||||
# -*- coding: utf-8 -*-
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import argparse
|
||||
import sys
|
||||
import traceback
|
||||
|
||||
# --- Importar Utilidades (mantener como estaba) ---
|
||||
try:
|
||||
from processors.processor_utils import format_variable_name
|
||||
SCL_SUFFIX = "_sympy_processed"
|
||||
GROUPED_COMMENT = "// Logic included in grouped IF"
|
||||
except ImportError:
|
||||
print("Advertencia: No se pudo importar 'format_variable_name'. Usando fallback.")
|
||||
def format_variable_name(name): # Fallback BÁSICO
|
||||
if not name: return "_INVALID_NAME_"
|
||||
if name.startswith('"') and name.endswith('"'): return name
|
||||
prefix = "#" if name.startswith("#") else ""
|
||||
if prefix: name = name[1:]
|
||||
if name and name[0].isdigit(): name = "_" + name
|
||||
name = re.sub(r"[^a-zA-Z0-9_]", "_", name)
|
||||
return prefix + name
|
||||
SCL_SUFFIX = "_sympy_processed"
|
||||
GROUPED_COMMENT = "// Logic included in grouped IF"
|
||||
|
||||
# --- NUEVA FUNCIÓN para formatear valores iniciales SCL ---
|
||||
def format_scl_start_value(value, datatype):
|
||||
"""Formatea un valor para la inicialización SCL según el tipo."""
|
||||
if value is None: return None
|
||||
datatype_lower = datatype.lower() if datatype else ""
|
||||
value_str = str(value)
|
||||
|
||||
if "bool" in datatype_lower:
|
||||
return "TRUE" if value_str.lower() == 'true' else "FALSE"
|
||||
elif "string" in datatype_lower:
|
||||
escaped_value = value_str.replace("'", "''")
|
||||
if escaped_value.startswith("'") and escaped_value.endswith("'"): escaped_value = escaped_value[1:-1]
|
||||
return f"'{escaped_value}'"
|
||||
elif "char" in datatype_lower: # Añadido Char
|
||||
escaped_value = value_str.replace("'", "''")
|
||||
if escaped_value.startswith("'") and escaped_value.endswith("'"): escaped_value = escaped_value[1:-1]
|
||||
return f"'{escaped_value}'"
|
||||
elif any(t in datatype_lower for t in ["int", "byte", "word", "dint", "dword", "lint", "lword", "sint", "usint", "uint", "udint", "ulint"]): # Ampliado
|
||||
try: return str(int(value_str))
|
||||
except ValueError:
|
||||
if re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', value_str): return value_str
|
||||
return f"'{value_str}'" # O como string si no es entero ni símbolo
|
||||
elif "real" in datatype_lower or "lreal" in datatype_lower:
|
||||
try:
|
||||
f_val = float(value_str); s_val = str(f_val)
|
||||
if '.' not in s_val and 'e' not in s_val.lower(): s_val += ".0"
|
||||
return s_val
|
||||
except ValueError:
|
||||
if re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', value_str): return value_str
|
||||
return f"'{value_str}'"
|
||||
elif "time" in datatype_lower: # Añadido Time, S5Time, LTime
|
||||
# Quitar T#, LT#, S5T# si existen
|
||||
prefix = ""
|
||||
if value_str.upper().startswith("T#"): prefix="T#"; value_str = value_str[2:]
|
||||
elif value_str.upper().startswith("LT#"): prefix="LT#"; value_str = value_str[3:]
|
||||
elif value_str.upper().startswith("S5T#"): prefix="S5T#"; value_str = value_str[4:]
|
||||
# Devolver con el prefijo correcto o T# por defecto si no había
|
||||
if prefix: return f"{prefix}{value_str}"
|
||||
elif "s5time" in datatype_lower: return f"S5T#{value_str}"
|
||||
elif "ltime" in datatype_lower: return f"LT#{value_str}"
|
||||
else: return f"T#{value_str}" # Default a TIME
|
||||
elif "date" in datatype_lower: # Añadido Date, DT, TOD
|
||||
if value_str.upper().startswith("D#"): return value_str
|
||||
elif "dt" in datatype_lower or "date_and_time" in datatype_lower:
|
||||
if value_str.upper().startswith("DT#"): return value_str
|
||||
else: return f"DT#{value_str}" # Añadir prefijo DT#
|
||||
elif "tod" in datatype_lower or "time_of_day" in datatype_lower:
|
||||
if value_str.upper().startswith("TOD#"): return value_str
|
||||
else: return f"TOD#{value_str}" # Añadir prefijo TOD#
|
||||
else: return f"D#{value_str}" # Default a Date
|
||||
# Fallback genérico
|
||||
else:
|
||||
if re.match(r'^[a-zA-Z_][a-zA-Z0-9_."#\[\]]+$', value_str): # Permitir más caracteres en símbolos/tipos
|
||||
# Si es un UDT o Struct complejo, podría venir con comillas, quitarlas
|
||||
if value_str.startswith('"') and value_str.endswith('"'):
|
||||
return value_str[1:-1]
|
||||
return value_str
|
||||
else:
|
||||
escaped_value = value_str.replace("'", "''")
|
||||
return f"'{escaped_value}'"
|
||||
|
||||
# --- NUEVA FUNCIÓN RECURSIVA para generar declaraciones SCL (VAR/STRUCT/ARRAY) ---
|
||||
|
||||
# --- Función Principal de Generación SCL (MODIFICADA) ---
|
||||
def generate_scl(processed_json_filepath, output_scl_filepath):
|
||||
"""Genera un archivo SCL a partir del JSON procesado (FC/FB o DB)."""
|
||||
|
||||
if not os.path.exists(processed_json_filepath):
|
||||
print(f"Error: Archivo JSON procesado no encontrado en '{processed_json_filepath}'")
|
||||
return
|
||||
|
||||
print(f"Cargando JSON procesado desde: {processed_json_filepath}")
|
||||
try:
|
||||
with open(processed_json_filepath, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
except Exception as e:
|
||||
print(f"Error al cargar o parsear JSON: {e}"); traceback.print_exc(); return
|
||||
|
||||
# --- Extracción de Información del Bloque (Común) ---
|
||||
block_name = data.get('block_name', 'UnknownBlock')
|
||||
block_number = data.get('block_number')
|
||||
block_lang_original = data.get('language', 'Unknown') # Será "DB" para Data Blocks
|
||||
block_type = data.get('block_type', 'Unknown') # FC, FB, GlobalDB
|
||||
block_comment = data.get('block_comment', '')
|
||||
scl_block_name = format_variable_name(block_name) # Nombre SCL seguro
|
||||
print(f"Generando SCL para: {block_type} '{scl_block_name}' (Original: {block_name}, Lang: {block_lang_original})")
|
||||
scl_output = []
|
||||
|
||||
# --- GENERACIÓN PARA DATA BLOCK (DB) ---
|
||||
if block_lang_original == "DB":
|
||||
print("Modo de generación: DATA_BLOCK")
|
||||
scl_output.append(f"// Block Type: {block_type}")
|
||||
scl_output.append(f"// Block Name (Original): {block_name}")
|
||||
if block_number: scl_output.append(f"// Block Number: {block_number}")
|
||||
if block_comment: scl_output.append(f"// Block Comment: {block_comment}")
|
||||
scl_output.append("")
|
||||
scl_output.append(f"DATA_BLOCK \"{scl_block_name}\"")
|
||||
scl_output.append("{ S7_Optimized_Access := 'TRUE' }") # Asumir optimizado
|
||||
scl_output.append("VERSION : 0.1")
|
||||
scl_output.append("")
|
||||
interface_data = data.get('interface', {})
|
||||
static_vars = interface_data.get('Static', [])
|
||||
if static_vars:
|
||||
scl_output.append("VAR")
|
||||
scl_output.extend(generate_scl_declarations(static_vars, indent_level=1))
|
||||
scl_output.append("END_VAR")
|
||||
scl_output.append("")
|
||||
else:
|
||||
print("Advertencia: No se encontró sección 'Static' o está vacía en la interfaz del DB.")
|
||||
scl_output.append("VAR"); scl_output.append("END_VAR"); scl_output.append("")
|
||||
scl_output.append("BEGIN"); scl_output.append(""); scl_output.append("END_DATA_BLOCK")
|
||||
|
||||
# --- GENERACIÓN PARA FUNCTION BLOCK / FUNCTION (FC/FB) ---
|
||||
else:
|
||||
print("Modo de generación: FUNCTION_BLOCK / FUNCTION")
|
||||
scl_block_keyword = "FUNCTION_BLOCK" if block_type == "FB" else "FUNCTION"
|
||||
# Cabecera del Bloque
|
||||
scl_output.append(f"// Block Type: {block_type}")
|
||||
scl_output.append(f"// Block Name (Original): {block_name}")
|
||||
if block_number: scl_output.append(f"// Block Number: {block_number}")
|
||||
scl_output.append(f"// Original Language: {block_lang_original}")
|
||||
if block_comment: scl_output.append(f"// Block Comment: {block_comment}")
|
||||
scl_output.append("")
|
||||
# Manejar tipo de retorno para FUNCTION
|
||||
return_type = "Void" # Default
|
||||
interface_data = data.get('interface', {})
|
||||
if scl_block_keyword == "FUNCTION" and interface_data.get('Return'):
|
||||
return_member = interface_data['Return'][0] # Asumir un solo valor de retorno
|
||||
return_type_raw = return_member.get('datatype', 'Void')
|
||||
return_type = return_type_raw.strip('"') if return_type_raw.startswith('"') and return_type_raw.endswith('"') else return_type_raw
|
||||
# Añadir comillas si es UDT
|
||||
if return_type != return_type_raw: return_type = f'"{return_type}"'
|
||||
|
||||
scl_output.append(f"{scl_block_keyword} \"{scl_block_name}\" : {return_type}" if scl_block_keyword == "FUNCTION" else f"{scl_block_keyword} \"{scl_block_name}\"")
|
||||
scl_output.append("{ S7_Optimized_Access := 'TRUE' }")
|
||||
scl_output.append("VERSION : 0.1"); scl_output.append("")
|
||||
|
||||
# Declaraciones de Interfaz FC/FB
|
||||
section_order = ["Input", "Output", "InOut", "Static", "Temp", "Constant"] # Return ya está en cabecera
|
||||
declared_temps = set()
|
||||
for section_name in section_order:
|
||||
vars_in_section = interface_data.get(section_name, [])
|
||||
if vars_in_section:
|
||||
scl_section_keyword = f"VAR_{section_name.upper()}"
|
||||
if section_name == "Static": scl_section_keyword = "VAR_STAT"
|
||||
if section_name == "Temp": scl_section_keyword = "VAR_TEMP"
|
||||
if section_name == "Constant": scl_section_keyword = "CONSTANT"
|
||||
scl_output.append(scl_section_keyword)
|
||||
scl_output.extend(generate_scl_declarations(vars_in_section, indent_level=1))
|
||||
if section_name == "Temp": declared_temps.update(format_variable_name(v.get("name")) for v in vars_in_section if v.get("name"))
|
||||
scl_output.append("END_VAR"); scl_output.append("")
|
||||
|
||||
# Declaraciones VAR_TEMP adicionales detectadas
|
||||
temp_vars = set()
|
||||
temp_pattern = re.compile(r'"?#(_temp_[a-zA-Z0-9_]+)"?|"?(_temp_[a-zA-Z0-9_]+)"?')
|
||||
for network in data.get('networks', []):
|
||||
for instruction in network.get('logic', []):
|
||||
scl_code = instruction.get('scl', ''); edge_update_code = instruction.get('_edge_mem_update_scl','')
|
||||
code_to_scan = (scl_code if scl_code else '') + '\n' + (edge_update_code if edge_update_code else '')
|
||||
if code_to_scan:
|
||||
found_temps = temp_pattern.findall(code_to_scan)
|
||||
for temp_tuple in found_temps:
|
||||
temp_name = next((t for t in temp_tuple if t), None)
|
||||
if temp_name: temp_vars.add("#"+temp_name if not temp_name.startswith("#") else temp_name)
|
||||
additional_temps = sorted(list(temp_vars - declared_temps))
|
||||
if additional_temps:
|
||||
if not interface_data.get("Temp"): scl_output.append("VAR_TEMP")
|
||||
for var_name in additional_temps:
|
||||
scl_name = format_variable_name(var_name); inferred_type = "Bool" # Asumir Bool
|
||||
scl_output.append(f" {scl_name} : {inferred_type}; // Auto-generated temporary")
|
||||
if not interface_data.get("Temp"): scl_output.append("END_VAR"); scl_output.append("")
|
||||
|
||||
# Cuerpo del Bloque FC/FB
|
||||
scl_output.append("BEGIN"); scl_output.append("")
|
||||
# Iterar por redes y lógica (como antes, incluyendo manejo STL Markdown)
|
||||
for i, network in enumerate(data.get('networks', [])):
|
||||
network_title = network.get('title', f'Network {network.get("id")}')
|
||||
network_comment = network.get('comment', '')
|
||||
network_lang = network.get('language', 'LAD')
|
||||
scl_output.append(f" // Network {i+1}: {network_title} (Original Language: {network_lang})")
|
||||
if network_comment:
|
||||
for line in network_comment.splitlines(): scl_output.append(f" // {line}")
|
||||
scl_output.append("")
|
||||
network_has_code = False
|
||||
if network_lang == "STL":
|
||||
network_has_code = True
|
||||
if network.get('logic') and network['logic'][0].get("type") == "RAW_STL_CHUNK":
|
||||
raw_stl_code = network['logic'][0].get("stl", "// ERROR: STL code missing")
|
||||
scl_output.append(f" {'//'} ```STL")
|
||||
for stl_line in raw_stl_code.splitlines(): scl_output.append(f" {stl_line}")
|
||||
scl_output.append(f" {'//'} ```")
|
||||
else: scl_output.append(" // ERROR: Contenido STL inesperado.")
|
||||
else: # LAD, FBD, SCL, etc.
|
||||
for instruction in network.get('logic', []):
|
||||
instruction_type = instruction.get("type", ""); scl_code = instruction.get('scl', ''); is_grouped = instruction.get("grouped", False)
|
||||
if is_grouped: continue
|
||||
if (instruction_type.endswith(SCL_SUFFIX) or instruction_type in ["RAW_SCL_CHUNK", "UNSUPPORTED_LANG"]) and scl_code:
|
||||
is_only_comment = all(line.strip().startswith("//") for line in scl_code.splitlines() if line.strip())
|
||||
is_if_block = scl_code.strip().startswith("IF")
|
||||
if not is_only_comment or is_if_block:
|
||||
network_has_code = True
|
||||
for line in scl_code.splitlines(): scl_output.append(f" {line}")
|
||||
if network_has_code: scl_output.append("")
|
||||
else: scl_output.append(f" // Network did not produce printable SCL code."); scl_output.append("")
|
||||
# Fin del bloque FC/FB
|
||||
scl_output.append(f"END_{scl_block_keyword}")
|
||||
|
||||
# --- Escritura del Archivo SCL (Común) ---
|
||||
print(f"Escribiendo archivo SCL en: {output_scl_filepath}")
|
||||
try:
|
||||
with open(output_scl_filepath, 'w', encoding='utf-8') as f:
|
||||
for line in scl_output: f.write(line + '\n')
|
||||
print("Generación de SCL completada.")
|
||||
except Exception as e:
|
||||
print(f"Error al escribir el archivo SCL: {e}"); traceback.print_exc()
|
||||
|
||||
|
||||
# --- Ejecución (sin cambios) ---
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(description="Generate final SCL file from processed JSON (FC/FB/DB).")
|
||||
parser.add_argument("source_xml_filepath", help="Path to the original source XML file.")
|
||||
args = parser.parse_args()
|
||||
source_xml_file = args.source_xml_filepath
|
||||
|
||||
if not os.path.exists(source_xml_file):
|
||||
print(f"Advertencia (x3): Archivo XML original no encontrado: '{source_xml_file}', pero se intentará encontrar el JSON procesado.")
|
||||
# Continuar, pero verificar existencia del JSON procesado
|
||||
|
||||
xml_filename_base = os.path.splitext(os.path.basename(source_xml_file))[0]
|
||||
base_dir = os.path.dirname(source_xml_file)
|
||||
input_json_file = os.path.join(base_dir, f"{xml_filename_base}_simplified_processed.json")
|
||||
output_scl_file = os.path.join(base_dir, f"{xml_filename_base}_simplified_processed.scl")
|
||||
|
||||
print(f"(x3) Generando SCL: '{os.path.relpath(input_json_file)}' -> '{os.path.relpath(output_scl_file)}'")
|
||||
|
||||
if not os.path.exists(input_json_file):
|
||||
print(f"Error Fatal (x3): Archivo JSON procesado no encontrado: '{input_json_file}'")
|
||||
print(f"Asegúrate de que 'x1_to_json.py' y 'x2_process.py' se ejecutaron correctamente para '{os.path.relpath(source_xml_file)}'.")
|
||||
sys.exit(1)
|
||||
else:
|
||||
try:
|
||||
generate_scl(input_json_file, output_scl_file)
|
||||
except Exception as e:
|
||||
print(f"Error Crítico (x3) durante la generación de SCL desde '{input_json_file}': {e}")
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
107
x0_main.py
107
x0_main.py
|
|
@ -3,7 +3,8 @@ import subprocess
|
|||
import os
|
||||
import sys
|
||||
import locale
|
||||
import glob # <--- Importar glob para buscar archivos
|
||||
import glob # <--- Importar glob para buscar archivos
|
||||
|
||||
|
||||
# (Función get_console_encoding y variable CONSOLE_ENCODING como en la respuesta anterior)
|
||||
def get_console_encoding():
|
||||
|
|
@ -11,12 +12,14 @@ def get_console_encoding():
|
|||
try:
|
||||
return locale.getpreferredencoding(False)
|
||||
except Exception:
|
||||
return 'cp1252'
|
||||
return "cp1252"
|
||||
|
||||
|
||||
CONSOLE_ENCODING = get_console_encoding()
|
||||
# Descomenta la siguiente línea si quieres ver la codificación detectada:
|
||||
# print(f"Detected console encoding: {CONSOLE_ENCODING}")
|
||||
|
||||
|
||||
# (Función run_script como en la respuesta anterior, usando CONSOLE_ENCODING)
|
||||
def run_script(script_name, xml_arg):
|
||||
"""Runs a given script with the specified XML file argument."""
|
||||
|
|
@ -24,12 +27,14 @@ def run_script(script_name, xml_arg):
|
|||
command = [sys.executable, script_path, xml_arg]
|
||||
print(f"\n--- Running {script_name} with argument: {xml_arg} ---")
|
||||
try:
|
||||
result = subprocess.run(command,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding=CONSOLE_ENCODING,
|
||||
errors='replace') # 'replace' para evitar errores
|
||||
result = subprocess.run(
|
||||
command,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding=CONSOLE_ENCODING,
|
||||
errors="replace",
|
||||
) # 'replace' para evitar errores
|
||||
|
||||
# Imprimir stdout y stderr
|
||||
# Eliminar saltos de línea extra al final si existen
|
||||
|
|
@ -49,26 +54,35 @@ def run_script(script_name, xml_arg):
|
|||
except subprocess.CalledProcessError as e:
|
||||
print(f"Error running {script_name}:")
|
||||
print(f"Return code: {e.returncode}")
|
||||
stdout_decoded = e.stdout.decode(CONSOLE_ENCODING, errors='replace').strip() if isinstance(e.stdout, bytes) else (e.stdout or "").strip()
|
||||
stderr_decoded = e.stderr.decode(CONSOLE_ENCODING, errors='replace').strip() if isinstance(e.stderr, bytes) else (e.stderr or "").strip()
|
||||
stdout_decoded = (
|
||||
e.stdout.decode(CONSOLE_ENCODING, errors="replace").strip()
|
||||
if isinstance(e.stdout, bytes)
|
||||
else (e.stdout or "").strip()
|
||||
)
|
||||
stderr_decoded = (
|
||||
e.stderr.decode(CONSOLE_ENCODING, errors="replace").strip()
|
||||
if isinstance(e.stderr, bytes)
|
||||
else (e.stderr or "").strip()
|
||||
)
|
||||
if stdout_decoded:
|
||||
print("--- Stdout ---")
|
||||
print(stdout_decoded)
|
||||
print("--- Stdout ---")
|
||||
print(stdout_decoded)
|
||||
if stderr_decoded:
|
||||
print("--- Stderr ---")
|
||||
print(stderr_decoded)
|
||||
print("--- Stderr ---")
|
||||
print(stderr_decoded)
|
||||
print("--------------")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"An unexpected error occurred while running {script_name}: {e}")
|
||||
return False
|
||||
|
||||
|
||||
# --- NUEVA FUNCIÓN PARA SELECCIONAR ARCHIVO ---
|
||||
def select_xml_file():
|
||||
"""Busca archivos .xml, los lista y pide al usuario que elija uno."""
|
||||
print("No XML file specified. Searching for XML files in current directory...")
|
||||
# Buscar archivos .xml en el directorio actual (.)
|
||||
xml_files = sorted(glob.glob('*.xml')) # sorted para orden alfabético
|
||||
xml_files = sorted(glob.glob("*.xml")) # sorted para orden alfabético
|
||||
|
||||
if not xml_files:
|
||||
print("Error: No .xml files found in the current directory.")
|
||||
|
|
@ -80,7 +94,9 @@ def select_xml_file():
|
|||
|
||||
while True:
|
||||
try:
|
||||
choice = input(f"Enter the number of the file to process (1-{len(xml_files)}): ")
|
||||
choice = input(
|
||||
f"Enter the number of the file to process (1-{len(xml_files)}): "
|
||||
)
|
||||
choice_num = int(choice)
|
||||
if 1 <= choice_num <= len(xml_files):
|
||||
selected_file = xml_files[choice_num - 1]
|
||||
|
|
@ -90,9 +106,11 @@ def select_xml_file():
|
|||
print("Invalid choice. Please enter a number from the list.")
|
||||
except ValueError:
|
||||
print("Invalid input. Please enter a number.")
|
||||
except EOFError: # Manejar si la entrada se cierra inesperadamente
|
||||
print("\nSelection cancelled.")
|
||||
sys.exit(1)
|
||||
except EOFError: # Manejar si la entrada se cierra inesperadamente
|
||||
print("\nSelection cancelled.")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
# --- FIN NUEVA FUNCIÓN ---
|
||||
|
||||
|
||||
|
|
@ -100,30 +118,36 @@ if __name__ == "__main__":
|
|||
# Imports necesarios para esta sección
|
||||
import os
|
||||
import sys
|
||||
import glob # Asegúrate de que glob esté importado al principio del archivo
|
||||
import glob # Asegúrate de que glob esté importado al principio del archivo
|
||||
|
||||
# Directorio base donde buscar los archivos XML (relativo al script)
|
||||
base_search_dir = "XML Project"
|
||||
script_dir = os.path.dirname(__file__) # Directorio donde está x0_main.py
|
||||
script_dir = os.path.dirname(__file__) # Directorio donde está x0_main.py
|
||||
xml_project_dir = os.path.join(script_dir, base_search_dir)
|
||||
|
||||
print(f"Buscando archivos XML recursivamente en: '{xml_project_dir}'")
|
||||
|
||||
# Verificar si el directorio 'XML Project' existe
|
||||
if not os.path.isdir(xml_project_dir):
|
||||
print(f"Error: El directorio '{xml_project_dir}' no existe o no es un directorio.")
|
||||
print("Por favor, crea el directorio 'XML Project' en la misma carpeta que este script y coloca tus archivos XML dentro.")
|
||||
print(
|
||||
f"Error: El directorio '{xml_project_dir}' no existe o no es un directorio."
|
||||
)
|
||||
print(
|
||||
"Por favor, crea el directorio 'XML Project' en la misma carpeta que este script y coloca tus archivos XML dentro."
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
# Buscar todos los archivos .xml recursivamente dentro de xml_project_dir
|
||||
# Usamos os.path.join para construir la ruta de búsqueda correctamente
|
||||
# y '**/*.xml' para la recursividad con glob
|
||||
search_pattern = os.path.join(xml_project_dir, '**', '*.xml')
|
||||
search_pattern = os.path.join(xml_project_dir, "**", "*.xml")
|
||||
xml_files_found = glob.glob(search_pattern, recursive=True)
|
||||
|
||||
if not xml_files_found:
|
||||
print(f"No se encontraron archivos XML en '{xml_project_dir}' o sus subdirectorios.")
|
||||
sys.exit(0) # Salir limpiamente si no hay archivos
|
||||
print(
|
||||
f"No se encontraron archivos XML en '{xml_project_dir}' o sus subdirectorios."
|
||||
)
|
||||
sys.exit(0) # Salir limpiamente si no hay archivos
|
||||
|
||||
print(f"Se encontraron {len(xml_files_found)} archivos XML para procesar:")
|
||||
# Ordenar para un procesamiento predecible (opcional)
|
||||
|
|
@ -141,7 +165,9 @@ if __name__ == "__main__":
|
|||
processed_count = 0
|
||||
failed_count = 0
|
||||
for xml_filepath in xml_files_found:
|
||||
print(f"\n--- Iniciando pipeline para: {os.path.relpath(xml_filepath, script_dir)} ---")
|
||||
print(
|
||||
f"\n--- Iniciando pipeline para: {os.path.relpath(xml_filepath, script_dir)} ---"
|
||||
)
|
||||
|
||||
# Usar la ruta absoluta para evitar problemas si los scripts cambian de directorio
|
||||
absolute_xml_filepath = os.path.abspath(xml_filepath)
|
||||
|
|
@ -150,26 +176,37 @@ if __name__ == "__main__":
|
|||
# La función run_script ya está definida en tu script x0_main.py
|
||||
success = True
|
||||
if not run_script(script1, absolute_xml_filepath):
|
||||
print(f"\nPipeline falló en el script '{script1}' para el archivo: {os.path.relpath(xml_filepath, script_dir)}")
|
||||
print(
|
||||
f"\nPipeline falló en el script '{script1}' para el archivo: {os.path.relpath(xml_filepath, script_dir)}"
|
||||
)
|
||||
success = False
|
||||
elif not run_script(script2, absolute_xml_filepath):
|
||||
print(f"\nPipeline falló en el script '{script2}' para el archivo: {os.path.relpath(xml_filepath, script_dir)}")
|
||||
print(
|
||||
f"\nPipeline falló en el script '{script2}' para el archivo: {os.path.relpath(xml_filepath, script_dir)}"
|
||||
)
|
||||
success = False
|
||||
elif not run_script(script3, absolute_xml_filepath):
|
||||
print(f"\nPipeline falló en el script '{script3}' para el archivo: {os.path.relpath(xml_filepath, script_dir)}")
|
||||
print(
|
||||
f"\nPipeline falló en el script '{script3}' para el archivo: {os.path.relpath(xml_filepath, script_dir)}"
|
||||
)
|
||||
success = False
|
||||
|
||||
if success:
|
||||
print(f"--- Pipeline completado exitosamente para: {os.path.relpath(xml_filepath, script_dir)} ---")
|
||||
print(
|
||||
f"--- Pipeline completado exitosamente para: {os.path.relpath(xml_filepath, script_dir)} ---"
|
||||
)
|
||||
processed_count += 1
|
||||
else:
|
||||
failed_count += 1
|
||||
print(f"--- Pipeline falló para: {os.path.relpath(xml_filepath, script_dir)} ---")
|
||||
|
||||
print(
|
||||
f"--- Pipeline falló para: {os.path.relpath(xml_filepath, script_dir)} ---"
|
||||
)
|
||||
|
||||
print("\n--- Resumen Final del Procesamiento ---")
|
||||
print(f"Total de archivos XML encontrados: {len(xml_files_found)}")
|
||||
print(f"Archivos procesados exitosamente por el pipeline completo: {processed_count}")
|
||||
print(
|
||||
f"Archivos procesados exitosamente por el pipeline completo: {processed_count}"
|
||||
)
|
||||
print(f"Archivos que fallaron en algún punto del pipeline: {failed_count}")
|
||||
print("---------------------------------------")
|
||||
xml_filename = None
|
||||
|
|
@ -217,4 +254,4 @@ if __name__ == "__main__":
|
|||
else:
|
||||
print("\nPipeline failed at script:", script2)
|
||||
else:
|
||||
print("\nPipeline failed at script:", script1)
|
||||
print("\nPipeline failed at script:", script1)
|
||||
|
|
|
|||
683
x1_to_json.py
683
x1_to_json.py
File diff suppressed because it is too large
Load Diff
485
x2_process.py
485
x2_process.py
|
|
@ -5,29 +5,30 @@ import os
|
|||
import copy
|
||||
import traceback
|
||||
import re
|
||||
import importlib
|
||||
import sys
|
||||
import sympy # Import sympy
|
||||
import importlib
|
||||
import sys
|
||||
import sympy # Import sympy
|
||||
|
||||
# Import necessary components from processors directory
|
||||
from processors.processor_utils import (
|
||||
format_variable_name, # Keep if used outside processors
|
||||
sympy_expr_to_scl, # Needed for IF grouping and maybe others
|
||||
format_variable_name, # Keep if used outside processors
|
||||
sympy_expr_to_scl, # Needed for IF grouping and maybe others
|
||||
# get_target_scl_name might be used here? Unlikely.
|
||||
)
|
||||
from processors.symbol_manager import SymbolManager # Import the manager
|
||||
from processors.symbol_manager import SymbolManager # Import the manager
|
||||
|
||||
# --- Constantes y Configuración ---
|
||||
# SCL_SUFFIX = "_scl" # Old suffix
|
||||
SCL_SUFFIX = "_sympy_processed" # New suffix to indicate processing method
|
||||
SCL_SUFFIX = "_sympy_processed" # New suffix to indicate processing method
|
||||
GROUPED_COMMENT = "// Logic included in grouped IF"
|
||||
SIMPLIFIED_IF_COMMENT = "// Simplified IF condition by script" # May still be useful
|
||||
SIMPLIFIED_IF_COMMENT = "// Simplified IF condition by script" # May still be useful
|
||||
|
||||
# Global data dictionary (consider passing 'data' as argument if needed elsewhere)
|
||||
# It's currently used by process_group_ifs implicitly via the outer scope,
|
||||
# which works but passing it explicitly might be cleaner.
|
||||
data = {}
|
||||
|
||||
|
||||
def process_group_ifs(instruction, network_id, sympy_map, symbol_manager, data):
|
||||
"""
|
||||
Busca condiciones (ya procesadas -> tienen expr SymPy en sympy_map)
|
||||
|
|
@ -37,24 +38,45 @@ def process_group_ifs(instruction, network_id, sympy_map, symbol_manager, data):
|
|||
(Esta es la implementación de la función como la tenías en el archivo original)
|
||||
"""
|
||||
instr_uid = instruction["instruction_uid"]
|
||||
instr_type_original = instruction.get("type", "").replace(SCL_SUFFIX, "").replace("_error", "")
|
||||
instr_type_original = (
|
||||
instruction.get("type", "").replace(SCL_SUFFIX, "").replace("_error", "")
|
||||
)
|
||||
made_change = False
|
||||
|
||||
# Check if this instruction *could* generate a condition suitable for grouping
|
||||
# It must have been processed by the new SymPy method
|
||||
if (
|
||||
not instruction.get("type", "").endswith(SCL_SUFFIX) # Check if processed by new method
|
||||
not instruction.get("type", "").endswith(
|
||||
SCL_SUFFIX
|
||||
) # Check if processed by new method
|
||||
or "_error" in instruction.get("type", "")
|
||||
or instruction.get("grouped", False)
|
||||
or instr_type_original not in [ # Original types that produce boolean results
|
||||
"Contact", "O", "Eq", "Ne", "Gt", "Lt", "Ge", "Le", "PBox", "NBox", "And", "Xor", "Not" # Add others like comparison
|
||||
or instr_type_original
|
||||
not in [ # Original types that produce boolean results
|
||||
"Contact",
|
||||
"O",
|
||||
"Eq",
|
||||
"Ne",
|
||||
"Gt",
|
||||
"Lt",
|
||||
"Ge",
|
||||
"Le",
|
||||
"PBox",
|
||||
"NBox",
|
||||
"And",
|
||||
"Xor",
|
||||
"Not", # Add others like comparison
|
||||
]
|
||||
):
|
||||
return False
|
||||
|
||||
# Avoid reagruping if SCL already contains a complex IF (less likely now)
|
||||
current_scl = instruction.get("scl", "")
|
||||
if current_scl.strip().startswith("IF") and "END_IF;" in current_scl and GROUPED_COMMENT not in current_scl:
|
||||
if (
|
||||
current_scl.strip().startswith("IF")
|
||||
and "END_IF;" in current_scl
|
||||
and GROUPED_COMMENT not in current_scl
|
||||
):
|
||||
return False
|
||||
|
||||
# *** Get the SymPy expression for the condition ***
|
||||
|
|
@ -62,20 +84,34 @@ def process_group_ifs(instruction, network_id, sympy_map, symbol_manager, data):
|
|||
sympy_condition_expr = sympy_map.get(map_key_out)
|
||||
|
||||
# No SymPy expression found or trivial conditions
|
||||
if sympy_condition_expr is None or sympy_condition_expr in [sympy.true, sympy.false]:
|
||||
if sympy_condition_expr is None or sympy_condition_expr in [
|
||||
sympy.true,
|
||||
sympy.false,
|
||||
]:
|
||||
return False
|
||||
|
||||
# --- Find consumer instructions (logic similar to before) ---
|
||||
grouped_instructions_cores = []
|
||||
consumer_instr_list = []
|
||||
network_logic = next((net["logic"] for net in data["networks"] if net["id"] == network_id), [])
|
||||
if not network_logic: return False
|
||||
network_logic = next(
|
||||
(net["logic"] for net in data["networks"] if net["id"] == network_id), []
|
||||
)
|
||||
if not network_logic:
|
||||
return False
|
||||
|
||||
groupable_types = [ # Types whose *final SCL* we want to group
|
||||
"Move", "Add", "Sub", "Mul", "Div", "Mod", "Convert",
|
||||
"Call_FC", "Call_FB", # Assuming these generate final SCL in their processors now
|
||||
groupable_types = [ # Types whose *final SCL* we want to group
|
||||
"Move",
|
||||
"Add",
|
||||
"Sub",
|
||||
"Mul",
|
||||
"Div",
|
||||
"Mod",
|
||||
"Convert",
|
||||
"Call_FC",
|
||||
"Call_FB", # Assuming these generate final SCL in their processors now
|
||||
# SCoil/RCoil might also be groupable if their SCL is final assignment
|
||||
"SCoil", "RCoil"
|
||||
"SCoil",
|
||||
"RCoil",
|
||||
]
|
||||
|
||||
for consumer_instr in network_logic:
|
||||
|
|
@ -84,31 +120,45 @@ def process_group_ifs(instruction, network_id, sympy_map, symbol_manager, data):
|
|||
continue
|
||||
|
||||
consumer_en = consumer_instr.get("inputs", {}).get("en")
|
||||
consumer_type = consumer_instr.get("type", "") # Current type suffix matters
|
||||
consumer_type_original = consumer_type.replace(SCL_SUFFIX, "").replace("_error", "")
|
||||
consumer_type = consumer_instr.get("type", "") # Current type suffix matters
|
||||
consumer_type_original = consumer_type.replace(SCL_SUFFIX, "").replace(
|
||||
"_error", ""
|
||||
)
|
||||
|
||||
is_enabled_by_us = False
|
||||
if ( isinstance(consumer_en, dict) and consumer_en.get("type") == "connection" and
|
||||
consumer_en.get("source_instruction_uid") == instr_uid and
|
||||
consumer_en.get("source_pin") == "out"):
|
||||
if (
|
||||
isinstance(consumer_en, dict)
|
||||
and consumer_en.get("type") == "connection"
|
||||
and consumer_en.get("source_instruction_uid") == instr_uid
|
||||
and consumer_en.get("source_pin") == "out"
|
||||
):
|
||||
is_enabled_by_us = True
|
||||
|
||||
# Check if consumer is groupable AND has its final SCL generated
|
||||
# The suffix check needs adjustment based on how terminating processors set it.
|
||||
# Assuming processors like Move, Add, Call, SCoil, RCoil NOW generate final SCL and add a suffix.
|
||||
if ( is_enabled_by_us and consumer_type.endswith(SCL_SUFFIX) and # Or a specific "final_scl" suffix
|
||||
consumer_type_original in groupable_types ):
|
||||
if (
|
||||
is_enabled_by_us
|
||||
and consumer_type.endswith(SCL_SUFFIX) # Or a specific "final_scl" suffix
|
||||
and consumer_type_original in groupable_types
|
||||
):
|
||||
|
||||
consumer_scl = consumer_instr.get("scl", "")
|
||||
# Extract core SCL (logic is similar, maybe simpler if SCL is cleaner now)
|
||||
core_scl = None
|
||||
if consumer_scl:
|
||||
# If consumer SCL itself is an IF generated by EN, take the body
|
||||
if consumer_scl.strip().startswith("IF"):
|
||||
match = re.search(r"THEN\s*(.*?)\s*END_IF;", consumer_scl, re.DOTALL | re.IGNORECASE)
|
||||
core_scl = match.group(1).strip() if match else None
|
||||
elif not consumer_scl.strip().startswith("//"): # Otherwise, take the whole line if not comment
|
||||
core_scl = consumer_scl.strip()
|
||||
# If consumer SCL itself is an IF generated by EN, take the body
|
||||
if consumer_scl.strip().startswith("IF"):
|
||||
match = re.search(
|
||||
r"THEN\s*(.*?)\s*END_IF;",
|
||||
consumer_scl,
|
||||
re.DOTALL | re.IGNORECASE,
|
||||
)
|
||||
core_scl = match.group(1).strip() if match else None
|
||||
elif not consumer_scl.strip().startswith(
|
||||
"//"
|
||||
): # Otherwise, take the whole line if not comment
|
||||
core_scl = consumer_scl.strip()
|
||||
|
||||
if core_scl:
|
||||
grouped_instructions_cores.append(core_scl)
|
||||
|
|
@ -116,15 +166,19 @@ def process_group_ifs(instruction, network_id, sympy_map, symbol_manager, data):
|
|||
|
||||
# --- If groupable consumers found ---
|
||||
if len(grouped_instructions_cores) > 1:
|
||||
print(f"INFO: Agrupando {len(grouped_instructions_cores)} instr. bajo condición de {instr_type_original} UID {instr_uid}")
|
||||
print(
|
||||
f"INFO: Agrupando {len(grouped_instructions_cores)} instr. bajo condición de {instr_type_original} UID {instr_uid}"
|
||||
)
|
||||
|
||||
# *** Simplify the SymPy condition ***
|
||||
try:
|
||||
#simplified_expr = sympy.simplify_logic(sympy_condition_expr, force=True)
|
||||
simplified_expr = sympy.logic.boolalg.to_dnf(sympy_condition_expr, simplify=True)
|
||||
# simplified_expr = sympy.simplify_logic(sympy_condition_expr, force=True)
|
||||
simplified_expr = sympy.logic.boolalg.to_dnf(
|
||||
sympy_condition_expr, simplify=True
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"Error simplifying condition for grouping UID {instr_uid}: {e}")
|
||||
simplified_expr = sympy_condition_expr # Fallback
|
||||
simplified_expr = sympy_condition_expr # Fallback
|
||||
|
||||
# *** Convert simplified condition to SCL string ***
|
||||
condition_scl_simplified = sympy_expr_to_scl(simplified_expr, symbol_manager)
|
||||
|
|
@ -132,7 +186,9 @@ def process_group_ifs(instruction, network_id, sympy_map, symbol_manager, data):
|
|||
# *** Build the grouped IF SCL ***
|
||||
scl_grouped_lines = [f"IF {condition_scl_simplified} THEN"]
|
||||
for core_line in grouped_instructions_cores:
|
||||
indented_core = "\n".join([f" {line.strip()}" for line in core_line.splitlines()])
|
||||
indented_core = "\n".join(
|
||||
[f" {line.strip()}" for line in core_line.splitlines()]
|
||||
)
|
||||
scl_grouped_lines.append(indented_core)
|
||||
scl_grouped_lines.append("END_IF;")
|
||||
final_grouped_scl = "\n".join(scl_grouped_lines)
|
||||
|
|
@ -147,18 +203,19 @@ def process_group_ifs(instruction, network_id, sympy_map, symbol_manager, data):
|
|||
|
||||
return made_change
|
||||
|
||||
|
||||
def load_processors(processors_dir="processors"):
|
||||
"""
|
||||
Escanea el directorio, importa módulos, construye el mapa y una lista
|
||||
ordenada por prioridad.
|
||||
"""
|
||||
processor_map = {}
|
||||
processor_list_unsorted = [] # Lista para guardar (priority, type_name, func)
|
||||
default_priority = 10 # Prioridad si no se define en get_processor_info
|
||||
processor_list_unsorted = [] # Lista para guardar (priority, type_name, func)
|
||||
default_priority = 10 # Prioridad si no se define en get_processor_info
|
||||
|
||||
if not os.path.isdir(processors_dir):
|
||||
print(f"Error: Directorio de procesadores no encontrado: '{processors_dir}'")
|
||||
return processor_map, [] # Devuelve mapa vacío y lista vacía
|
||||
return processor_map, [] # Devuelve mapa vacío y lista vacía
|
||||
|
||||
print(f"Cargando procesadores desde: '{processors_dir}'")
|
||||
processors_package = os.path.basename(processors_dir)
|
||||
|
|
@ -170,7 +227,9 @@ def load_processors(processors_dir="processors"):
|
|||
try:
|
||||
module = importlib.import_module(full_module_name)
|
||||
|
||||
if hasattr(module, 'get_processor_info') and callable(module.get_processor_info):
|
||||
if hasattr(module, "get_processor_info") and callable(
|
||||
module.get_processor_info
|
||||
):
|
||||
processor_info = module.get_processor_info()
|
||||
info_list = []
|
||||
if isinstance(processor_info, dict):
|
||||
|
|
@ -178,29 +237,51 @@ def load_processors(processors_dir="processors"):
|
|||
elif isinstance(processor_info, list):
|
||||
info_list = processor_info
|
||||
else:
|
||||
print(f" Advertencia: get_processor_info en {full_module_name} devolvió tipo inesperado. Se ignora.")
|
||||
print(
|
||||
f" Advertencia: get_processor_info en {full_module_name} devolvió tipo inesperado. Se ignora."
|
||||
)
|
||||
continue
|
||||
|
||||
for info in info_list:
|
||||
if isinstance(info, dict) and 'type_name' in info and 'processor_func' in info:
|
||||
type_name = info['type_name'].lower()
|
||||
processor_func = info['processor_func']
|
||||
if (
|
||||
isinstance(info, dict)
|
||||
and "type_name" in info
|
||||
and "processor_func" in info
|
||||
):
|
||||
type_name = info["type_name"].lower()
|
||||
processor_func = info["processor_func"]
|
||||
# Obtener prioridad, usar default si no existe
|
||||
priority = info.get('priority', default_priority)
|
||||
priority = info.get("priority", default_priority)
|
||||
|
||||
if callable(processor_func):
|
||||
if type_name in processor_map:
|
||||
print(f" Advertencia: '{type_name}' en {full_module_name} sobrescribe definición anterior.")
|
||||
print(
|
||||
f" Advertencia: '{type_name}' en {full_module_name} sobrescribe definición anterior."
|
||||
)
|
||||
processor_map[type_name] = processor_func
|
||||
# Añadir a la lista para ordenar
|
||||
processor_list_unsorted.append({'priority': priority, 'type_name': type_name, 'func': processor_func})
|
||||
print(f" - Cargado '{type_name}' (Prio: {priority}) desde {module_name_rel}.py")
|
||||
processor_list_unsorted.append(
|
||||
{
|
||||
"priority": priority,
|
||||
"type_name": type_name,
|
||||
"func": processor_func,
|
||||
}
|
||||
)
|
||||
print(
|
||||
f" - Cargado '{type_name}' (Prio: {priority}) desde {module_name_rel}.py"
|
||||
)
|
||||
else:
|
||||
print(f" Advertencia: 'processor_func' para '{type_name}' en {full_module_name} no es callable.")
|
||||
print(
|
||||
f" Advertencia: 'processor_func' para '{type_name}' en {full_module_name} no es callable."
|
||||
)
|
||||
else:
|
||||
print(f" Advertencia: Entrada inválida en {full_module_name}: {info}")
|
||||
print(
|
||||
f" Advertencia: Entrada inválida en {full_module_name}: {info}"
|
||||
)
|
||||
else:
|
||||
print(f" Advertencia: Módulo {module_name_rel}.py no tiene 'get_processor_info'.")
|
||||
print(
|
||||
f" Advertencia: Módulo {module_name_rel}.py no tiene 'get_processor_info'."
|
||||
)
|
||||
|
||||
except ImportError as e:
|
||||
print(f"Error importando {full_module_name}: {e}")
|
||||
|
|
@ -209,22 +290,24 @@ def load_processors(processors_dir="processors"):
|
|||
traceback.print_exc()
|
||||
|
||||
# Ordenar la lista por prioridad (menor primero)
|
||||
processor_list_sorted = sorted(processor_list_unsorted, key=lambda x: x['priority'])
|
||||
processor_list_sorted = sorted(processor_list_unsorted, key=lambda x: x["priority"])
|
||||
|
||||
print(f"\nTotal de tipos de procesadores cargados: {len(processor_map)}")
|
||||
print(f"Orden de procesamiento por prioridad: {[item['type_name'] for item in processor_list_sorted]}")
|
||||
print(
|
||||
f"Orden de procesamiento por prioridad: {[item['type_name'] for item in processor_list_sorted]}"
|
||||
)
|
||||
|
||||
# Devolver el mapa (para lookup rápido si es necesario) y la lista ordenada
|
||||
return processor_map, processor_list_sorted
|
||||
|
||||
|
||||
# --- Bucle Principal de Procesamiento (Modificado para STL) ---
|
||||
def process_json_to_scl(json_filepath):
|
||||
"""
|
||||
Lee el JSON simplificado, aplica los procesadores dinámicamente cargados
|
||||
siguiendo un orden de prioridad (ignorando redes STL), y guarda el JSON procesado.
|
||||
Lee JSON simplificado, aplica procesadores dinámicos (ignorando redes STL y bloques DB),
|
||||
y guarda JSON procesado.
|
||||
"""
|
||||
global data # Necesario para que load_processors y process_group_ifs (definidas fuera) puedan acceder a ella.
|
||||
# Considerar pasar 'data' como argumento si es posible refactorizar.
|
||||
global data
|
||||
|
||||
if not os.path.exists(json_filepath):
|
||||
print(f"Error: JSON no encontrado: {json_filepath}")
|
||||
|
|
@ -232,48 +315,83 @@ def process_json_to_scl(json_filepath):
|
|||
print(f"Cargando JSON desde: {json_filepath}")
|
||||
try:
|
||||
with open(json_filepath, "r", encoding="utf-8") as f:
|
||||
data = json.load(f) # Carga en 'data' global
|
||||
data = json.load(f)
|
||||
except Exception as e:
|
||||
print(f"Error al cargar JSON: {e}")
|
||||
traceback.print_exc()
|
||||
return
|
||||
|
||||
# --- Carga dinámica de procesadores ---
|
||||
# --- Obtener lenguaje del bloque principal ---
|
||||
block_language = data.get("language", "Unknown")
|
||||
block_type = data.get("block_type", "Unknown") # FC, FB, GlobalDB
|
||||
print(f"Procesando bloque tipo: {block_type}, Lenguaje principal: {block_language}")
|
||||
|
||||
# --- SI ES UN DB, SALTAR EL PROCESAMIENTO LÓGICO ---
|
||||
if block_language == "DB":
|
||||
print(
|
||||
"INFO: El bloque es un Data Block (DB). Saltando procesamiento lógico de x2."
|
||||
)
|
||||
# Simplemente guardamos una copia (o el mismo archivo si no se requiere sufijo)
|
||||
output_filename = json_filepath.replace(
|
||||
"_simplified.json", "_simplified_processed.json"
|
||||
)
|
||||
print(f"Guardando JSON de DB (sin cambios lógicos) en: {output_filename}")
|
||||
try:
|
||||
with open(output_filename, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, indent=4, ensure_ascii=False)
|
||||
print("Guardado de DB completado.")
|
||||
except Exception as e:
|
||||
print(f"Error Crítico al guardar JSON del DB: {e}")
|
||||
traceback.print_exc()
|
||||
return # <<< SALIR TEMPRANO PARA DBs
|
||||
|
||||
# --- SI NO ES DB, CONTINUAR CON EL PROCESAMIENTO LÓGICO (FC/FB) ---
|
||||
print("INFO: El bloque es FC/FB. Iniciando procesamiento lógico...")
|
||||
|
||||
script_dir = os.path.dirname(__file__)
|
||||
processors_dir_path = os.path.join(script_dir, 'processors')
|
||||
processors_dir_path = os.path.join(script_dir, "processors")
|
||||
processor_map, sorted_processors = load_processors(processors_dir_path)
|
||||
if not processor_map:
|
||||
print("Error crítico: No se cargaron procesadores. Abortando.")
|
||||
return
|
||||
|
||||
# --- Crear mapas de acceso por red ---
|
||||
network_access_maps = {}
|
||||
# (La lógica para llenar network_access_maps no cambia, puedes copiarla de tu original)
|
||||
# Crear mapas de acceso por red (copiado/adaptado de versión anterior)
|
||||
for network in data.get("networks", []):
|
||||
net_id = network["id"]
|
||||
current_access_map = {}
|
||||
for instr in network.get("logic", []):
|
||||
for _, source in instr.get("inputs", {}).items():
|
||||
sources_to_check = (source if isinstance(source, list) else ([source] if isinstance(source, dict) else []))
|
||||
for src in sources_to_check:
|
||||
if (isinstance(src, dict) and src.get("uid") and src.get("type") in ["variable", "constant"]):
|
||||
current_access_map[src["uid"]] = src
|
||||
for _, dest_list in instr.get("outputs", {}).items():
|
||||
if isinstance(dest_list, list):
|
||||
for dest in dest_list:
|
||||
if (isinstance(dest, dict) and dest.get("uid") and dest.get("type") in ["variable", "constant"]):
|
||||
current_access_map[dest["uid"]] = dest
|
||||
for _, source in instr.get("inputs", {}).items():
|
||||
sources_to_check = (
|
||||
source
|
||||
if isinstance(source, list)
|
||||
else ([source] if isinstance(source, dict) else [])
|
||||
)
|
||||
for src in sources_to_check:
|
||||
if (
|
||||
isinstance(src, dict)
|
||||
and src.get("uid")
|
||||
and src.get("type") in ["variable", "constant"]
|
||||
):
|
||||
current_access_map[src["uid"]] = src
|
||||
for _, dest_list in instr.get("outputs", {}).items():
|
||||
if isinstance(dest_list, list):
|
||||
for dest in dest_list:
|
||||
if (
|
||||
isinstance(dest, dict)
|
||||
and dest.get("uid")
|
||||
and dest.get("type") in ["variable", "constant"]
|
||||
):
|
||||
current_access_map[dest["uid"]] = dest
|
||||
network_access_maps[net_id] = current_access_map
|
||||
|
||||
# --- Inicializar mapa SymPy y SymbolManager ---
|
||||
symbol_manager = SymbolManager()
|
||||
sympy_map = {}
|
||||
|
||||
max_passes = 30
|
||||
passes = 0
|
||||
processing_complete = False
|
||||
|
||||
print("\n--- Iniciando Bucle de Procesamiento Iterativo (con SymPy y prioridad) ---")
|
||||
print("\n--- Iniciando Bucle de Procesamiento Iterativo (FC/FB) ---")
|
||||
while passes < max_passes and not processing_complete:
|
||||
passes += 1
|
||||
made_change_in_base_pass = False
|
||||
|
|
@ -284,90 +402,99 @@ def process_json_to_scl(json_filepath):
|
|||
|
||||
# --- FASE 1: Procesadores Base (Ignorando STL) ---
|
||||
print(f" Fase 1 (SymPy Base - Orden por Prioridad):")
|
||||
num_sympy_processed_this_pass = 0
|
||||
num_sympy_processed_this_pass = 0 # Resetear contador para el pase
|
||||
for processor_info in sorted_processors:
|
||||
current_type_name = processor_info['type_name']
|
||||
func_to_call = processor_info['func']
|
||||
|
||||
current_type_name = processor_info["type_name"]
|
||||
func_to_call = processor_info["func"]
|
||||
for network in data.get("networks", []):
|
||||
network_id = network["id"]
|
||||
network_lang = network.get("language", "LAD") # Obtener lenguaje de la red
|
||||
|
||||
# *** IGNORAR REDES STL EN ESTA FASE ***
|
||||
network_lang = network.get("language", "LAD")
|
||||
if network_lang == "STL":
|
||||
continue # Saltar al siguiente network
|
||||
continue # Saltar STL
|
||||
|
||||
access_map = network_access_maps.get(network_id, {})
|
||||
network_logic = network.get("logic", [])
|
||||
|
||||
for instruction in network_logic:
|
||||
instr_uid = instruction.get("instruction_uid")
|
||||
instr_type_original = instruction.get("type", "Unknown")
|
||||
|
||||
# Saltar si ya procesado, error, agrupado o es chunk STL/SCL/Unsupported
|
||||
if (instr_type_original.endswith(SCL_SUFFIX)
|
||||
if (
|
||||
instr_type_original.endswith(SCL_SUFFIX)
|
||||
or "_error" in instr_type_original
|
||||
or instruction.get("grouped", False)
|
||||
or instr_type_original in ["RAW_STL_CHUNK", "RAW_SCL_CHUNK", "UNSUPPORTED_LANG"]):
|
||||
or instr_type_original
|
||||
in ["RAW_STL_CHUNK", "RAW_SCL_CHUNK", "UNSUPPORTED_LANG"]
|
||||
):
|
||||
continue
|
||||
|
||||
# Determinar tipo efectivo (como antes)
|
||||
lookup_key = instr_type_original.lower()
|
||||
effective_type_name = lookup_key
|
||||
if instr_type_original == "Call":
|
||||
block_type = instruction.get("block_type", "").upper()
|
||||
if block_type == "FC": effective_type_name = "call_fc"
|
||||
elif block_type == "FB": effective_type_name = "call_fb"
|
||||
block_type = instruction.get("block_type", "").upper()
|
||||
if block_type == "FC":
|
||||
effective_type_name = "call_fc"
|
||||
elif block_type == "FB":
|
||||
effective_type_name = "call_fb"
|
||||
|
||||
# Llamar al procesador si coincide el tipo
|
||||
if effective_type_name == current_type_name:
|
||||
try:
|
||||
# Pasa sympy_map, symbol_manager y data
|
||||
changed = func_to_call(instruction, network_id, sympy_map, symbol_manager, data)
|
||||
changed = func_to_call(
|
||||
instruction, network_id, sympy_map, symbol_manager, data
|
||||
)
|
||||
if changed:
|
||||
made_change_in_base_pass = True
|
||||
num_sympy_processed_this_pass += 1
|
||||
except Exception as e:
|
||||
print(f"ERROR(SymPy Base) al procesar {instr_type_original} UID {instr_uid}: {e}")
|
||||
traceback.print_exc()
|
||||
instruction["scl"] = f"// ERROR en SymPy procesador base: {e}"
|
||||
instruction["type"] = instr_type_original + "_error"
|
||||
made_change_in_base_pass = True # Marcar cambio aunque sea error
|
||||
print(f" -> {num_sympy_processed_this_pass} instrucciones (no STL) procesadas con SymPy.")
|
||||
|
||||
print(
|
||||
f"ERROR(SymPy Base) al procesar {instr_type_original} UID {instr_uid}: {e}"
|
||||
)
|
||||
traceback.print_exc()
|
||||
instruction["scl"] = (
|
||||
f"// ERROR en SymPy procesador base: {e}"
|
||||
)
|
||||
instruction["type"] = instr_type_original + "_error"
|
||||
made_change_in_base_pass = True
|
||||
print(
|
||||
f" -> {num_sympy_processed_this_pass} instrucciones (no STL) procesadas con SymPy."
|
||||
)
|
||||
|
||||
# --- FASE 2: Agrupación IF (Ignorando STL) ---
|
||||
if made_change_in_base_pass or passes == 1:
|
||||
if (
|
||||
made_change_in_base_pass or passes == 1
|
||||
): # Ejecutar siempre en el primer pase
|
||||
print(f" Fase 2 (Agrupación IF con Simplificación):")
|
||||
num_grouped_this_pass = 0
|
||||
num_grouped_this_pass = 0 # Resetear contador para el pase
|
||||
for network in data.get("networks", []):
|
||||
network_id = network["id"]
|
||||
network_lang = network.get("language", "LAD") # Obtener lenguaje
|
||||
|
||||
# *** IGNORAR REDES STL EN ESTA FASE ***
|
||||
network_lang = network.get("language", "LAD")
|
||||
if network_lang == "STL":
|
||||
continue # Saltar red STL
|
||||
|
||||
continue # Saltar STL
|
||||
network_logic = network.get("logic", [])
|
||||
for instruction in network_logic:
|
||||
try:
|
||||
# Llama a process_group_ifs (que necesita acceso a 'data' global o pasado)
|
||||
group_changed = process_group_ifs(instruction, network_id, sympy_map, symbol_manager, data)
|
||||
if group_changed:
|
||||
made_change_in_group_pass = True
|
||||
num_grouped_this_pass += 1
|
||||
group_changed = process_group_ifs(
|
||||
instruction, network_id, sympy_map, symbol_manager, data
|
||||
)
|
||||
if group_changed:
|
||||
made_change_in_group_pass = True
|
||||
num_grouped_this_pass += 1
|
||||
except Exception as e:
|
||||
print(f"ERROR(GroupLoop) al intentar agrupar desde UID {instruction.get('instruction_uid')}: {e}")
|
||||
traceback.print_exc()
|
||||
print(f" -> {num_grouped_this_pass} agrupaciones realizadas (en redes no STL).")
|
||||
|
||||
print(
|
||||
f"ERROR(GroupLoop) al intentar agrupar desde UID {instruction.get('instruction_uid')}: {e}"
|
||||
)
|
||||
traceback.print_exc()
|
||||
print(
|
||||
f" -> {num_grouped_this_pass} agrupaciones realizadas (en redes no STL)."
|
||||
)
|
||||
|
||||
# --- Comprobar si se completó el procesamiento ---
|
||||
if not made_change_in_base_pass and not made_change_in_group_pass:
|
||||
print(f"\n--- No se hicieron más cambios en el pase {passes}. Proceso iterativo completado. ---")
|
||||
print(
|
||||
f"\n--- No se hicieron más cambios en el pase {passes}. Proceso iterativo completado. ---"
|
||||
)
|
||||
processing_complete = True
|
||||
else:
|
||||
print(f"--- Fin Pase {passes}: {num_sympy_processed_this_pass} proc SymPy, {num_grouped_this_pass} agrup. Continuando...")
|
||||
print(
|
||||
f"--- Fin Pase {passes}: {num_sympy_processed_this_pass} proc SymPy, {num_grouped_this_pass} agrup. Continuando..."
|
||||
)
|
||||
|
||||
# --- Comprobar límite de pases ---
|
||||
if passes == max_passes and not processing_complete:
|
||||
|
|
@ -376,46 +503,51 @@ def process_json_to_scl(json_filepath):
|
|||
# --- FIN BUCLE ITERATIVO ---
|
||||
|
||||
# --- Verificación Final (Ajustada para RAW_STL_CHUNK) ---
|
||||
print("\n--- Verificación Final de Instrucciones No Procesadas ---")
|
||||
print("\n--- Verificación Final de Instrucciones No Procesadas (FC/FB) ---")
|
||||
unprocessed_count = 0
|
||||
unprocessed_details = []
|
||||
# Añadir RAW_STL_CHUNK a los tipos ignorados
|
||||
ignored_types = ['raw_scl_chunk', 'unsupported_lang', 'raw_stl_chunk'] # Añadido raw_stl_chunk
|
||||
|
||||
ignored_types = [
|
||||
"raw_scl_chunk",
|
||||
"unsupported_lang",
|
||||
"raw_stl_chunk",
|
||||
] # Añadido raw_stl_chunk
|
||||
for network in data.get("networks", []):
|
||||
network_id = network.get("id", "Unknown ID")
|
||||
network_title = network.get("title", f"Network {network_id}")
|
||||
network_lang = network.get("language", "LAD") # Obtener lenguaje
|
||||
|
||||
# No verificar instrucciones dentro de redes STL, ya que no se procesan
|
||||
if network_lang == "STL":
|
||||
continue
|
||||
|
||||
for instruction in network.get("logic", []):
|
||||
instr_uid = instruction.get("instruction_uid", "Unknown UID")
|
||||
instr_type = instruction.get("type", "Unknown Type")
|
||||
is_grouped = instruction.get("grouped", False)
|
||||
|
||||
# Condición revisada para ignorar los chunks crudos
|
||||
if (not instr_type.endswith(SCL_SUFFIX) and
|
||||
"_error" not in instr_type and
|
||||
not is_grouped and
|
||||
instr_type.lower() not in ignored_types): # Verifica contra lista actualizada
|
||||
unprocessed_count += 1
|
||||
unprocessed_details.append(
|
||||
f" - Red '{network_title}' (ID: {network_id}, Lang: {network_lang}), "
|
||||
f"Instrucción UID: {instr_uid}, Tipo: '{instr_type}'"
|
||||
)
|
||||
|
||||
network_id = network.get("id", "Unknown ID")
|
||||
network_title = network.get("title", f"Network {network_id}")
|
||||
network_lang = network.get("language", "LAD")
|
||||
if network_lang == "STL":
|
||||
continue # No verificar redes STL
|
||||
for instruction in network.get("logic", []):
|
||||
instr_uid = instruction.get("instruction_uid", "Unknown UID")
|
||||
instr_type = instruction.get("type", "Unknown Type")
|
||||
is_grouped = instruction.get("grouped", False)
|
||||
if (
|
||||
not instr_type.endswith(SCL_SUFFIX)
|
||||
and "_error" not in instr_type
|
||||
and not is_grouped
|
||||
and instr_type.lower() not in ignored_types
|
||||
):
|
||||
unprocessed_count += 1
|
||||
unprocessed_details.append(
|
||||
f" - Red '{network_title}' (ID: {network_id}, Lang: {network_lang}), "
|
||||
f"Instrucción UID: {instr_uid}, Tipo: '{instr_type}'"
|
||||
)
|
||||
if unprocessed_count > 0:
|
||||
print(f"ADVERTENCIA: Se encontraron {unprocessed_count} instrucciones (no STL) que parecen no haber sido procesadas:")
|
||||
for detail in unprocessed_details: print(detail)
|
||||
print(
|
||||
f"ADVERTENCIA: Se encontraron {unprocessed_count} instrucciones (no STL) que parecen no haber sido procesadas:"
|
||||
)
|
||||
for detail in unprocessed_details:
|
||||
print(detail)
|
||||
else:
|
||||
print("INFO: Todas las instrucciones relevantes (no STL) parecen haber sido procesadas o agrupadas.")
|
||||
print(
|
||||
"INFO: Todas las instrucciones relevantes (no STL) parecen haber sido procesadas o agrupadas."
|
||||
)
|
||||
|
||||
# --- Guardar JSON Final ---
|
||||
output_filename = json_filepath.replace("_simplified.json", "_simplified_processed.json")
|
||||
print(f"\nGuardando JSON procesado en: {output_filename}")
|
||||
output_filename = json_filepath.replace(
|
||||
"_simplified.json", "_simplified_processed.json"
|
||||
)
|
||||
print(f"\nGuardando JSON procesado (FC/FB) en: {output_filename}")
|
||||
try:
|
||||
with open(output_filename, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, indent=4, ensure_ascii=False)
|
||||
|
|
@ -424,6 +556,7 @@ def process_json_to_scl(json_filepath):
|
|||
print(f"Error Crítico al guardar JSON procesado: {e}")
|
||||
traceback.print_exc()
|
||||
|
||||
|
||||
# --- Ejecución (sin cambios) ---
|
||||
if __name__ == "__main__":
|
||||
# Imports necesarios solo para la ejecución como script principal
|
||||
|
|
@ -436,43 +569,55 @@ if __name__ == "__main__":
|
|||
description="Process simplified JSON (_simplified.json) to embed SCL logic (SymPy version). Expects original XML filepath as argument."
|
||||
)
|
||||
parser.add_argument(
|
||||
"source_xml_filepath", # Argumento posicional obligatorio
|
||||
"source_xml_filepath", # Argumento posicional obligatorio
|
||||
help="Path to the original source XML file (passed from x0_main.py, used to derive JSON input name).",
|
||||
)
|
||||
args = parser.parse_args() # Parsea los argumentos de sys.argv
|
||||
args = parser.parse_args() # Parsea los argumentos de sys.argv
|
||||
|
||||
source_xml_file = args.source_xml_filepath # Obtiene la ruta del XML original
|
||||
source_xml_file = args.source_xml_filepath # Obtiene la ruta del XML original
|
||||
|
||||
# Verificar si el archivo XML original existe (como referencia, útil para depuración)
|
||||
# No es estrictamente necesario para la lógica aquí, pero ayuda a confirmar
|
||||
if not os.path.exists(source_xml_file):
|
||||
print(f"Advertencia (x2): Archivo XML original no encontrado: '{source_xml_file}', pero se intentará encontrar el JSON correspondiente.")
|
||||
# No salir necesariamente, pero es bueno saberlo.
|
||||
print(
|
||||
f"Advertencia (x2): Archivo XML original no encontrado: '{source_xml_file}', pero se intentará encontrar el JSON correspondiente."
|
||||
)
|
||||
# No salir necesariamente, pero es bueno saberlo.
|
||||
|
||||
# Derivar nombre del archivo JSON de entrada (_simplified.json)
|
||||
xml_filename_base = os.path.splitext(os.path.basename(source_xml_file))[0]
|
||||
# Asumir que el JSON simplificado está en el mismo directorio que el XML original
|
||||
input_dir = os.path.dirname(source_xml_file) # Directorio del XML original
|
||||
input_dir = os.path.dirname(source_xml_file) # Directorio del XML original
|
||||
input_json_file = os.path.join(input_dir, f"{xml_filename_base}_simplified.json")
|
||||
|
||||
# Determinar el nombre esperado del archivo JSON procesado de salida
|
||||
output_json_file = os.path.join(input_dir, f"{xml_filename_base}_simplified_processed.json")
|
||||
|
||||
print(f"(x2) Procesando: '{os.path.relpath(input_json_file)}' -> '{os.path.relpath(output_json_file)}'")
|
||||
output_json_file = os.path.join(
|
||||
input_dir, f"{xml_filename_base}_simplified_processed.json"
|
||||
)
|
||||
|
||||
print(
|
||||
f"(x2) Procesando: '{os.path.relpath(input_json_file)}' -> '{os.path.relpath(output_json_file)}'"
|
||||
)
|
||||
|
||||
# Verificar si el archivo JSON de entrada (_simplified.json) EXISTE antes de procesar
|
||||
if not os.path.exists(input_json_file):
|
||||
print(f"Error Fatal (x2): El archivo de entrada JSON simplificado no existe: '{input_json_file}'")
|
||||
print(f"Asegúrate de que 'x1_to_json.py' se ejecutó correctamente para '{os.path.relpath(source_xml_file)}'.")
|
||||
sys.exit(1) # Salir si el archivo necesario no está
|
||||
print(
|
||||
f"Error Fatal (x2): El archivo de entrada JSON simplificado no existe: '{input_json_file}'"
|
||||
)
|
||||
print(
|
||||
f"Asegúrate de que 'x1_to_json.py' se ejecutó correctamente para '{os.path.relpath(source_xml_file)}'."
|
||||
)
|
||||
sys.exit(1) # Salir si el archivo necesario no está
|
||||
else:
|
||||
# Llamar a la función principal de procesamiento del script
|
||||
# Asumiendo que tu función principal se llama process_json_to_scl(input_json_path)
|
||||
try:
|
||||
process_json_to_scl(input_json_file)
|
||||
except Exception as e:
|
||||
print(f"Error Crítico (x2) durante el procesamiento de '{input_json_file}': {e}")
|
||||
print(
|
||||
f"Error Crítico (x2) durante el procesamiento de '{input_json_file}': {e}"
|
||||
)
|
||||
import traceback
|
||||
|
||||
traceback.print_exc()
|
||||
sys.exit(1) # Salir con error si la función principal falla
|
||||
sys.exit(1) # Salir con error si la función principal falla
|
||||
|
|
|
|||
|
|
@ -5,290 +5,481 @@ import os
|
|||
import re
|
||||
import argparse
|
||||
import sys
|
||||
import traceback # Importar traceback para errores
|
||||
import traceback # Importar traceback para errores
|
||||
|
||||
# --- Importar Utilidades y Constantes (Asumiendo ubicación) ---
|
||||
try:
|
||||
# Intenta importar desde el paquete de procesadores si está estructurado así
|
||||
from processors.processor_utils import format_variable_name
|
||||
|
||||
# Definir SCL_SUFFIX aquí o importarlo si está centralizado
|
||||
SCL_SUFFIX = "_sympy_processed" # Asegúrate que coincida con x2_process.py
|
||||
GROUPED_COMMENT = "// Logic included in grouped IF" # Opcional, si se usa para filtrar
|
||||
SCL_SUFFIX = "_sympy_processed" # Asegúrate que coincida con x2_process.py
|
||||
GROUPED_COMMENT = (
|
||||
"// Logic included in grouped IF" # Opcional, si se usa para filtrar
|
||||
)
|
||||
except ImportError:
|
||||
print("Advertencia: No se pudo importar 'format_variable_name' desde processors.processor_utils.")
|
||||
print("Usando una implementación local básica (¡PUEDE FALLAR CON NOMBRES COMPLEJOS!).")
|
||||
print(
|
||||
"Advertencia: No se pudo importar 'format_variable_name' desde processors.processor_utils."
|
||||
)
|
||||
print(
|
||||
"Usando una implementación local básica (¡PUEDE FALLAR CON NOMBRES COMPLEJOS!)."
|
||||
)
|
||||
|
||||
# Implementación local BÁSICA como fallback (MENOS RECOMENDADA)
|
||||
def format_variable_name(name):
|
||||
if not name: return "_INVALID_NAME_"
|
||||
if name.startswith('"') and name.endswith('"'): return name # Mantener comillas
|
||||
if not name:
|
||||
return "_INVALID_NAME_"
|
||||
if name.startswith('"') and name.endswith('"'):
|
||||
return name # Mantener comillas
|
||||
prefix = "#" if name.startswith("#") else ""
|
||||
if prefix: name = name[1:]
|
||||
if name and name[0].isdigit(): name = "_" + name
|
||||
if prefix:
|
||||
name = name[1:]
|
||||
if name and name[0].isdigit():
|
||||
name = "_" + name
|
||||
name = re.sub(r"[^a-zA-Z0-9_]", "_", name)
|
||||
return prefix + name
|
||||
|
||||
SCL_SUFFIX = "_sympy_processed"
|
||||
GROUPED_COMMENT = "// Logic included in grouped IF"
|
||||
|
||||
|
||||
# --- Función Principal de Generación SCL ---
|
||||
# para formatear valores iniciales
|
||||
def format_scl_start_value(value, datatype):
|
||||
"""Formatea un valor para la inicialización SCL según el tipo."""
|
||||
if value is None:
|
||||
return None
|
||||
datatype_lower = datatype.lower() if datatype else ""
|
||||
value_str = str(value)
|
||||
|
||||
if "bool" in datatype_lower:
|
||||
return "TRUE" if value_str.lower() == "true" else "FALSE"
|
||||
elif "string" in datatype_lower:
|
||||
escaped_value = value_str.replace("'", "''")
|
||||
if escaped_value.startswith("'") and escaped_value.endswith("'"):
|
||||
escaped_value = escaped_value[1:-1]
|
||||
return f"'{escaped_value}'"
|
||||
elif "char" in datatype_lower: # Añadido Char
|
||||
escaped_value = value_str.replace("'", "''")
|
||||
if escaped_value.startswith("'") and escaped_value.endswith("'"):
|
||||
escaped_value = escaped_value[1:-1]
|
||||
return f"'{escaped_value}'"
|
||||
elif any(
|
||||
t in datatype_lower
|
||||
for t in [
|
||||
"int",
|
||||
"byte",
|
||||
"word",
|
||||
"dint",
|
||||
"dword",
|
||||
"lint",
|
||||
"lword",
|
||||
"sint",
|
||||
"usint",
|
||||
"uint",
|
||||
"udint",
|
||||
"ulint",
|
||||
]
|
||||
): # Ampliado
|
||||
try:
|
||||
return str(int(value_str))
|
||||
except ValueError:
|
||||
if re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", value_str):
|
||||
return value_str
|
||||
return f"'{value_str}'" # O como string si no es entero ni símbolo
|
||||
elif "real" in datatype_lower or "lreal" in datatype_lower:
|
||||
try:
|
||||
f_val = float(value_str)
|
||||
s_val = str(f_val)
|
||||
if "." not in s_val and "e" not in s_val.lower():
|
||||
s_val += ".0"
|
||||
return s_val
|
||||
except ValueError:
|
||||
if re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", value_str):
|
||||
return value_str
|
||||
return f"'{value_str}'"
|
||||
elif "time" in datatype_lower: # Añadido Time, S5Time, LTime
|
||||
# Quitar T#, LT#, S5T# si existen
|
||||
prefix = ""
|
||||
if value_str.upper().startswith("T#"):
|
||||
prefix = "T#"
|
||||
value_str = value_str[2:]
|
||||
elif value_str.upper().startswith("LT#"):
|
||||
prefix = "LT#"
|
||||
value_str = value_str[3:]
|
||||
elif value_str.upper().startswith("S5T#"):
|
||||
prefix = "S5T#"
|
||||
value_str = value_str[4:]
|
||||
# Devolver con el prefijo correcto o T# por defecto si no había
|
||||
if prefix:
|
||||
return f"{prefix}{value_str}"
|
||||
elif "s5time" in datatype_lower:
|
||||
return f"S5T#{value_str}"
|
||||
elif "ltime" in datatype_lower:
|
||||
return f"LT#{value_str}"
|
||||
else:
|
||||
return f"T#{value_str}" # Default a TIME
|
||||
elif "date" in datatype_lower: # Añadido Date, DT, TOD
|
||||
if value_str.upper().startswith("D#"):
|
||||
return value_str
|
||||
elif "dt" in datatype_lower or "date_and_time" in datatype_lower:
|
||||
if value_str.upper().startswith("DT#"):
|
||||
return value_str
|
||||
else:
|
||||
return f"DT#{value_str}" # Añadir prefijo DT#
|
||||
elif "tod" in datatype_lower or "time_of_day" in datatype_lower:
|
||||
if value_str.upper().startswith("TOD#"):
|
||||
return value_str
|
||||
else:
|
||||
return f"TOD#{value_str}" # Añadir prefijo TOD#
|
||||
else:
|
||||
return f"D#{value_str}" # Default a Date
|
||||
# Fallback genérico
|
||||
else:
|
||||
if re.match(
|
||||
r'^[a-zA-Z_][a-zA-Z0-9_."#\[\]]+$', value_str
|
||||
): # Permitir más caracteres en símbolos/tipos
|
||||
# Si es un UDT o Struct complejo, podría venir con comillas, quitarlas
|
||||
if value_str.startswith('"') and value_str.endswith('"'):
|
||||
return value_str[1:-1]
|
||||
return value_str
|
||||
else:
|
||||
escaped_value = value_str.replace("'", "''")
|
||||
return f"'{escaped_value}'"
|
||||
|
||||
|
||||
# --- NUEVA FUNCIÓN RECURSIVA para generar declaraciones SCL (VAR/STRUCT/ARRAY) ---
|
||||
def generate_scl_declarations(variables, indent_level=1):
|
||||
"""Genera las líneas SCL para declarar variables, structs y arrays."""
|
||||
scl_lines = []
|
||||
indent = " " * indent_level
|
||||
for var in variables:
|
||||
var_name_scl = format_variable_name(var.get("name"))
|
||||
var_dtype_raw = var.get("datatype", "VARIANT")
|
||||
# Limpiar comillas de tipos de datos UDT ("MyType" -> MyType)
|
||||
var_dtype = (
|
||||
var_dtype_raw.strip('"')
|
||||
if var_dtype_raw.startswith('"') and var_dtype_raw.endswith('"')
|
||||
else var_dtype_raw
|
||||
)
|
||||
|
||||
var_comment = var.get("comment")
|
||||
start_value = var.get("start_value")
|
||||
children = var.get("children") # Para structs
|
||||
array_elements = var.get("array_elements") # Para arrays
|
||||
|
||||
# Manejar tipos de datos Array especiales
|
||||
array_match = re.match(r"(Array\[.*\]\s+of\s+)(.*)", var_dtype, re.IGNORECASE)
|
||||
base_type_for_init = var_dtype
|
||||
declaration_dtype = var_dtype
|
||||
if array_match:
|
||||
array_prefix = array_match.group(1)
|
||||
base_type_raw = array_match.group(2).strip()
|
||||
# Limpiar comillas del tipo base del array
|
||||
base_type_for_init = (
|
||||
base_type_raw.strip('"')
|
||||
if base_type_raw.startswith('"') and base_type_raw.endswith('"')
|
||||
else base_type_raw
|
||||
)
|
||||
declaration_dtype = (
|
||||
f'{array_prefix}"{base_type_for_init}"'
|
||||
if '"' not in base_type_raw
|
||||
else f"{array_prefix}{base_type_raw}"
|
||||
) # Reconstruir con comillas si es UDT
|
||||
|
||||
# Reconstruir declaración con comillas si es UDT y no array
|
||||
elif (
|
||||
not array_match and var_dtype != base_type_for_init
|
||||
): # Es un tipo que necesita comillas (UDT)
|
||||
declaration_dtype = f'"{var_dtype}"'
|
||||
|
||||
declaration_line = f"{indent}{var_name_scl} : {declaration_dtype}"
|
||||
init_value = None
|
||||
|
||||
# ---- Arrays ----
|
||||
if array_elements:
|
||||
# Ordenar índices (asumiendo que son numéricos)
|
||||
try:
|
||||
sorted_indices = sorted(array_elements.keys(), key=int)
|
||||
except ValueError:
|
||||
sorted_indices = sorted(
|
||||
array_elements.keys()
|
||||
) # Fallback a orden alfabético
|
||||
|
||||
init_values = [
|
||||
format_scl_start_value(array_elements[idx], base_type_for_init)
|
||||
for idx in sorted_indices
|
||||
]
|
||||
valid_inits = [v for v in init_values if v is not None]
|
||||
if valid_inits:
|
||||
init_value = f"[{', '.join(valid_inits)}]"
|
||||
|
||||
# ---- Structs ----
|
||||
elif children:
|
||||
# No añadir comentario // Struct aquí, es redundante
|
||||
scl_lines.append(declaration_line) # Añadir línea de declaración base
|
||||
scl_lines.append(f"{indent}STRUCT")
|
||||
scl_lines.extend(generate_scl_declarations(children, indent_level + 1))
|
||||
scl_lines.append(f"{indent}END_STRUCT;")
|
||||
if var_comment:
|
||||
scl_lines.append(f"{indent}// {var_comment}")
|
||||
scl_lines.append("") # Línea extra
|
||||
continue # Saltar resto para Struct
|
||||
|
||||
# ---- Tipos Simples ----
|
||||
else:
|
||||
if start_value is not None:
|
||||
init_value = format_scl_start_value(start_value, var_dtype)
|
||||
|
||||
# Añadir inicialización si existe
|
||||
if init_value:
|
||||
declaration_line += f" := {init_value}"
|
||||
declaration_line += ";"
|
||||
if var_comment:
|
||||
declaration_line += f" // {var_comment}"
|
||||
scl_lines.append(declaration_line)
|
||||
|
||||
return scl_lines
|
||||
|
||||
|
||||
# --- Función Principal de Generación SCL ---
|
||||
def generate_scl(processed_json_filepath, output_scl_filepath):
|
||||
"""Genera un archivo SCL a partir del JSON procesado por x2_process (versión SymPy)."""
|
||||
"""Genera un archivo SCL a partir del JSON procesado (FC/FB o DB)."""
|
||||
|
||||
if not os.path.exists(processed_json_filepath):
|
||||
print(f"Error: Archivo JSON procesado no encontrado en '{processed_json_filepath}'")
|
||||
print(
|
||||
f"Error: Archivo JSON procesado no encontrado en '{processed_json_filepath}'"
|
||||
)
|
||||
return
|
||||
|
||||
print(f"Cargando JSON procesado desde: {processed_json_filepath}")
|
||||
try:
|
||||
with open(processed_json_filepath, 'r', encoding='utf-8') as f:
|
||||
with open(processed_json_filepath, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
except Exception as e:
|
||||
print(f"Error al cargar o parsear JSON: {e}")
|
||||
traceback.print_exc()
|
||||
return
|
||||
|
||||
# --- Extracción de Información del Bloque ---
|
||||
block_name = data.get('block_name', 'UnknownBlock')
|
||||
block_number = data.get('block_number')
|
||||
block_lang_original = data.get('language', 'LAD') # Lenguaje original
|
||||
# Determinar tipo de bloque SCL (Asumir FB si no se especifica)
|
||||
# Idealmente, x1_to_json.py guardaría esto en data['block_type_scl'] = 'FC' o 'FB'
|
||||
block_type_scl = data.get('block_type_scl', 'FUNCTION_BLOCK')
|
||||
block_comment = data.get('block_comment', '')
|
||||
|
||||
# Usar format_variable_name para el nombre del bloque en SCL
|
||||
scl_block_name = format_variable_name(block_name)
|
||||
print(f"Generando SCL para {block_type_scl}: {scl_block_name} (Original: {block_name})")
|
||||
|
||||
# --- Identificación de Variables Temporales y Estáticas ---
|
||||
# La detección basada en regex sobre el SCL final debería seguir funcionando
|
||||
temp_vars = set()
|
||||
stat_vars = set()
|
||||
# Regex mejorado para capturar variables temporales que empiezan con # o _temp_
|
||||
# y estáticas (si usas un prefijo como 'stat_' o para bits de memoria de flanco)
|
||||
temp_pattern = re.compile(r'"?#(_temp_[a-zA-Z0-9_]+)"?|"?(_temp_[a-zA-Z0-9_]+)"?') # Captura con o sin #
|
||||
stat_pattern = re.compile(r'"?(stat_[a-zA-Z0-9_]+)"?') # Para memorias de flanco si usan prefijo 'stat_'
|
||||
|
||||
edge_memory_bits = set() # Para detectar bits de memoria de flanco por nombre
|
||||
|
||||
for network in data.get('networks', []):
|
||||
for instruction in network.get('logic', []):
|
||||
scl_code = instruction.get('scl', '')
|
||||
# Buscar también en _edge_mem_update_scl si existe
|
||||
edge_update_code = instruction.get('_edge_mem_update_scl','')
|
||||
code_to_scan = (scl_code if scl_code else '') + '\n' + (edge_update_code if edge_update_code else '')
|
||||
|
||||
if code_to_scan:
|
||||
# Buscar #_temp_... o _temp_...
|
||||
found_temps = temp_pattern.findall(code_to_scan)
|
||||
for temp_tuple in found_temps:
|
||||
# findall devuelve tuplas por los grupos de captura, tomar el no vacío
|
||||
temp_name = next((t for t in temp_tuple if t), None)
|
||||
if temp_name:
|
||||
temp_vars.add("#"+temp_name if not temp_name.startswith("#") else temp_name) # Asegurar que empiece con #
|
||||
|
||||
# Buscar estáticas (ej: stat_...)
|
||||
found_stats = stat_pattern.findall(code_to_scan)
|
||||
stat_vars.update(found_stats)
|
||||
|
||||
# Identificar explícitamente bits de memoria usados por PBox/NBox
|
||||
# Asumiendo que el nombre se guarda en el JSON (requiere ajuste en x1/x2)
|
||||
# if instruction.get("type","").startswith(("PBox", "NBox")):
|
||||
# mem_bit_info = instruction.get("inputs", {}).get("bit")
|
||||
# if mem_bit_info and mem_bit_info.get("type") == "variable":
|
||||
# edge_memory_bits.add(format_variable_name(mem_bit_info.get("name")))
|
||||
|
||||
|
||||
print(f"Variables temporales (#_temp_...) detectadas: {len(temp_vars)}")
|
||||
# Si se detectan memorias de flanco, añadirlas a stat_vars si no tienen prefijo 'stat_'
|
||||
# stat_vars.update(edge_memory_bits - stat_vars) # Añadir solo las nuevas
|
||||
print(f"Variables estáticas (stat_...) detectadas: {len(stat_vars)}")
|
||||
|
||||
# --- Construcción del String SCL ---
|
||||
# --- Extracción de Información del Bloque (Común) ---
|
||||
block_name = data.get("block_name", "UnknownBlock")
|
||||
block_number = data.get("block_number")
|
||||
block_lang_original = data.get("language", "Unknown") # Será "DB" para Data Blocks
|
||||
block_type = data.get("block_type", "Unknown") # FC, FB, GlobalDB
|
||||
block_comment = data.get("block_comment", "")
|
||||
scl_block_name = format_variable_name(block_name) # Nombre SCL seguro
|
||||
print(
|
||||
f"Generando SCL para: {block_type} '{scl_block_name}' (Original: {block_name}, Lang: {block_lang_original})"
|
||||
)
|
||||
scl_output = []
|
||||
|
||||
# Cabecera del Bloque
|
||||
scl_output.append(f"// Block Name (Original): {block_name}")
|
||||
if block_number: scl_output.append(f"// Block Number: {block_number}")
|
||||
scl_output.append(f"// Original Language: {block_lang_original}")
|
||||
if block_comment: scl_output.append(f"// Block Comment: {block_comment}")
|
||||
scl_output.append("")
|
||||
scl_output.append(f"{block_type_scl} \"{scl_block_name}\"")
|
||||
scl_output.append("{ S7_Optimized_Access := 'TRUE' }")
|
||||
scl_output.append("VERSION : 0.1")
|
||||
scl_output.append("")
|
||||
# --- GENERACIÓN PARA DATA BLOCK (DB) ---
|
||||
if block_lang_original == "DB":
|
||||
print("Modo de generación: DATA_BLOCK")
|
||||
scl_output.append(f"// Block Type: {block_type}")
|
||||
scl_output.append(f"// Block Name (Original): {block_name}")
|
||||
if block_number:
|
||||
scl_output.append(f"// Block Number: {block_number}")
|
||||
if block_comment:
|
||||
scl_output.append(f"// Block Comment: {block_comment}")
|
||||
scl_output.append("")
|
||||
scl_output.append(f'DATA_BLOCK "{scl_block_name}"')
|
||||
scl_output.append("{ S7_Optimized_Access := 'TRUE' }") # Asumir optimizado
|
||||
scl_output.append("VERSION : 0.1")
|
||||
scl_output.append("")
|
||||
interface_data = data.get("interface", {})
|
||||
static_vars = interface_data.get("Static", [])
|
||||
if static_vars:
|
||||
scl_output.append("VAR")
|
||||
scl_output.extend(generate_scl_declarations(static_vars, indent_level=1))
|
||||
scl_output.append("END_VAR")
|
||||
scl_output.append("")
|
||||
else:
|
||||
print(
|
||||
"Advertencia: No se encontró sección 'Static' o está vacía en la interfaz del DB."
|
||||
)
|
||||
scl_output.append("VAR")
|
||||
scl_output.append("END_VAR")
|
||||
scl_output.append("")
|
||||
scl_output.append("BEGIN")
|
||||
scl_output.append("")
|
||||
scl_output.append("END_DATA_BLOCK")
|
||||
|
||||
# Declaraciones de Interfaz (Implementación básica)
|
||||
interface_sections = ["Input", "Output", "InOut", "Static", "Temp", "Constant", "Return"]
|
||||
interface_data = data.get('interface', {})
|
||||
# --- GENERACIÓN PARA FUNCTION BLOCK / FUNCTION (FC/FB) ---
|
||||
else:
|
||||
print("Modo de generación: FUNCTION_BLOCK / FUNCTION")
|
||||
scl_block_keyword = "FUNCTION_BLOCK" if block_type == "FB" else "FUNCTION"
|
||||
# Cabecera del Bloque
|
||||
scl_output.append(f"// Block Type: {block_type}")
|
||||
scl_output.append(f"// Block Name (Original): {block_name}")
|
||||
if block_number:
|
||||
scl_output.append(f"// Block Number: {block_number}")
|
||||
scl_output.append(f"// Original Language: {block_lang_original}")
|
||||
if block_comment:
|
||||
scl_output.append(f"// Block Comment: {block_comment}")
|
||||
scl_output.append("")
|
||||
# Manejar tipo de retorno para FUNCTION
|
||||
return_type = "Void" # Default
|
||||
interface_data = data.get("interface", {})
|
||||
if scl_block_keyword == "FUNCTION" and interface_data.get("Return"):
|
||||
return_member = interface_data["Return"][
|
||||
0
|
||||
] # Asumir un solo valor de retorno
|
||||
return_type_raw = return_member.get("datatype", "Void")
|
||||
return_type = (
|
||||
return_type_raw.strip('"')
|
||||
if return_type_raw.startswith('"') and return_type_raw.endswith('"')
|
||||
else return_type_raw
|
||||
)
|
||||
# Añadir comillas si es UDT
|
||||
if return_type != return_type_raw:
|
||||
return_type = f'"{return_type}"'
|
||||
|
||||
for section_name in interface_sections:
|
||||
scl_section_name = section_name
|
||||
# Ajustar nombres de sección para SCL (Static -> STAT, Temp -> TEMP)
|
||||
if section_name == "Static": scl_section_name = "STAT"
|
||||
if section_name == "Temp": scl_section_name = "TEMP" # Usar VAR_TEMP para variables #temp
|
||||
|
||||
vars_in_section = interface_data.get(section_name, [])
|
||||
# No declarar VAR_TEMP aquí, se hará después con las detectadas/originales
|
||||
if section_name == "Temp": continue
|
||||
|
||||
# No declarar VAR_STAT aquí si ya lo hacemos abajo con las detectadas
|
||||
if section_name == "Static" and stat_vars: continue
|
||||
|
||||
|
||||
if vars_in_section or (section_name == "Static" and stat_vars): # Incluir STAT si hay detectadas
|
||||
# Usar VAR para Input/Output/InOut/Constant/Return
|
||||
var_keyword = "VAR" if section_name != "Static" else "VAR_STAT"
|
||||
scl_output.append(f"{var_keyword}_{section_name.upper()}")
|
||||
|
||||
for var in vars_in_section:
|
||||
var_name = var.get('name')
|
||||
var_dtype = var.get('datatype', 'VARIANT') # Default a VARIANT
|
||||
if var_name:
|
||||
# Usar format_variable_name CORRECTO
|
||||
scl_name = format_variable_name(var_name)
|
||||
scl_output.append(f" {scl_name} : {var_dtype};")
|
||||
|
||||
# Declarar stat_vars detectadas si esta es la sección STAT
|
||||
if section_name == "Static" and stat_vars:
|
||||
for var_name in sorted(list(stat_vars)):
|
||||
# Asumir Bool para stat_, podría necesitar inferencia
|
||||
scl_output.append(f" {format_variable_name(var_name)} : Bool; // Auto-detected STAT")
|
||||
|
||||
scl_output.append("END_VAR")
|
||||
scl_output.append("")
|
||||
|
||||
# Declaraciones Estáticas (Si no estaban en la interfaz y se detectaron)
|
||||
# Esto es redundante si la sección VAR_STAT ya se generó arriba
|
||||
# if stat_vars and not interface_data.get("Static"):
|
||||
# scl_output.append("VAR_STAT")
|
||||
# for var_name in sorted(list(stat_vars)):
|
||||
# scl_output.append(f" {format_variable_name(var_name)} : Bool; // Auto-detected STAT")
|
||||
# scl_output.append("END_VAR")
|
||||
# scl_output.append("")
|
||||
|
||||
|
||||
# Declaraciones Temporales (Interfaz Temp + _temp_ detectadas)
|
||||
scl_output.append("VAR_TEMP")
|
||||
declared_temps = set()
|
||||
interface_temps = interface_data.get('Temp', [])
|
||||
if interface_temps:
|
||||
for var in interface_temps:
|
||||
var_name = var.get('name')
|
||||
var_dtype = var.get('datatype', 'VARIANT')
|
||||
if var_name:
|
||||
scl_name = format_variable_name(var_name)
|
||||
scl_output.append(f" {scl_name} : {var_dtype};")
|
||||
declared_temps.add(scl_name) # Marcar como declarada
|
||||
|
||||
# Declarar las _temp_ generadas si no estaban ya en la interfaz Temp
|
||||
if temp_vars:
|
||||
for var_name in sorted(list(temp_vars)):
|
||||
scl_name = format_variable_name(var_name) # #_temp_...
|
||||
if scl_name not in declared_temps:
|
||||
# Inferencia básica de tipo
|
||||
inferred_type = "Bool" # Asumir Bool para la mayoría de temps de lógica
|
||||
# Se podría mejorar si los procesadores añadieran info de tipo
|
||||
scl_output.append(f" {scl_name} : {inferred_type}; // Auto-generated temporary")
|
||||
declared_temps.add(scl_name)
|
||||
scl_output.append("END_VAR")
|
||||
scl_output.append("")
|
||||
|
||||
# Cuerpo del Bloque
|
||||
scl_output.append("BEGIN")
|
||||
scl_output.append("")
|
||||
|
||||
# Iterar por redes y lógica
|
||||
for i, network in enumerate(data.get('networks', [])):
|
||||
network_title = network.get('title', f'Network {network.get("id")}')
|
||||
network_comment = network.get('comment', '')
|
||||
network_lang = network.get('language', 'LAD') # O el lenguaje original
|
||||
|
||||
scl_output.append(f" // Network {i+1}: {network_title} (Original Language: {network_lang})")
|
||||
if network_comment:
|
||||
for line in network_comment.splitlines():
|
||||
scl_output.append(f" // {line}")
|
||||
scl_output.append(
|
||||
f'{scl_block_keyword} "{scl_block_name}" : {return_type}'
|
||||
if scl_block_keyword == "FUNCTION"
|
||||
else f'{scl_block_keyword} "{scl_block_name}"'
|
||||
)
|
||||
scl_output.append("{ S7_Optimized_Access := 'TRUE' }")
|
||||
scl_output.append("VERSION : 0.1")
|
||||
scl_output.append("")
|
||||
|
||||
network_has_code = False
|
||||
# Declaraciones de Interfaz FC/FB
|
||||
section_order = [
|
||||
"Input",
|
||||
"Output",
|
||||
"InOut",
|
||||
"Static",
|
||||
"Temp",
|
||||
"Constant",
|
||||
] # Return ya está en cabecera
|
||||
declared_temps = set()
|
||||
for section_name in section_order:
|
||||
vars_in_section = interface_data.get(section_name, [])
|
||||
if vars_in_section:
|
||||
scl_section_keyword = f"VAR_{section_name.upper()}"
|
||||
if section_name == "Static":
|
||||
scl_section_keyword = "VAR_STAT"
|
||||
if section_name == "Temp":
|
||||
scl_section_keyword = "VAR_TEMP"
|
||||
if section_name == "Constant":
|
||||
scl_section_keyword = "CONSTANT"
|
||||
scl_output.append(scl_section_keyword)
|
||||
scl_output.extend(
|
||||
generate_scl_declarations(vars_in_section, indent_level=1)
|
||||
)
|
||||
if section_name == "Temp":
|
||||
declared_temps.update(
|
||||
format_variable_name(v.get("name"))
|
||||
for v in vars_in_section
|
||||
if v.get("name")
|
||||
)
|
||||
scl_output.append("END_VAR")
|
||||
scl_output.append("")
|
||||
|
||||
# --- NUEVO MANEJO STL con formato Markdown ---
|
||||
if network_lang == "STL":
|
||||
network_has_code = True # Marcar que la red tiene contenido
|
||||
if network.get('logic') and isinstance(network['logic'], list) and len(network['logic']) > 0:
|
||||
stl_chunk = network['logic'][0]
|
||||
if stl_chunk.get("type") == "RAW_STL_CHUNK" and "stl" in stl_chunk:
|
||||
raw_stl_code = stl_chunk["stl"]
|
||||
# Añadir marcador de inicio (como comentario SCL para evitar errores)
|
||||
scl_output.append(f" {'//'} ```STL") # Doble '//' para asegurar que sea comentario
|
||||
# Escribir el código STL crudo, indentado
|
||||
# Declaraciones VAR_TEMP adicionales detectadas
|
||||
temp_vars = set()
|
||||
temp_pattern = re.compile(
|
||||
r'"?#(_temp_[a-zA-Z0-9_]+)"?|"?(_temp_[a-zA-Z0-9_]+)"?'
|
||||
)
|
||||
for network in data.get("networks", []):
|
||||
for instruction in network.get("logic", []):
|
||||
scl_code = instruction.get("scl", "")
|
||||
edge_update_code = instruction.get("_edge_mem_update_scl", "")
|
||||
code_to_scan = (
|
||||
(scl_code if scl_code else "")
|
||||
+ "\n"
|
||||
+ (edge_update_code if edge_update_code else "")
|
||||
)
|
||||
if code_to_scan:
|
||||
found_temps = temp_pattern.findall(code_to_scan)
|
||||
for temp_tuple in found_temps:
|
||||
temp_name = next((t for t in temp_tuple if t), None)
|
||||
if temp_name:
|
||||
temp_vars.add(
|
||||
"#" + temp_name
|
||||
if not temp_name.startswith("#")
|
||||
else temp_name
|
||||
)
|
||||
additional_temps = sorted(list(temp_vars - declared_temps))
|
||||
if additional_temps:
|
||||
if not interface_data.get("Temp"):
|
||||
scl_output.append("VAR_TEMP")
|
||||
for var_name in additional_temps:
|
||||
scl_name = format_variable_name(var_name)
|
||||
inferred_type = "Bool" # Asumir Bool
|
||||
scl_output.append(
|
||||
f" {scl_name} : {inferred_type}; // Auto-generated temporary"
|
||||
)
|
||||
if not interface_data.get("Temp"):
|
||||
scl_output.append("END_VAR")
|
||||
scl_output.append("")
|
||||
|
||||
# Cuerpo del Bloque FC/FB
|
||||
scl_output.append("BEGIN")
|
||||
scl_output.append("")
|
||||
# Iterar por redes y lógica (como antes, incluyendo manejo STL Markdown)
|
||||
for i, network in enumerate(data.get("networks", [])):
|
||||
network_title = network.get("title", f'Network {network.get("id")}')
|
||||
network_comment = network.get("comment", "")
|
||||
network_lang = network.get("language", "LAD")
|
||||
scl_output.append(
|
||||
f" // Network {i+1}: {network_title} (Original Language: {network_lang})"
|
||||
)
|
||||
if network_comment:
|
||||
for line in network_comment.splitlines():
|
||||
scl_output.append(f" // {line}")
|
||||
scl_output.append("")
|
||||
network_has_code = False
|
||||
if network_lang == "STL":
|
||||
network_has_code = True
|
||||
if (
|
||||
network.get("logic")
|
||||
and network["logic"][0].get("type") == "RAW_STL_CHUNK"
|
||||
):
|
||||
raw_stl_code = network["logic"][0].get(
|
||||
"stl", "// ERROR: STL code missing"
|
||||
)
|
||||
scl_output.append(f" {'//'} ```STL")
|
||||
for stl_line in raw_stl_code.splitlines():
|
||||
# Añadir indentación estándar de SCL
|
||||
scl_output.append(f" {stl_line}") # <-- STL sin comentar
|
||||
# Añadir marcador de fin (como comentario SCL)
|
||||
scl_output.append(f" {stl_line}")
|
||||
scl_output.append(f" {'//'} ```")
|
||||
else:
|
||||
scl_output.append(" // ERROR: Contenido STL inesperado en JSON.")
|
||||
else:
|
||||
scl_output.append(" // ERROR: No se encontró lógica STL en JSON para esta red.")
|
||||
scl_output.append("") # Línea en blanco después de la red STL
|
||||
# --- FIN NUEVO MANEJO STL con formato Markdown ---
|
||||
else:
|
||||
|
||||
# Iterar sobre la 'logica' de la red
|
||||
for instruction in network.get('logic', []):
|
||||
instruction_type = instruction.get("type", "")
|
||||
scl_code = instruction.get('scl', "") # Obtener SCL generado por x2
|
||||
|
||||
# Saltar instrucciones agrupadas
|
||||
if instruction.get("grouped", False):
|
||||
continue
|
||||
|
||||
# Escribir SCL si es un tipo procesado y tiene código relevante
|
||||
# (Ignorar comentarios de depuración de SymPy)
|
||||
if instruction_type.endswith(SCL_SUFFIX) and scl_code:
|
||||
is_internal_sympy_comment_only = scl_code.strip().startswith("// SymPy") or \
|
||||
scl_code.strip().startswith("// PBox SymPy processed") or \
|
||||
scl_code.strip().startswith("// NBox SymPy processed")
|
||||
# O podría ser más genérico: ignorar cualquier línea que solo sea comentario SCL
|
||||
is_only_comment = all(line.strip().startswith("//") for line in scl_code.splitlines())
|
||||
|
||||
|
||||
# Escribir solo si NO es un comentario interno de SymPy O si es un bloque IF (que sí debe escribirse)
|
||||
if not is_only_comment or scl_code.strip().startswith("IF"):
|
||||
network_has_code = True
|
||||
for line in scl_code.splitlines():
|
||||
# Añadir indentación estándar
|
||||
scl_output.append(f" {line}")
|
||||
|
||||
# Incluir también tipos especiales directamente
|
||||
elif instruction_type in ["RAW_SCL_CHUNK", "UNSUPPORTED_LANG"] and scl_code:
|
||||
network_has_code = True
|
||||
for line in scl_code.splitlines():
|
||||
scl_output.append(f" {line}") # Indentar
|
||||
|
||||
# Podríamos añadir comentarios para errores si se desea
|
||||
# elif "_error" in instruction_type:
|
||||
# network_has_code = True
|
||||
# scl_output.append(f" // ERROR processing instruction UID {instruction.get('instruction_uid')}: {instruction.get('scl', 'No details')}")
|
||||
|
||||
|
||||
scl_output.append(" // ERROR: Contenido STL inesperado.")
|
||||
else: # LAD, FBD, SCL, etc.
|
||||
for instruction in network.get("logic", []):
|
||||
instruction_type = instruction.get("type", "")
|
||||
scl_code = instruction.get("scl", "")
|
||||
is_grouped = instruction.get("grouped", False)
|
||||
if is_grouped:
|
||||
continue
|
||||
if (
|
||||
instruction_type.endswith(SCL_SUFFIX)
|
||||
or instruction_type in ["RAW_SCL_CHUNK", "UNSUPPORTED_LANG"]
|
||||
) and scl_code:
|
||||
is_only_comment = all(
|
||||
line.strip().startswith("//")
|
||||
for line in scl_code.splitlines()
|
||||
if line.strip()
|
||||
)
|
||||
is_if_block = scl_code.strip().startswith("IF")
|
||||
if not is_only_comment or is_if_block:
|
||||
network_has_code = True
|
||||
for line in scl_code.splitlines():
|
||||
scl_output.append(f" {line}")
|
||||
if network_has_code:
|
||||
scl_output.append("") # Línea en blanco después del código de la red
|
||||
scl_output.append("")
|
||||
else:
|
||||
scl_output.append(f" // Network did not produce printable SCL code.")
|
||||
scl_output.append("")
|
||||
# Fin del bloque FC/FB
|
||||
scl_output.append(f"END_{scl_block_keyword}")
|
||||
|
||||
# Fin del bloque
|
||||
scl_output.append("END_FUNCTION_BLOCK") # O END_FUNCTION si es FC
|
||||
|
||||
# --- Escritura del Archivo SCL ---
|
||||
# --- Escritura del Archivo SCL (Común) ---
|
||||
print(f"Escribiendo archivo SCL en: {output_scl_filepath}")
|
||||
try:
|
||||
with open(output_scl_filepath, 'w', encoding='utf-8') as f:
|
||||
with open(output_scl_filepath, "w", encoding="utf-8") as f:
|
||||
for line in scl_output:
|
||||
f.write(line + '\n')
|
||||
f.write(line + "\n")
|
||||
print("Generación de SCL completada.")
|
||||
except Exception as e:
|
||||
print(f"Error al escribir el archivo SCL: {e}")
|
||||
|
|
@ -301,47 +492,61 @@ if __name__ == "__main__":
|
|||
import argparse
|
||||
import os
|
||||
import sys
|
||||
import traceback # Asegurarse que traceback está importado si se usa en generate_scl
|
||||
import traceback # Asegurarse que traceback está importado si se usa en generate_scl
|
||||
|
||||
# Configurar ArgumentParser para recibir la ruta del XML original obligatoria
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Generate final SCL file from processed JSON (_simplified_processed.json). Expects original XML filepath as argument."
|
||||
)
|
||||
parser.add_argument(
|
||||
"source_xml_filepath", # Argumento posicional obligatorio
|
||||
"source_xml_filepath", # Argumento posicional obligatorio
|
||||
help="Path to the original source XML file (passed from x0_main.py, used to derive input/output names).",
|
||||
)
|
||||
args = parser.parse_args() # Parsea los argumentos de sys.argv
|
||||
args = parser.parse_args() # Parsea los argumentos de sys.argv
|
||||
|
||||
source_xml_file = args.source_xml_filepath # Obtiene la ruta del XML original
|
||||
source_xml_file = args.source_xml_filepath # Obtiene la ruta del XML original
|
||||
|
||||
# Verificar si el archivo XML original existe (como referencia)
|
||||
if not os.path.exists(source_xml_file):
|
||||
print(f"Advertencia (x3): Archivo XML original no encontrado: '{source_xml_file}', pero se intentará encontrar el JSON procesado.")
|
||||
# No salir necesariamente.
|
||||
print(
|
||||
f"Advertencia (x3): Archivo XML original no encontrado: '{source_xml_file}', pero se intentará encontrar el JSON procesado."
|
||||
)
|
||||
# No salir necesariamente.
|
||||
|
||||
# Derivar nombres de archivos de entrada (JSON procesado) y salida (SCL)
|
||||
xml_filename_base = os.path.splitext(os.path.basename(source_xml_file))[0]
|
||||
# Asumir que los archivos están en el mismo directorio que el XML original
|
||||
base_dir = os.path.dirname(source_xml_file) # Directorio del XML original
|
||||
base_dir = os.path.dirname(source_xml_file) # Directorio del XML original
|
||||
|
||||
input_json_file = os.path.join(base_dir, f"{xml_filename_base}_simplified_processed.json")
|
||||
output_scl_file = os.path.join(base_dir, f"{xml_filename_base}_simplified_processed.scl")
|
||||
input_json_file = os.path.join(
|
||||
base_dir, f"{xml_filename_base}_simplified_processed.json"
|
||||
)
|
||||
output_scl_file = os.path.join(
|
||||
base_dir, f"{xml_filename_base}_simplified_processed.scl"
|
||||
)
|
||||
|
||||
print(f"(x3) Generando SCL: '{os.path.relpath(input_json_file)}' -> '{os.path.relpath(output_scl_file)}'")
|
||||
print(
|
||||
f"(x3) Generando SCL: '{os.path.relpath(input_json_file)}' -> '{os.path.relpath(output_scl_file)}'"
|
||||
)
|
||||
|
||||
# Verificar si el archivo JSON procesado de entrada EXISTE
|
||||
if not os.path.exists(input_json_file):
|
||||
print(f"Error Fatal (x3): Archivo JSON procesado no encontrado: '{input_json_file}'")
|
||||
print(f"Asegúrate de que 'x2_process.py' se ejecutó correctamente para '{os.path.relpath(source_xml_file)}'.")
|
||||
sys.exit(1) # Salir si el archivo necesario no está
|
||||
print(
|
||||
f"Error Fatal (x3): Archivo JSON procesado no encontrado: '{input_json_file}'"
|
||||
)
|
||||
print(
|
||||
f"Asegúrate de que 'x2_process.py' se ejecutó correctamente para '{os.path.relpath(source_xml_file)}'."
|
||||
)
|
||||
sys.exit(1) # Salir si el archivo necesario no está
|
||||
else:
|
||||
# Llamar a la función principal de generación SCL del script
|
||||
# Asumiendo que tu función principal se llama generate_scl(input_json_path, output_scl_path)
|
||||
try:
|
||||
generate_scl(input_json_file, output_scl_file)
|
||||
except Exception as e:
|
||||
print(f"Error Crítico (x3) durante la generación de SCL desde '{input_json_file}': {e}")
|
||||
print(
|
||||
f"Error Crítico (x3) durante la generación de SCL desde '{input_json_file}': {e}"
|
||||
)
|
||||
# traceback ya debería estar importado si generate_scl lo necesita
|
||||
traceback.print_exc()
|
||||
sys.exit(1) # Salir con error si la función principal falla
|
||||
sys.exit(1) # Salir con error si la función principal falla
|
||||
|
|
|
|||
Loading…
Reference in New Issue