322 lines
16 KiB
Python
322 lines
16 KiB
Python
# x3_generate_scl.py
|
|
# -*- coding: utf-8 -*-
|
|
import json
|
|
import os
|
|
import re
|
|
import argparse
|
|
import sys
|
|
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
|
|
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!).")
|
|
# 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
|
|
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"
|
|
|
|
|
|
# --- 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)."""
|
|
|
|
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 ---
|
|
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 ---
|
|
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("")
|
|
|
|
# Declaraciones de Interfaz (Implementación básica)
|
|
interface_sections = ["Input", "Output", "InOut", "Static", "Temp", "Constant", "Return"]
|
|
interface_data = data.get('interface', {})
|
|
|
|
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("")
|
|
|
|
network_has_code = False
|
|
|
|
# --- 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
|
|
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" {'//'} ```")
|
|
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')}")
|
|
|
|
|
|
if network_has_code:
|
|
scl_output.append("") # Línea en blanco después del código de la red
|
|
else:
|
|
scl_output.append(f" // Network did not produce printable SCL code.")
|
|
scl_output.append("")
|
|
|
|
# Fin del bloque
|
|
scl_output.append("END_FUNCTION_BLOCK") # O END_FUNCTION si es FC
|
|
|
|
# --- Escritura del Archivo SCL ---
|
|
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 ---
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Generate final SCL file from processed JSON (SymPy version).")
|
|
parser.add_argument(
|
|
"source_xml_filepath",
|
|
nargs="?",
|
|
default="TestLAD.xml",
|
|
help="Path to the original source XML file (used to derive input/output names, default: TestLAD.xml)"
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
xml_filename_base = os.path.splitext(os.path.basename(args.source_xml_filepath))[0]
|
|
# Usar directorio del script si no hay ruta, o la ruta del XML si la tiene
|
|
xml_dir = os.path.dirname(args.source_xml_filepath)
|
|
base_dir = xml_dir if xml_dir else os.path.dirname(__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")
|
|
|
|
if not os.path.exists(input_json_file):
|
|
print(f"Error: Processed JSON file not found: '{input_json_file}'")
|
|
print(f"Ensure 'x2_process.py' ran successfully for '{args.source_xml_filepath}'.")
|
|
sys.exit(1)
|
|
else:
|
|
generate_scl(input_json_file, output_scl_file) |