Refactorizado de process en scripts individuales

This commit is contained in:
Miguel 2025-04-19 14:43:12 +02:00
parent e3b4d413c9
commit 98dbc7a5d7
31 changed files with 2048 additions and 4358 deletions

View File

@ -133,14 +133,6 @@
}
},
"outputs": {
"RET_VAL": [
{
"uid": "23",
"scope": "LocalVariable",
"type": "variable",
"name": "\"Block_Move_Err\""
}
],
"DSTBLK": [
{
"uid": "24",
@ -148,6 +140,14 @@
"type": "variable",
"name": "\"Filler_Head_Variables\".\"FillerHead\""
}
],
"RET_VAL": [
{
"uid": "23",
"scope": "LocalVariable",
"type": "variable",
"name": "\"Block_Move_Err\""
}
]
}
}

View File

@ -136,14 +136,6 @@
}
},
"outputs": {
"RET_VAL": [
{
"uid": "23",
"scope": "LocalVariable",
"type": "variable",
"name": "\"Block_Move_Err\""
}
],
"DSTBLK": [
{
"uid": "24",
@ -151,6 +143,14 @@
"type": "variable",
"name": "\"Filler_Head_Variables\".\"FillerHead\""
}
],
"RET_VAL": [
{
"uid": "23",
"scope": "LocalVariable",
"type": "variable",
"name": "\"Block_Move_Err\""
}
]
},
"scl": "IF \"AUX FALSE\" THEN\n \"Block_Move_Err\" := BLKMOV(SRCBLK := \"HMI_PID\".\"PPM303\", DSTBLK => \"Filler_Head_Variables\".\"FillerHead\"); // ADVERTENCIA: BLKMOV usado directamente, probablemente no compile!\nEND_IF;"

View File

@ -251,17 +251,17 @@
},
"negated_pins": {},
"inputs": {
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "34",
"source_pin": "out"
},
"in2": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "37",
"source_pin": "out"
},
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "34",
"source_pin": "out"
}
},
"outputs": {
@ -326,18 +326,18 @@
"template_values": {},
"negated_pins": {},
"inputs": {
"s": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "25",
"source_pin": "out"
},
"timer": {
"uid": "22",
"scope": "GlobalVariable",
"type": "variable",
"name": "\"mHVM302_Dly\""
},
"s": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "25",
"source_pin": "out"
},
"tv": {
"uid": "23",
"scope": "TypedConstant",
@ -414,18 +414,18 @@
"template_values": {},
"negated_pins": {},
"inputs": {
"s": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "24",
"source_pin": "out"
},
"timer": {
"uid": "22",
"scope": "GlobalVariable",
"type": "variable",
"name": "\"mResetTotalizerTmr\""
},
"s": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "24",
"source_pin": "out"
},
"tv": {
"uid": "23",
"scope": "TypedConstant",
@ -501,17 +501,17 @@
},
"negated_pins": {},
"inputs": {
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "26",
"source_pin": "out"
},
"in2": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "27",
"source_pin": "out"
},
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "26",
"source_pin": "out"
}
},
"outputs": {
@ -525,18 +525,18 @@
"template_values": {},
"negated_pins": {},
"inputs": {
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "28",
"source_pin": "out"
},
"timer": {
"uid": "23",
"scope": "GlobalVariable",
"type": "variable",
"name": "\"mResetFTN301TotTmr\""
},
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "28",
"source_pin": "out"
},
"tv": {
"uid": "24",
"scope": "TypedConstant",
@ -636,17 +636,17 @@
},
"negated_pins": {},
"inputs": {
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "27",
"source_pin": "out"
},
"in2": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "28",
"source_pin": "out"
},
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "27",
"source_pin": "out"
}
},
"outputs": {
@ -660,18 +660,18 @@
"template_values": {},
"negated_pins": {},
"inputs": {
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "29",
"source_pin": "out"
},
"timer": {
"uid": "23",
"scope": "GlobalVariable",
"type": "variable",
"name": "\"mResetFTP302TotTmr\""
},
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "29",
"source_pin": "out"
},
"tv": {
"uid": "24",
"scope": "TypedConstant",
@ -795,17 +795,17 @@
},
"negated_pins": {},
"inputs": {
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "26",
"source_pin": "out"
},
"in2": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "27",
"source_pin": "out"
},
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "26",
"source_pin": "out"
}
},
"outputs": {
@ -819,18 +819,18 @@
"template_values": {},
"negated_pins": {},
"inputs": {
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "28",
"source_pin": "out"
},
"timer": {
"uid": "23",
"scope": "GlobalVariable",
"type": "variable",
"name": "\"mResetFTM303TotTmr\""
},
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "28",
"source_pin": "out"
},
"tv": {
"uid": "24",
"scope": "TypedConstant",
@ -930,17 +930,17 @@
},
"negated_pins": {},
"inputs": {
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "26",
"source_pin": "out"
},
"in2": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "27",
"source_pin": "out"
},
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "26",
"source_pin": "out"
}
},
"outputs": {
@ -954,18 +954,18 @@
"template_values": {},
"negated_pins": {},
"inputs": {
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "28",
"source_pin": "out"
},
"timer": {
"uid": "23",
"scope": "GlobalVariable",
"type": "variable",
"name": "\"mResetProductTotTmr\""
},
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "28",
"source_pin": "out"
},
"tv": {
"uid": "24",
"scope": "TypedConstant",
@ -1163,12 +1163,6 @@
"type": "variable",
"name": "\"T_Pulse_Recipe_Edit\""
},
"s": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "38",
"source_pin": "out"
},
"tv": {
"uid": "28",
"scope": "TypedConstant",
@ -1176,6 +1170,12 @@
"datatype": "TypedConstant",
"value": "S5T#500ms"
},
"s": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "38",
"source_pin": "out"
},
"en": {
"type": "connection",
"source_instruction_uid": "38",

View File

@ -259,17 +259,17 @@
},
"negated_pins": {},
"inputs": {
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "34",
"source_pin": "out"
},
"in2": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "37",
"source_pin": "out"
},
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "34",
"source_pin": "out"
}
},
"outputs": {
@ -337,18 +337,18 @@
"template_values": {},
"negated_pins": {},
"inputs": {
"s": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "25",
"source_pin": "out"
},
"timer": {
"uid": "22",
"scope": "GlobalVariable",
"type": "variable",
"name": "\"mHVM302_Dly\""
},
"s": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "25",
"source_pin": "out"
},
"tv": {
"uid": "23",
"scope": "TypedConstant",
@ -428,18 +428,18 @@
"template_values": {},
"negated_pins": {},
"inputs": {
"s": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "24",
"source_pin": "out"
},
"timer": {
"uid": "22",
"scope": "GlobalVariable",
"type": "variable",
"name": "\"mResetTotalizerTmr\""
},
"s": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "24",
"source_pin": "out"
},
"tv": {
"uid": "23",
"scope": "TypedConstant",
@ -518,17 +518,17 @@
},
"negated_pins": {},
"inputs": {
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "26",
"source_pin": "out"
},
"in2": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "27",
"source_pin": "out"
},
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "26",
"source_pin": "out"
}
},
"outputs": {
@ -543,18 +543,18 @@
"template_values": {},
"negated_pins": {},
"inputs": {
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "28",
"source_pin": "out"
},
"timer": {
"uid": "23",
"scope": "GlobalVariable",
"type": "variable",
"name": "\"mResetFTN301TotTmr\""
},
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "28",
"source_pin": "out"
},
"tv": {
"uid": "24",
"scope": "TypedConstant",
@ -658,17 +658,17 @@
},
"negated_pins": {},
"inputs": {
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "27",
"source_pin": "out"
},
"in2": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "28",
"source_pin": "out"
},
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "27",
"source_pin": "out"
}
},
"outputs": {
@ -683,18 +683,18 @@
"template_values": {},
"negated_pins": {},
"inputs": {
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "29",
"source_pin": "out"
},
"timer": {
"uid": "23",
"scope": "GlobalVariable",
"type": "variable",
"name": "\"mResetFTP302TotTmr\""
},
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "29",
"source_pin": "out"
},
"tv": {
"uid": "24",
"scope": "TypedConstant",
@ -823,17 +823,17 @@
},
"negated_pins": {},
"inputs": {
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "26",
"source_pin": "out"
},
"in2": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "27",
"source_pin": "out"
},
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "26",
"source_pin": "out"
}
},
"outputs": {
@ -848,18 +848,18 @@
"template_values": {},
"negated_pins": {},
"inputs": {
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "28",
"source_pin": "out"
},
"timer": {
"uid": "23",
"scope": "GlobalVariable",
"type": "variable",
"name": "\"mResetFTM303TotTmr\""
},
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "28",
"source_pin": "out"
},
"tv": {
"uid": "24",
"scope": "TypedConstant",
@ -963,17 +963,17 @@
},
"negated_pins": {},
"inputs": {
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "26",
"source_pin": "out"
},
"in2": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "27",
"source_pin": "out"
},
"in1": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "26",
"source_pin": "out"
}
},
"outputs": {
@ -988,18 +988,18 @@
"template_values": {},
"negated_pins": {},
"inputs": {
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "28",
"source_pin": "out"
},
"timer": {
"uid": "23",
"scope": "GlobalVariable",
"type": "variable",
"name": "\"mResetProductTotTmr\""
},
"s": {
"type": "connection",
"source_instruction_type": "O",
"source_instruction_uid": "28",
"source_pin": "out"
},
"tv": {
"uid": "24",
"scope": "TypedConstant",
@ -1205,12 +1205,6 @@
"type": "variable",
"name": "\"T_Pulse_Recipe_Edit\""
},
"s": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "38",
"source_pin": "out"
},
"tv": {
"uid": "28",
"scope": "TypedConstant",
@ -1218,6 +1212,12 @@
"datatype": "TypedConstant",
"value": "S5T#500ms"
},
"s": {
"type": "connection",
"source_instruction_type": "Contact",
"source_instruction_uid": "38",
"source_pin": "out"
},
"en": {
"type": "connection",
"source_instruction_uid": "38",

View File

@ -1,170 +0,0 @@
# LAD-to-SCL Conversion Pipeline: Documentación de Referencia
## 1. Visión General
Este documento describe un pipeline de scripts de Python diseñado para convertir bloques de función o funciones (FC/FB) escritos en Ladder Logic (LAD) desde archivos XML de TIA Portal Openness a un código SCL (Structured Control Language) semánticamente equivalente.
El proceso se divide en tres etapas principales, cada una manejada por un script específico:
1. **XML a JSON Enriquecido (`x1_to_json.py`):** Parsea el XML de Openness, extrae la estructura lógica (incluyendo llamadas a FC/FB y temporizadores S5), conexiones explícitas e **infiere conexiones implícitas** (especialmente las habilitaciones EN) para crear un archivo JSON detallado. Mapea tipos de instrucción LAD/FBD (p.ej., `Se`, `NBox`) a nombres internos consistentes (p.ej., `TON_S5`, `N_TRIG`).
2. **Procesamiento Semántico (`x2_process.py`):** Lee el JSON enriquecido y, de forma iterativa, traduce cada instrucción (usando los tipos mapeados) a su equivalente SCL, manejando dependencias, propagando el estado lógico (RLO), traduciendo temporizadores S5 a IEC, generando lógica de flancos y agrupando lógica paralela bajo bloques `IF`. El SCL generado se almacena *dentro* del propio JSON.
3. **Generación de SCL Final (`x3_generate_scl.py`):** Lee el JSON completamente procesado y ensambla el código SCL final en un archivo `.scl` formateado, incluyendo declaraciones de variables (Input, Output, InOut, Temp, Stat - incluyendo instancias de temporizadores y bits de memoria de flancos) y el cuerpo del programa (`FUNCTION` o `FUNCTION_BLOCK`).
## 2. Etapas del Pipeline
### Etapa 1: XML a JSON Enriquecido (`x1_to_json.py`)
* **Propósito:** Transformar la compleja y a veces ambigua estructura XML de Openness en un formato JSON estructurado y más fácil de procesar, añadiendo información clave que está implícita en el LAD visual pero no siempre explícita en el XML.
* **Entrada:** Archivo `.xml` exportado desde TIA Portal Openness para un FC o FB.
* **Salida:** Archivo `_simplified.json`.
* **Proceso Clave:**
1. **Parseo XML:** Utiliza `lxml` para leer el archivo XML.
2. **Extracción de Metadatos:** Obtiene nombre del bloque, número, lenguaje original, comentario del bloque. Detecta si es `SW.Blocks.FC` o `SW.Blocks.FB`.
3. **Extracción de Interfaz:** Parsea las secciones `Input`, `Output`, `InOut`, `Temp`, `Constant`, `Return` para obtener la declaración de variables.
4. **Parseo de Redes (`CompileUnit`):** Itera sobre cada red lógica.
* **`parse_network`:**
* **Parseo de Componentes:**
* `Access`: Identifica variables globales/locales, constantes literales y tipadas (`parse_access`). Utiliza `get_symbol_name` para formatear nombres simbólicos con comillas (`"DB"."Var"`).
* `Part`: Identifica instrucciones estándar LAD/FBD (`parse_part`). **Importante:** Detecta pines negados (`<Negated>`) y los almacena en `negated_pins`. **Mapea** nombres XML (`Se`, `Sd`, `PBox`, `NBox`, `RCoil`, `SCoil`, `SdCoil`) a tipos internos (`TON_S5`, `TONR_S5`, `P_TRIG`, `N_TRIG`, `R`, `S`, `SR`).
* `Call`: Identifica llamadas a otros FCs o FBs (`parse_call`), extrayendo el nombre del bloque llamado, tipo (FC/FB) e información de instancia DB (`instance_db`) si aplica (formateado con comillas).
* Crea `access_map` y `parts_and_calls_map` para referencia rápida por UID.
* **Parseo de Conexiones (`Wire`):**
* Construye `wire_connections`: Un mapa `(dest_uid, dest_pin) -> [(src_uid, src_pin), ...]`.
* Construye `source_connections`: Un mapa `(src_uid, src_pin) -> [(dest_uid, dest_pin), ...]`.
* **Construcción Lógica Inicial:** Crea una lista de diccionarios (`all_logic_steps`), uno por cada `Part` o `Call`, rellenando `inputs` y `outputs` **solo con las conexiones explícitas** encontradas en los `Wire`.
* **Inferencia de Conexión `EN` (¡Paso Crucial!):**
* Itera sobre los bloques funcionales (`Move`, `Add`, `Call`, `BLKMOV`, etc.) que *no* tienen una entrada `en` explícita definida después del paso anterior.
* Utiliza una **heurística de búsqueda lineal hacia atrás** para encontrar la fuente de RLO más probable que precede a este bloque (la salida `out` de un bloque lógico como `Contact`/`O`/`Eq`/`P_TRIG`/`N_TRIG` o la salida `eno` de un bloque funcional anterior).
* Si se encuentra una fuente inferida, **añade la conexión `en`** al diccionario `inputs` del bloque funcional actual. Esto enriquece el JSON con la dependencia lógica implícita.
* **Ordenamiento:** Ordena las instrucciones en la lista `logic` final (generalmente por UID).
5. **Escritura JSON:** Guarda la estructura de datos completa en el archivo `_simplified.json`.
### Etapa 2: Procesamiento Semántico (`x2_process.py`)
* **Propósito:** Traducir la lógica LAD/FBD (representada en el JSON enriquecido) a código SCL embebido, resolviendo dependencias entre instrucciones, generando lógica SCL equivalente (incluyendo manejo de flancos y temporizadores), y aplicando optimizaciones de agrupación `IF`.
* **Entrada:** Archivo `_simplified.json` (generado por la Etapa 1).
* **Salida:** Archivo `_simplified_processed.json`.
* **Proceso Clave:**
1. **Carga de JSON y Preparación:** Lee el JSON de entrada y reconstruye mapas de acceso por red. Inicializa el `scl_map` global (o por red).
2. **Bucle Iterativo:** Repite los siguientes pasos hasta que no se realicen cambios en un pase completo o se alcance un límite máximo de pases (`max_passes`).
* **Fase 1: Procesadores Base:**
* Itera sobre cada instrucción en cada red.
* Si la instrucción no ha sido procesada (`_scl` o `_error`) ni agrupada (`grouped`), busca el procesador adecuado (`process_xxx`) basado en su `type` (usando los tipos mapeados como `TON_S5`, `P_TRIG`, etc.).
* **`process_xxx` (Ejecución):**
* Llama a `get_scl_representation` para obtener el SCL de sus pines de entrada, buscándolos en `scl_map` o directamente en `access_map`. `get_scl_representation` ahora **convierte constantes S5T# a T#**.
* Si alguna dependencia no está resuelta (devuelve `None`), el procesador retorna `False`.
* Si las dependencias están resueltas:
* **Generadores RLO (`Contact`, `O`, `Eq`, `P_TRIG`, `N_TRIG`):** Calculan la expresión booleana SCL resultante y la almacenan en `scl_map` bajo la clave `(network_id, instr_uid, 'out')`. `P_TRIG`/`N_TRIG` generan lógica explícita de flancos y la llamada para actualizar el bit de memoria (`stat_... := CLK;`). Guardan comentarios/lógica en `instruction['scl']`.
* **Bloques Funcionales (`Move`, `Add`, `Convert`, `Mod`, `BLKMOV`, `Call`):**
* Obtienen la condición `EN` (explícita o inferida).
* Generan el código SCL *core* (la asignación, cálculo o llamada). `process_call` ahora usa correctamente `instance_db` para FBs. `process_blkmov` traduce a `DST := SRC;` (con advertencia sobre tipos).
* Generan el SCL final, **envolviendo el core en `IF en_scl THEN ... END_IF;`** si `en_scl` no es `"TRUE"`.
* Almacenan el SCL final en `instruction['scl']`.
* Almacenan el nombre de la variable de salida/temporal (prefijado con `#` si es temporal) en `scl_map` para `out`/`out1`/`RET_VAL`.
* Almacenan el `en_scl` en `scl_map` para `eno`.
* **Bobinas (`Coil`, `R`, `S`, `SR`):** Obtienen el RLO de entrada (`in`, o `S`/`R1` para `SR`), obtienen el operando destino, generan la asignación (`:= TRUE`/`:= FALSE`) o lógica `IF/ELSIF` envuelta en `IF RLO THEN ...` si aplica, y la guardan en `instruction['scl']`.
* **Temporizadores (`TON_S5`, `TONR_S5`):** Generan una llamada a un bloque IEC (`TON`/`TONR`) usando un nombre de instancia (`stat_timer_...`). Parsean el valor `PT` (T#...). Mapean `Q` y `ET` a la instancia (`Instance.Q`, `Instance.ET`) en `scl_map`. Almacenan la llamada en `instruction['scl']`.
* Marcan la instrucción como procesada añadiendo `_scl` a `instruction['type']`.
* Retornan `True`.
* Se registra si hubo algún cambio en esta fase (`made_change_in_base_pass`).
* **Fase 2: Agrupación de IFs (`process_group_ifs`):**
* Itera sobre las instrucciones *ya procesadas* (`_scl`) que son generadoras de condición (`Contact`, `O`, `Eq`, `P_TRIG`, `N_TRIG`, etc.).
* Obtiene la `condition_scl` de `scl_map`.
* Busca todos los bloques funcionales, bobinas o temporizadores (`Move_scl`, `Add_scl`, `Coil_scl`, `TON_S5_scl`, etc.) cuyo pin de habilitación (`en`, `in`, `s`) esté conectado a la salida `out` de esta instrucción generadora.
* Si encuentra **más de uno**:
* Extrae el código *core* (lo que está dentro del `IF...END_IF;` si existe, o el código completo si no) de cada consumidor.
* Construye un **único bloque `IF condition_scl THEN ... END_IF;`** que contiene todos los cores extraídos, indentados.
* **Sobrescribe** el campo `scl` de la instrucción *generadora de condición* con este nuevo bloque `IF` agrupado.
* Marca cada instrucción consumidora con `grouped = True` y cambia su `scl` a un comentario (`GROUPED_COMMENT`) para evitar que `x3_generate_scl.py` lo use.
* Se registra si hubo algún cambio en esta fase (`made_change_in_group_pass`).
* **Condición de Salida:** El bucle termina si `made_change_in_base_pass` y `made_change_in_group_pass` son ambos `False`.
3. **Verificación Final:** Comprueba si quedaron instrucciones sin procesar, sin agrupar y sin errores, e informa al usuario.
4. **Escritura JSON:** Guarda el JSON modificado (con SCL embebido y marcas de agrupación) en el archivo `_simplified_processed.json`.
### Etapa 3: Generación de SCL Final (`x3_generate_scl.py`)
* **Propósito:** Ensamblar un archivo `.scl` completo y formateado a partir del JSON procesado.
* **Entrada:** Archivo `_simplified_processed.json` (generado por la Etapa 2).
* **Salida:** Archivo `.scl`.
* **Proceso Clave:**
1. **Carga de JSON:** Lee el archivo JSON procesado.
2. **Detección de Variables y Tipo de Bloque:**
* Escanea el SCL generado en busca de variables `#_temp_...` y `stat_...`.
* Identifica los tipos de las variables `stat_...` (Bool, TON, TONR) según sus nombres (`stat_nbox_mem...`, `stat_timer_Se...`, etc.).
* Determina si el bloque debe ser `FUNCTION_BLOCK` (si hay variables `STAT` o `TEMP`) o `FUNCTION`.
3. **Generación de Cabecera:** Escribe el encabezado del bloque (`FUNCTION_BLOCK name` o `FUNCTION name`, `VERSION`, etc.). Utiliza `format_variable_name` (la versión corregida) para el nombre del bloque.
4. **Generación de Declaraciones VAR:**
* Escribe las secciones `VAR_INPUT`, `VAR_OUTPUT`, `VAR_IN_OUT` usando `format_variable_name` para los nombres de las variables de la interfaz.
* **Escribe `VAR_STAT`:** Declara las variables `stat_...` detectadas con sus tipos inferidos (Bool, TON, TONR) y nombres entre comillas.
* **Escribe `VAR_TEMP`:** Declara las variables `#_temp_...` detectadas (declaradas como `"_temp_..."` sin el `#` pero con comillas) y las variables de la sección `Temp` de la interfaz. Utiliza inferencia de tipo básica para las variables `#_temp_...`.
5. **Generación del Cuerpo (`BEGIN`/`END_FUNCTION_BLOCK` o `END_FUNCTION`):**
* Itera sobre las `networks` en el JSON.
* Añade comentarios de red.
* Itera sobre la `logic` de cada red.
* Para cada `instruction`:
* **Verifica el flag `grouped`:** Si `instruction.get('grouped', False)` es `True`, **ignora** esta instrucción.
* Si no está agrupada, obtiene el valor del campo `scl`.
* **Limpia comentarios internos:** Elimina comentarios informativos específicos (`// RLO:`, `// Comparison:`, `// Logic O:`, `// N/P_TRIG Output Logic:`) si son la única parte de la línea, pero conserva los comentarios de actualización de memoria de flancos, errores y agrupación.
* Indenta y añade las líneas del SCL resultante al output.
* Añade líneas en blanco entre redes si contienen código.
6. **Escritura de Archivo:** Escribe el string SCL completo al archivo `.scl`.
## 3. Cómo Extender para Nuevas Instrucciones LAD/FBD
Añadir soporte para un nuevo tipo de instrucción (p.ej., un comparador `GT`, una función matemática `SQRT`, o un temporizador IEC `TP`) requiere modificar los scripts `x1_to_json.py` y `x2_process.py`.
**Pasos:**
1. **Analizar el XML:** Exporta un bloque simple que use la nueva instrucción y examina el XML de Openness. Identifica:
* Si se representa como `<Part Name="NombreInstruccion">` o `<Call Name="NombreInstruccion">`.
* Sus pines de entrada y salida (`NameCon`/`IdentCon` en los `Wire`).
* Cualquier atributo o `TemplateValue` relevante.
* Si tiene pines negados (`<Negated>`).
2. **Modificar `x1_to_json.py` (`parse_part` o `parse_call` y `parse_network`):**
* **Parseo:** Asegúrate de que `parse_part` o `parse_call` capture correctamente la nueva instrucción.
* **Mapeo de Tipo:** Si el nombre XML no es ideal (como `Se`), mapéalo a un nombre interno consistente en `parse_part` (p.ej., si fuera un temporizador IEC TP, podrías mapear `TPartName` -> `TP_IEC`).
* **Clasificación:** Decide si la nueva instrucción es un `functional_block_type` (necesita inferencia EN?) o un `rlo_generator`. Actualiza estas listas en `parse_network` si es necesario.
* **Inferencia EN:** Revisa si la lógica de inferencia EN en `parse_network` necesita ajustes.
* **Pines:** Asegúrate de que sus pines de entrada/salida esperados estén en `possible_input_pins` / `possible_output_pins` en `parse_network`.
3. **Modificar `x2_process.py` (continuación):**
* **Lógica SCL (continuación):**
* ...
* Almacena el SCL final en `instruction['scl']`.
* Actualiza `instruction['type']` con el sufijo `_scl`.
* **Actualiza `scl_map`:** Añade entradas para los pines de salida (`out`, `Q`, etc.) con su representación SCL (puede ser un nombre de variable temporal prefijado con `#`, el resultado de una expresión, o el acceso a un miembro de instancia como `"Instance".Q`). Añade la entrada para `eno` si el bloque lo tiene (generalmente `scl_map[key_eno] = en_scl`).
* Retorna `True`.
* **Añadir a la Lista:** Inserta la nueva función `process_nombre_instruccion` en la lista `base_processors_list` en el orden de prioridad correcto (generalmente, generadores de valores/condiciones antes que consumidores, y las llamadas (`process_call`) al final o cerca del final). Actualiza el `processor_map` si es necesario (especialmente si manejas tipos como `Call_FC`, `Call_FB`).
* **Agrupación:** Considera si este nuevo bloque debería ser parte de la lógica de agrupación (¿es un bloque funcional, bobina o temporizador que puede ser habilitado en paralelo?). Si es así, añádelo a la lista `groupable_types_original` en `process_group_ifs`. El código de agrupación existente debería funcionar si el procesador del nuevo bloque genera correctamente la estructura `IF EN THEN ... END_IF;` (o código simple si EN=TRUE).
4. **Modificar `x3_generate_scl.py`:**
* **Declaraciones STAT:** Si la nueva instrucción introduce la necesidad de un nuevo tipo de variable estática (p.ej., un tipo de datos específico para un contador, o un nuevo tipo de instancia de FB IEC), necesitarás:
* Añadir una nueva expresión regular (`stat_pattern_xxx`) para detectar el nombre de la instancia/variable estática generada por tu nuevo procesador en `x2_process.py`.
* Actualizar el bucle de detección para que use esta nueva regex y almacene el nombre y el tipo SCL correcto (`MyCounterType`, `IEC_Counter`, etc.) en el diccionario `stat_vars`.
* La lógica existente en `generate_scl` que escribe la sección `VAR_STAT` usará esta información para declarar la variable correctamente.
* **Declaraciones TEMP:** Si la nueva instrucción genera variables temporales con un patrón específico o requiere un tipo de dato específico que pueda ser inferido, puedes mejorar la lógica de inferencia de tipo en la sección `VAR_TEMP`.
* **Limpieza de Comentarios:** Si el nuevo procesador genera comentarios internos específicos que no quieres en el SCL final, ajusta la lógica de filtrado en la sección de generación del cuerpo del bloque.
5. **Probar:**
* Crea un archivo XML de prueba simple que use la nueva instrucción en diferentes contextos (con entradas/salidas conectadas, con EN explícito/implícito, negaciones si aplica).
* Ejecuta el pipeline completo (`x1_to_json.py`, `x2_process.py`, `x3_generate_scl.py`).
* **Verifica `_simplified.json`:** Asegúrate de que `x1` parseó correctamente la instrucción, mapeó su tipo, identificó sus conexiones y realizó la inferencia EN si era necesario.
* **Verifica `_simplified_processed.json`:** Comprueba que `x2` ejecutó tu nuevo procesador, generó el SCL esperado en el campo `scl`, marcó el tipo con `_scl`, y actualizó el `scl_map` correctamente para sus salidas. Verifica si la agrupación `IF` funcionó como se esperaba si la instrucción era parte de lógica paralela.
* **Verifica el archivo `.scl`:** Confirma que `x3` generó el SCL final correctamente, incluyendo las declaraciones necesarias (STAT/TEMP) y el código SCL de la instrucción (indentado y sin comentarios no deseados). Asegúrate de que no haya errores de sintaxis SCL obvios.
* **Importar en TIA Portal (Opcional pero Recomendado):** Intenta importar el archivo SCL generado en un proyecto de TIA Portal para validar la sintaxis y la estructura del bloque.
## 4. Futuras Mejoras y Consideraciones
* **Manejo de Tipos de Datos:** Implementar un seguimiento y conversión de tipos más robusto. Inferir tipos para variables temporales de forma más precisa. Usar funciones de conversión SCL explícitas (`INT_TO_DINT`, etc.) donde sea necesario, posiblemente requiriendo información de tipo adicional en el JSON.
* **Lógica de Flancos/Temporizadores:** Asegurar que la traducción de temporizadores S5 y la lógica de flancos generada sea completamente compatible con los bloques IEC estándar (`TON`, `TOF`, `TP`, `TONR`, `R_TRIG`, `F_TRIG`). Considerar la necesidad de declarar instancias de `R_TRIG`/`F_TRIG` en `VAR_STAT` en lugar de generar lógica explícita.
* **Soporte para FBs:** Mejorar el manejo de parámetros InOut y Return para llamadas a FC/FB. Potencialmente, requerir información de la interfaz del bloque llamado (parseando su propio XML o desde una biblioteca) para generar llamadas SCL más precisas.
* **Optimización SCL:** Explorar más optimizaciones SCL más allá de la agrupación de IFs (p.ej., simplificación de expresiones booleanas complejas, propagación de constantes).
* **Estructuras LAD Complejas:** La inferencia de EN actual (búsqueda lineal hacia atrás) es simple. Para manejar bifurcaciones (`Branch`), uniones (`Merge`) y saltos (`JMP`) complejos de forma robusta, `x1_to_json.py` necesitaría realizar un análisis de flujo de datos/control más sofisticado para determinar las dependencias lógicas correctas antes de generar el JSON.
* **Manejo de Errores:** Mejorar el reporte de errores con más contexto (nombre de red, UID, tipo de instrucción). Añadir más validaciones en cada etapa.
* **Interfaz Gráfica/Configuración:** Crear una interfaz más amigable o archivos de configuración para gestionar el proceso, seleccionar archivos y configurar opciones (p.ej., idioma por defecto, nivel de log).
* **Soporte para otros lenguajes (FBD/STL):** Extender el parser `x1_to_json.py` y los procesadores en `x2_process.py` para manejar otros lenguajes de origen, lo cual requeriría entender sus respectivas representaciones XML en Openness.
## 5. Conclusión
Este pipeline proporciona una base funcional para la conversión automática de LAD a SCL desde archivos TIA Portal Openness XML. La clave de su funcionamiento es la **separación de responsabilidades**: `x1` enriquece el modelo de datos con información implícita, `x2` realiza la traducción semántica iterativa y la optimización de agrupación, y `x3` ensambla el archivo SCL final. Al añadir procesadores específicos para cada tipo de instrucción LAD/FBD y refinar la lógica de inferencia y generación, la cobertura y calidad de la conversión pueden ser extendidas significativamente. La estructura modular facilita la incorporación de soporte para nuevas instrucciones y futuras mejoras.

View File

@ -1,687 +0,0 @@
# -*- coding: utf-8 -*-
import json
import os
from lxml import etree
import traceback
from collections import defaultdict
# --- Namespaces ---
ns = {
"iface": "http://www.siemens.com/automation/Openness/SW/Interface/v5",
"flg": "http://www.siemens.com/automation/Openness/SW/NetworkSource/FlgNet/v4",
}
# --- Helper Functions ---
# get_multilingual_text, get_symbol_name, parse_access - No changes needed from previous corrected version
def get_multilingual_text(element, default_lang="en-US", fallback_lang="it-IT"):
if element is None:
return ""
try:
# Try default language first
xpath_expr_default = (
f".//iface:MultilingualTextItem[iface:Culture='{default_lang}']/iface:Text"
)
text_items = element.xpath(xpath_expr_default, namespaces=ns)
if text_items and text_items[0].text is not None:
return text_items[0].text.strip()
# Try fallback language
xpath_expr_fallback = (
f".//iface:MultilingualTextItem[iface:Culture='{fallback_lang}']/iface:Text"
)
text_items = element.xpath(xpath_expr_fallback, namespaces=ns)
if text_items and text_items[0].text is not None:
return text_items[0].text.strip()
# Try any language if specific ones fail
xpath_expr_any = ".//iface:MultilingualTextItem/iface:Text"
text_items = element.xpath(xpath_expr_any, namespaces=ns)
if text_items and text_items[0].text is not None:
return text_items[0].text.strip()
return "" # No text found
except Exception as e:
# print(f"Advertencia: Error extrayendo MultilingualText: {e}") # Reduced verbosity
return ""
def get_symbol_name(symbol_element):
"""Extracts the full symbolic name, adding quotes around each component."""
if symbol_element is None:
return None
try:
# Namespace might be missing on Component, use local-name()
components = symbol_element.xpath("./*[local-name()='Component']/@Name")
# Ensure quotes are added correctly
return ".".join(f'"{c}"' for c in components) if components else None
except Exception as e:
print(f"Advertencia: Excepción en get_symbol_name: {e}")
return None
def parse_access(access_element):
"""Parses Access elements (variables, constants)."""
if access_element is None:
return None
uid = access_element.get("UId")
scope = access_element.get(
"Scope"
) # GlobalVariable, LocalVariable, LiteralConstant, TypedConstant etc.
info = {"uid": uid, "scope": scope, "type": "unknown_access"} # Default type
symbol = access_element.xpath("./*[local-name()='Symbol']")
constant = access_element.xpath("./*[local-name()='Constant']")
# Add check for ConstantValue tag directly under Access (sometimes happens for literals)
const_val_direct = access_element.xpath("./*[local-name()='ConstantValue']/text()")
if symbol:
info["type"] = "variable"
info["name"] = get_symbol_name(symbol[0])
if info["name"] is None:
info["type"] = "error_parsing_symbol"
print(f"Error: No se pudo parsear nombre símbolo Access UID={uid}")
elif constant:
info["type"] = "constant"
const_type_elem = constant[0].xpath("./*[local-name()='ConstantType']")
const_val_elem = constant[0].xpath("./*[local-name()='ConstantValue']")
info["datatype"] = (
const_type_elem[0].text.strip()
if const_type_elem and const_type_elem[0].text
else "Unknown"
)
info["value"] = (
const_val_elem[0].text.strip()
if const_val_elem and const_val_elem[0].text
else None
)
if info["value"] is None:
info["type"] = "error_parsing_constant"
print(f"Error: Constante sin valor Access UID={uid}")
# Handle S5Time specifically - store its original format
elif info["datatype"] == "Unknown" and scope == "TypedConstant":
if info["value"].upper().startswith("S5T#"):
info["datatype"] = "S5Time" # Mark as S5Time, value remains S5T#...
# Add other typed constant checks if necessary (e.g., C#, P#)
elif const_val_direct and scope == "LiteralConstant":
info["type"] = "constant"
info["value"] = const_val_direct[0].strip()
# Infer datatype for literals
val_lower = info["value"].lower()
if val_lower in ["true", "false"]:
info["datatype"] = "Bool"
elif info["value"].isdigit() or (
info["value"].startswith("-") and info["value"][1:].isdigit()
):
info["datatype"] = "Int" # Could be DInt etc, Int is safe default
elif "." in info["value"] or "e" in val_lower:
try:
float(info["value"])
info["datatype"] = "Real" # Could be LReal
except ValueError:
info["datatype"] = "String" # If float conversion fails
else:
info["datatype"] = "String" # Default literal type
# If still unknown, log warning
if info["type"] == "unknown_access":
# Don't warn for Constant scope as it might be handled later
if scope != "Constant":
print(
f"Advertencia: Access UID={uid} scope={scope} no es Symbol ni Constant reconocible."
)
# Ensure variable has a name
if info["type"] == "variable" and not info.get("name"):
print(f"Error Interno: parse_access var sin nombre UID {uid}.")
info["type"] = "error_no_name"
return info
def parse_part(part_element):
"""Parses Part elements (standard instructions), extracting UID, name, template values, and negated pins."""
if part_element is None:
return None
uid = part_element.get("UId")
name_orig = part_element.get("Name") # Instruction type (e.g., Contact, Coil, Move)
if not uid or not name_orig:
print(
f"Error: Part sin UID o Name: {etree.tostring(part_element, encoding='unicode')}"
)
return None
template_values = {}
try:
for tv in part_element.xpath("./*[local-name()='TemplateValue']"):
tv_name = tv.get("Name")
tv_type = tv.get("Type")
if tv_name and tv_type:
template_values[tv_name] = tv_type
except Exception as e:
print(f"Advertencia: Error extrayendo TemplateValues Part UID={uid}: {e}")
negated_pins = {}
try:
for negated_elem in part_element.xpath("./*[local-name()='Negated']"):
negated_pin_name = negated_elem.get("Name")
if negated_pin_name:
negated_pins[negated_pin_name] = True
except Exception as e:
print(f"Advertencia: Error extrayendo Negated Pins Part UID={uid}: {e}")
version = part_element.get("Version")
# Map XML names to internal types used by x2_process
name_mapped = name_orig
if name_orig == "Se":
name_mapped = "TON_S5"
elif name_orig == "Sd":
name_mapped = "TONR_S5"
elif name_orig == "PBox":
name_mapped = "P_TRIG"
elif name_orig == "NBox":
name_mapped = "N_TRIG"
elif name_orig == "RCoil":
name_mapped = "R"
elif name_orig == "SCoil":
name_mapped = "S"
elif name_orig == "SdCoil":
name_mapped = "SR" # Map S5 Set-Dominant to SR internal type
elif name_orig == "BLKMOV":
name_mapped = "BLKMOV" # Keep as is
# Add other mappings if necessary (e.g., RsCoil -> RS)
part_data = {
"uid": uid,
"type": name_mapped, # Use the mapped type
"original_type": name_orig, # Store original name for reference if needed
"template_values": template_values,
"negated_pins": negated_pins,
}
if version:
part_data["version"] = version
return part_data
# parse_call - No changes needed from previous corrected version
def parse_call(call_element):
"""Parses Call elements (FC/FB calls)."""
if call_element is None:
return None
uid = call_element.get("UId")
if not uid:
print(
f"Error: Call encontrado sin UID: {etree.tostring(call_element, encoding='unicode')}"
)
return None
# Use local-name() for CallInfo
call_info_elem = call_element.xpath("./*[local-name()='CallInfo']")
if not call_info_elem:
print(f"Error: Call UID {uid} sin elemento CallInfo.")
return None
call_info = call_info_elem[0]
block_name = call_info.get("Name")
block_type = call_info.get("BlockType") # FC, FB
if not block_name or not block_type:
print(f"Error: CallInfo para UID {uid} sin Name o BlockType.")
return None
call_data = {
"uid": uid,
"type": "Call", # Generic type for our JSON
"block_name": block_name,
"block_type": block_type,
"template_values": {}, # Add fields for consistency with parse_part
"negated_pins": {},
}
# Instance info for FBs
instance_name = None
if block_type == "FB":
# Use local-name() for Instance and Symbol
instance_elem = call_info.xpath("./*[local-name()='Instance']")
if instance_elem:
symbol_elem = instance_elem[0].xpath("./*[local-name()='Symbol']")
if symbol_elem:
instance_name = get_symbol_name(symbol_elem[0])
if instance_name:
call_data["instance_db"] = (
instance_name # Store the formatted name directly
)
return call_data
# --- Function parse_network (Main logic per network) ---
def parse_network(network_element):
if network_element is None:
return {
"id": "ERROR",
"title": "Invalid Network Element",
"logic": [],
"error": "Input element was None",
}
network_id = network_element.get("ID")
title_node = network_element.xpath(
".//*[local-name()='MultilingualText'][@CompositionName='Title']"
)
network_title = (
get_multilingual_text(title_node[0]) if title_node else f"Network {network_id}"
)
comment_node = network_element.xpath(
".//*[local-name()='MultilingualText'][@CompositionName='Comment']"
)
network_comment = get_multilingual_text(comment_node[0]) if comment_node else ""
flgnet_list = network_element.xpath(".//flg:FlgNet", namespaces=ns)
if not flgnet_list:
return {
"id": network_id,
"title": network_title,
"comment": network_comment,
"logic": [],
"error": "FlgNet not found",
}
flgnet = flgnet_list[0]
# 1. Parse Access, Parts, and Calls
access_map = {}
for acc in flgnet.xpath(".//flg:Access", namespaces=ns):
if acc_info := parse_access(acc):
access_map[acc_info["uid"]] = acc_info
parts_and_calls_map = {}
instruction_elements = flgnet.xpath(".//flg:Part | .//flg:Call", namespaces=ns)
for element in instruction_elements:
parsed_info = None
if element.tag == etree.QName(ns["flg"], "Part"):
parsed_info = parse_part(element)
elif element.tag == etree.QName(ns["flg"], "Call"):
parsed_info = parse_call(element)
if parsed_info and "uid" in parsed_info:
parts_and_calls_map[parsed_info["uid"]] = parsed_info
# 2. Parse Wires
wire_connections = defaultdict(list)
source_connections = defaultdict(list)
flg_ns_uri = ns["flg"]
for wire in flgnet.xpath(".//flg:Wire", namespaces=ns):
source_uid, source_pin, dest_uid, dest_pin = None, None, None, None
source_elem = wire.xpath("./*[1]")[0]
dest_elem = wire.xpath("./*[2]")[0]
if source_elem.tag == etree.QName(flg_ns_uri, "Powerrail"):
source_uid, source_pin = "POWERRAIL", "out"
elif source_elem.tag == etree.QName(flg_ns_uri, "IdentCon"):
source_uid, source_pin = source_elem.get("UId"), "value"
elif source_elem.tag == etree.QName(flg_ns_uri, "NameCon"):
source_uid, source_pin = source_elem.get("UId"), source_elem.get("Name")
if dest_elem.tag == etree.QName(flg_ns_uri, "IdentCon"):
dest_uid, dest_pin = dest_elem.get("UId"), "value"
elif dest_elem.tag == etree.QName(flg_ns_uri, "NameCon"):
dest_uid, dest_pin = dest_elem.get("UId"), dest_elem.get("Name")
elif dest_elem.tag == etree.QName(flg_ns_uri, "OpenCon"):
dest_uid, dest_pin = "OPEN", "in"
if dest_uid and dest_pin and source_uid is not None and source_pin is not None:
if dest_uid != "OPEN":
dest_key = (dest_uid, dest_pin)
source_info = (source_uid, source_pin)
if source_info not in wire_connections[dest_key]:
wire_connections[dest_key].append(source_info)
source_key = (source_uid, source_pin)
dest_info = (dest_uid, dest_pin)
if dest_info not in source_connections[source_key]:
source_connections[source_key].append(dest_info)
# 3. Build Initial Logic Representation
all_logic_steps = {}
for instr_uid, instr_info in parts_and_calls_map.items():
instruction_repr = {
"instruction_uid": instr_uid,
**instr_info,
"inputs": {},
"outputs": {},
}
# *** ADD DSTBLK to possible inputs ***
possible_input_pins = {
"en",
"in",
"in1",
"in2",
"in3",
"in4",
"operand",
"bit",
"pre",
"clk",
"s",
"tv",
"r",
"S",
"R1",
"SRCBLK",
"DSTBLK",
}
for dest_pin_name in possible_input_pins:
dest_key = (instr_uid, dest_pin_name)
if dest_key in wire_connections:
sources_list = wire_connections[dest_key]
input_sources_repr = []
for source_uid, source_pin in sources_list:
if source_uid == "POWERRAIL":
input_sources_repr.append({"type": "powerrail"})
elif source_uid in access_map:
input_sources_repr.append(access_map[source_uid])
elif source_uid in parts_and_calls_map:
input_sources_repr.append(
{
"type": "connection",
"source_instruction_type": parts_and_calls_map[
source_uid
]["type"],
"source_instruction_uid": source_uid,
"source_pin": source_pin,
}
)
else:
input_sources_repr.append(
{"type": "unknown_source", "uid": source_uid}
)
if len(input_sources_repr) == 1:
instruction_repr["inputs"][dest_pin_name] = input_sources_repr[0]
elif len(input_sources_repr) > 1:
instruction_repr["inputs"][dest_pin_name] = input_sources_repr
possible_output_pins = {"out", "out1", "Q", "eno", "RET_VAL", "q", "et"}
for src_pin_name in possible_output_pins:
source_key = (instr_uid, src_pin_name)
if source_key in source_connections:
dest_access_list = []
for dest_uid, dest_pin in source_connections[source_key]:
if dest_uid in access_map:
if access_map[dest_uid] not in dest_access_list:
dest_access_list.append(access_map[dest_uid])
if dest_access_list:
instruction_repr["outputs"][src_pin_name] = dest_access_list
all_logic_steps[instr_uid] = instruction_repr
# 4. EN Connection Inference
functional_block_types = {
"Move",
"Add",
"Sub",
"Mul",
"Div",
"Mod",
"Convert",
"Call",
"BLKMOV",
}
# Use MAPPED types for RLO generators
rlo_generators = {
"Contact",
"O",
"Eq",
"Ne",
"Gt",
"Lt",
"Ge",
"Le",
"And",
"Xor",
"P_TRIG",
"N_TRIG",
}
try:
sorted_uids = sorted(all_logic_steps.keys(), key=lambda x: int(x))
except ValueError:
sorted_uids = sorted(all_logic_steps.keys())
processed_for_en_inference = set()
current_logic_list_for_en = [all_logic_steps[uid] for uid in sorted_uids]
for i, instruction in enumerate(current_logic_list_for_en):
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"] # Use the mapped type
if (
instr_type in functional_block_types
and "en" not in instruction["inputs"]
and instr_uid not in processed_for_en_inference
):
inferred_en_source = None
if i > 0:
# Simple lookback to previous instruction
prev_instr = current_logic_list_for_en[i - 1]
prev_uid = prev_instr["instruction_uid"]
prev_type = prev_instr["type"]
# Check if previous instruction has a mappable 'out' or 'eno'
# We check source_connections map for actual wire existence
prev_has_out_wire = any(
dest[0] != "OPEN"
for dest in source_connections.get((prev_uid, "out"), [])
)
prev_has_eno_wire = any(
dest[0] != "OPEN"
for dest in source_connections.get((prev_uid, "eno"), [])
)
if prev_type in rlo_generators and prev_has_out_wire:
inferred_en_source = {
"type": "connection",
"source_instruction_uid": prev_uid,
"source_instruction_type": prev_type,
"source_pin": "out",
}
elif prev_type in functional_block_types and prev_has_eno_wire:
inferred_en_source = {
"type": "connection",
"source_instruction_uid": prev_uid,
"source_instruction_type": prev_type,
"source_pin": "eno",
}
if inferred_en_source:
all_logic_steps[instr_uid]["inputs"]["en"] = inferred_en_source
processed_for_en_inference.add(instr_uid)
# 5. Final Logic Ordering
network_logic = [all_logic_steps[uid] for uid in sorted_uids]
return {
"id": network_id,
"title": network_title,
"comment": network_comment,
"logic": network_logic,
}
# --- Main XML to JSON Conversion Function ---
# convert_xml_to_json - No significant changes needed from previous version
def convert_xml_to_json(xml_filepath, json_filepath):
print(f"Iniciando conversión de '{xml_filepath}' a '{json_filepath}'...")
if not os.path.exists(xml_filepath):
print(f"Error Crítico: Archivo XML no encontrado: '{xml_filepath}'")
return
try:
print("Paso 1: Parseando archivo XML...")
# Disable DTD loading for security and compatibility
parser = etree.XMLParser(
remove_blank_text=True, load_dtd=False, resolve_entities=False
)
tree = etree.parse(xml_filepath, parser)
root = tree.getroot()
print("Paso 1: Parseo XML completado.")
# Detect block type (FC or FB) - Look for SW.Blocks.FC or SW.Blocks.FB
block_node = None
block_xpath = ".//*[local-name()='SW.Blocks.FC' or local-name()='SW.Blocks.FB']"
block_list = root.xpath(block_xpath)
if not block_list:
print("Error Crítico: No se encontró <SW.Blocks.FC> o <SW.Blocks.FB>.")
return
block_node = block_list[0]
block_tag_name = etree.QName(
block_node.tag
).localname # SW.Blocks.FC or SW.Blocks.FB
print(
f"Paso 2: Bloque {block_tag_name} encontrado (ID={block_node.get('ID')})."
)
print("Paso 3: Extrayendo atributos del bloque...")
attribute_list_node = block_node.xpath("./*[local-name()='AttributeList']")
block_name_val, block_number_val, block_lang_val = "Unknown", None, "Unknown"
if attribute_list_node:
attr_list = attribute_list_node[0]
name_node = attr_list.xpath("./*[local-name()='Name']/text()")
block_name_val = name_node[0].strip() if name_node else block_name_val
num_node = attr_list.xpath("./*[local-name()='Number']/text()")
try:
block_number_val = int(num_node[0]) if num_node else None
except ValueError:
block_number_val = None # Handle non-integer Number
lang_node = attr_list.xpath(
"./*[local-name()='ProgrammingLanguage']/text()"
)
block_lang_val = lang_node[0].strip() if lang_node else block_lang_val
print(
f"Paso 3: Atributos: Nombre='{block_name_val}', Número={block_number_val}, Lenguaje='{block_lang_val}'"
)
else:
print("Advertencia: No se encontró AttributeList para el bloque.")
# Get block comment
block_comment_val = ""
comment_node_list = block_node.xpath(
"./*[local-name()='ObjectList']/*[local-name()='MultilingualText'][@CompositionName='Comment']"
)
if comment_node_list:
block_comment_val = get_multilingual_text(comment_node_list[0])
# Initialize result dictionary
result = {
"block_name": block_name_val,
"block_number": block_number_val,
"language": block_lang_val,
"block_comment": block_comment_val,
"interface": {},
"networks": [],
}
print("Paso 4: Extrayendo la interfaz del bloque...")
if attribute_list_node:
interface_node = attribute_list_node[0].xpath(
"./*[local-name()='Interface']"
)
if interface_node:
print("Paso 4: Nodo Interface encontrado.")
# Iterate through sections using the correct namespace prefix 'iface'
for section in interface_node[0].xpath(
".//iface:Section", namespaces=ns
):
section_name = section.get(
"Name"
) # Input, Output, InOut, Temp, Constant, Return
members = []
for member in section.xpath("./iface:Member", namespaces=ns):
member_name = member.get("Name")
member_dtype = member.get("Datatype")
if member_name and member_dtype:
member_info = {
"name": member_name,
"datatype": member_dtype,
}
members.append(member_info)
if members:
result["interface"][section_name] = members
if not result["interface"]:
print("Advertencia: Interface sin secciones iface:Section válidas.")
else:
print(
"Advertencia: No se encontró <Interface> DENTRO de <AttributeList>."
)
if not result["interface"]:
print("Advertencia: No se pudo extraer información de la interfaz.")
print("Paso 5: Extrayendo y PROCESANDO lógica de redes (CompileUnits)...")
networks_processed_count = 0
object_list_node = block_node.xpath("./*[local-name()='ObjectList']")
if object_list_node:
compile_units = object_list_node[0].xpath(
"./*[local-name()='SW.Blocks.CompileUnit']"
)
print(
f"Paso 5: Se encontraron {len(compile_units)} elementos SW.Blocks.CompileUnit."
)
for network_elem in compile_units:
networks_processed_count += 1
parsed_network = parse_network(network_elem)
if parsed_network and parsed_network.get("error") is None:
result["networks"].append(parsed_network)
elif parsed_network:
print(
f"Error: Falló parseo red ID={parsed_network.get('id')}: {parsed_network.get('error')}"
)
result["networks"].append(
parsed_network
) # Include network with error marker
else:
print(
f"Error Crítico: parse_network devolvió None para CompileUnit (ID={network_elem.get('ID')})."
)
if networks_processed_count == 0:
print("Advertencia: ObjectList sin SW.Blocks.CompileUnit.")
else:
print("Advertencia: No se encontró ObjectList.")
print("Paso 6: Escribiendo el resultado en el archivo JSON...")
if not result["interface"]:
print("ADVERTENCIA FINAL: 'interface' está vacía.")
if not result["networks"]:
print("ADVERTENCIA FINAL: 'networks' está vacía.")
try:
with open(json_filepath, "w", encoding="utf-8") as f:
json.dump(
result, f, indent=4, ensure_ascii=False
) # ensure_ascii=False is important
print(f"Paso 6: Escritura completada.")
print(f"Conversión finalizada. JSON guardado en: '{json_filepath}'")
except IOError as e:
print(
f"Error Crítico: No se pudo escribir JSON en '{json_filepath}'. Error: {e}"
)
except TypeError as e:
print(f"Error Crítico: Problema al serializar a JSON. Error: {e}")
except etree.XMLSyntaxError as e:
print(f"Error Crítico: Sintaxis XML en '{xml_filepath}'. Detalles: {e}")
except Exception as e:
print(f"Error Crítico: Error inesperado durante la conversión: {e}")
print("--- Traceback ---")
traceback.print_exc()
print("--- Fin Traceback ---")
# --- Punto de Entrada Principal ---
if __name__ == "__main__":
xml_file = "BlenderCtrl__Main.xml"
json_file = xml_file.replace(".xml", "_simplified.json")
convert_xml_to_json(xml_file, json_file)

File diff suppressed because it is too large Load Diff

View File

@ -1,218 +0,0 @@
# -*- coding: utf-8 -*-
import json
import os
import re
# --- Helper Functions ---
# Use the CORRECT version from x2_process.py
def format_variable_name(name):
"""Formats variable names for SCL, preserving quotes for structured names."""
if not name:
return "_INVALID_NAME_"
if name.startswith('"') and name.endswith('"'):
return name
prefix = ""
if name.startswith("#"):
prefix = "#"
name = name[1:]
if not name:
return "_INVALID_NAME_"
if not re.match(r"^[a-zA-Z_]", name[0]):
name = "_" + name
name = re.sub(r"[^a-zA-Z0-9_]", "_", name)
return prefix + name
def generate_scl(processed_json_filepath, output_scl_filepath):
"""Genera un archivo SCL a partir del JSON procesado."""
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 JSON: {e}")
return
# --- Block Info ---
block_name_orig = data.get("block_name", "UnknownBlock")
block_number = data.get("block_number")
block_lang = data.get("language", "LAD")
block_comment = data.get("block_comment", "")
# --- Variable Detection ---
temp_vars_base = set() # Base names like _temp_...
stat_vars = {} # Quoted Name -> TYPE (Bool, TON, TONR)
temp_pattern = re.compile(r"#(_temp_[a-zA-Z0-9_]+)") # Capture base name after #
# Regex needs to capture the QUOTED name from SCL
stat_pattern_bool = re.compile(
r'("stat_(?:ptrig|ntrig|sr)_mem_[a-zA-Z0-9_]+")'
) # Edge/SR mem bits
stat_pattern_ton = re.compile(r'("stat_TON_[a-zA-Z0-9_]+")') # TON instances
stat_pattern_tonr = re.compile(r'("stat_TONR_[a-zA-Z0-9_]+")') # TONR instances
for network in data.get("networks", []):
for instruction in network.get("logic", []):
scl_code = instruction.get("scl", "")
if scl_code:
temp_vars_base.update(temp_pattern.findall(scl_code))
for name in stat_pattern_bool.findall(scl_code):
stat_vars[name] = "Bool"
for name in stat_pattern_ton.findall(scl_code):
stat_vars[name] = "TON"
for name in stat_pattern_tonr.findall(scl_code):
stat_vars[name] = "TONR"
has_stat = bool(stat_vars)
interface_temps_list = data.get("interface", {}).get("Temp", [])
has_temp = bool(temp_vars_base or interface_temps_list)
block_type_keyword = "FUNCTION_BLOCK" if has_stat or has_temp else "FUNCTION"
scl_block_name = format_variable_name(block_name_orig) # Format name correctly
print(f"Generando SCL para bloque: {scl_block_name} como {block_type_keyword}")
print(f"Variables temporales (#_temp_...) detectadas: {len(temp_vars_base)}")
print(f"Variables estáticas (stat_...) detectadas: {len(stat_vars)}")
# --- Build SCL Output ---
scl_output = []
scl_output.append(f"// Block Name (Original): {block_name_orig}")
if block_number:
scl_output.append(f"// Block Number: {block_number}")
scl_output.append(f"// Original Language: {block_lang}")
if block_comment:
scl_output.append(f"// Block Comment: {block_comment}")
scl_output.append("")
scl_output.append(f"{block_type_keyword} {scl_block_name}")
scl_output.append("{ S7_Optimized_Access := 'TRUE' }")
scl_output.append("VERSION : 0.1")
scl_output.append("")
# --- VAR Sections ---
def add_var_section(section_name, members_list):
if not members_list:
return
scl_output.append(f"VAR_{section_name}")
for member in members_list:
scl_name = format_variable_name(
member["name"]
) # Format interface var names
scl_output.append(f" {scl_name} : {member['datatype']};")
scl_output.append("END_VAR")
scl_output.append("")
interface_data = data.get("interface", {})
add_var_section("INPUT", interface_data.get("Input", []))
add_var_section("OUTPUT", interface_data.get("Output", []))
add_var_section("IN_OUT", interface_data.get("InOut", []))
if stat_vars:
scl_output.append("VAR_STAT")
for var_name_quoted in sorted(stat_vars.keys()):
var_type = stat_vars[var_name_quoted]
comment = (
f"// Instance for {var_type}"
if var_type in ["TON", "TONR"]
else "// Memory Bit"
)
scl_output.append(f" {var_name_quoted} : {var_type}; {comment}")
scl_output.append("END_VAR")
scl_output.append("")
declared_temps_formatted = set()
temp_declarations = []
if interface_temps_list:
for var in interface_temps_list:
scl_name = format_variable_name(var["name"])
temp_declarations.append(
f" {scl_name} : {var['datatype']}; // From Interface"
)
declared_temps_formatted.add(scl_name)
if temp_vars_base:
for base_name in sorted(list(temp_vars_base)):
# Declare using the base name, quoted
scl_name_declare = format_variable_name(f'"{base_name}"')
if scl_name_declare not in declared_temps_formatted:
# Simple inference based on common pin names in the base name
inferred_type = "Bool" # Default
if "ret_val" in base_name:
inferred_type = "Int"
elif "_et" in base_name:
inferred_type = "Time"
temp_declarations.append(
f" {scl_name_declare} : {inferred_type}; // Auto-generated temporary"
)
declared_temps_formatted.add(scl_name_declare)
if temp_declarations:
scl_output.append("VAR_TEMP")
scl_output.extend(temp_declarations)
scl_output.append("END_VAR")
scl_output.append("")
# --- Block Body ---
scl_output.append("BEGIN")
scl_output.append("")
for i, network in enumerate(data.get("networks", [])):
network_title = network.get("title", f'Network {network.get("id")}')
network_comment = network.get("comment", "")
scl_output.append(f" // Network {i+1}: {network_title}")
if network_comment:
for line in network_comment.splitlines():
scl_output.append(f" // {line}")
scl_output.append("")
network_has_code = False
for instruction in network.get("logic", []):
if instruction.get("grouped", False):
continue
scl_code = instruction.get("scl")
if scl_code:
lines_to_add = []
for line in scl_code.splitlines():
line_strip = line.strip()
is_simple_info_comment = line_strip.startswith(
("// RLO:", "// Comparison Eq", "// Logic O")
)
# Keep essential comments (Errors, Grouping, Memory Updates) and actual SCL
if (
not is_simple_info_comment
or line_strip.startswith("// ERROR")
or line_strip.startswith(GROUPED_COMMENT)
or "Edge Memory Update" in line_strip
):
lines_to_add.append(line)
if lines_to_add:
network_has_code = True
for line in lines_to_add:
scl_output.append(f" {line}")
if network_has_code:
scl_output.append("")
end_keyword = (
"END_FUNCTION_BLOCK"
if block_type_keyword == "FUNCTION_BLOCK"
else "END_FUNCTION"
)
scl_output.append(end_keyword)
# --- Write File ---
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}")
# --- Ejecución ---
if __name__ == "__main__":
xml_filename_base = "BlenderCtrl__Main"
input_json_file = f"{xml_filename_base}_simplified_processed.json"
output_scl_file = input_json_file.replace("_simplified_processed.json", ".scl")
generate_scl(input_json_file, output_scl_file)

View File

@ -11,7 +11,7 @@ PROCESSORS_DIR = "processors"
FILE_HEADER = """# -*- coding: utf-8 -*-
# TODO: Import necessary functions from processor_utils
# from .processor_utils import get_scl_representation, format_variable_name, ...
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
@ -22,8 +22,7 @@ SCL_SUFFIX = "_scl"
# Pie de página estándar con la función get_processor_info de plantilla
def get_file_footer(func_name):
# Intenta adivinar el type_name quitando 'process_'
# Necesitará ajustes manuales para casos como 'call', 'edge_detector', 'comparison', 'math'
"""Generates the standard footer with a placeholder get_processor_info."""
type_name_guess = func_name.replace('process_', '')
return f"""
# --- Function code ends ---
@ -31,8 +30,8 @@ def get_file_footer(func_name):
# --- Processor Information Function ---
def get_processor_info():
\"\"\"Returns the type name and processing function for this module.\"\"\"
# TODO: Adjust the type_name if needed (e.g., for call, edge_detector, comparison, math)
# TODO: Return a list if this module handles multiple types (e.g., PBox/NBox, FC/FB)
# TODO: Adjust the type_name if needed (e.g., call, edge_detector, comparison, math).
# TODO: Return a list if this module handles multiple types (e.g., PBox/NBox, FC/FB).
type_name = "{type_name_guess}" # Basic guess
return {{'type_name': type_name, 'processor_func': {func_name}}}
"""
@ -40,7 +39,8 @@ def get_processor_info():
def extract_and_create_processors(source_py_file):
"""
Extracts top-level functions starting with 'process_' from the source file
and creates individual processor files in the PROCESSORS_DIR.
and creates individual processor files in the PROCESSORS_DIR, copying
the entire function body until the next top-level definition.
"""
if not os.path.exists(source_py_file):
print(f"Error: Source file not found: '{source_py_file}'")
@ -54,66 +54,58 @@ def extract_and_create_processors(source_py_file):
print(f"Error reading source file: {e}")
return
# Crear directorio de procesadores si no existe
os.makedirs(PROCESSORS_DIR, exist_ok=True)
print(f"Ensuring '{PROCESSORS_DIR}' directory exists.")
current_func_name = None
current_func_lines = []
processor_count = 0
print("Searching for processor functions (def process_...):")
# Usamos una expresión regular para encontrar definiciones de función de nivel superior
# que empiecen por 'process_'
func_def_pattern = re.compile(r"^def\s+(process_\w+)\s*\(")
processor_functions = [] # Store tuples of (name, start_line_index, end_line_index)
current_func_start = -1
current_func_name = None
# Pattern to find ANY top-level function definition
any_func_def_pattern = re.compile(r"^def\s+(\w+)\s*\(")
# Pattern specific to processor functions
process_func_def_pattern = re.compile(r"^def\s+(process_\w+)\s*\(")
# First pass: Identify start and end lines of all top-level functions
for i, line in enumerate(lines):
match = func_def_pattern.match(line)
match = any_func_def_pattern.match(line)
if match:
# Found a new top-level function definition
if current_func_name is not None:
# Mark the end of the *previous* function
# Only add if it was a 'process_' function
if current_func_name.startswith("process_"):
processor_functions.append((current_func_name, current_func_start, i))
if match: # Encontrada una nueva definición de función 'process_'
new_func_name = match.group(1)
# Start tracking the new function
current_func_name = match.group(1)
current_func_start = i
# Si estábamos procesando una función anterior, guardarla ahora
if current_func_name:
create_processor_file(current_func_name, current_func_lines)
processor_count += 1
# Add the last function found in the file (if it was a process_ function)
if current_func_name is not None and current_func_name.startswith("process_"):
processor_functions.append((current_func_name, current_func_start, len(lines)))
# Empezar a recolectar para la nueva función
print(f" - Found: {new_func_name}")
current_func_name = new_func_name
current_func_lines = [line] # Empezar con la línea 'def'
# Second pass: Create files using the identified line ranges
processor_count = 0
if not processor_functions:
print("\nWarning: No functions starting with 'process_' found at the top level.")
return
elif line.strip() == "" and not current_func_lines:
# Ignorar líneas en blanco antes de la primera función
continue
print(f"Found {len(processor_functions)} potential processor functions.")
elif current_func_name and not line.startswith(' '):
# Si estamos dentro de una función y encontramos una línea
# que NO empieza con espacio (y no es un 'def process_'),
# podría ser el fin de la función (o una definición de otra cosa).
# Por simplicidad, asumimos que marca el fin. Guardamos la actual.
# (Esto funciona si las 'def process_' están una tras otra o separadas
# por comentarios o definiciones de funciones NO 'process_')
create_processor_file(current_func_name, current_func_lines)
processor_count += 1
current_func_name = None
current_func_lines = []
for func_name, start_idx, end_idx in processor_functions:
print(f" - Processing: {func_name} (lines {start_idx+1}-{end_idx})")
func_lines = lines[start_idx:end_idx] # Extract lines for this function
# Remove trailing blank lines from the extracted block, often happens before next def
while func_lines and func_lines[-1].strip() == "":
func_lines.pop()
elif current_func_name:
# Si estamos recolectando líneas para una función, añadir la línea actual
current_func_lines.append(line)
# Guardar la última función encontrada después de salir del bucle
if current_func_name:
create_processor_file(current_func_name, current_func_lines)
create_processor_file(func_name, func_lines)
processor_count += 1
if processor_count == 0:
print("\nWarning: No functions starting with 'process_' found at the top level.")
else:
print(f"\nFinished processing. Attempted to create/check {processor_count} processor files in '{PROCESSORS_DIR}'.")
print(f"\nFinished processing. Attempted to create/check {processor_count} processor files in '{PROCESSORS_DIR}'.")
def create_processor_file(func_name, func_lines):
"""Creates the individual processor file if it doesn't exist."""
@ -128,7 +120,9 @@ def create_processor_file(func_name, func_lines):
try:
with open(target_filepath, 'w', encoding='utf-8') as f:
f.write(FILE_HEADER)
f.writelines(func_lines) # Escribir las líneas de la función
# Write the function lines, ensuring consistent newline endings
for line in func_lines:
f.write(line.rstrip() + '\n')
f.write(get_file_footer(func_name))
except Exception as e:
print(f" Error writing file '{target_filename}': {e}")

View File

75
processors/process_add.py Normal file
View File

@ -0,0 +1,75 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_add(instruction, network_id, scl_map, access_map, data):
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"]
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
en_input = instruction["inputs"].get("en")
en_scl = (
get_scl_representation(en_input, network_id, scl_map, access_map)
if en_input
else "TRUE"
)
in1_info = instruction["inputs"].get("in1")
in2_info = instruction["inputs"].get("in2")
in1_scl = get_scl_representation(in1_info, network_id, scl_map, access_map)
in2_scl = get_scl_representation(in2_info, network_id, scl_map, access_map)
if en_scl is None or in1_scl is None or in2_scl is None:
return False
target_scl = get_target_scl_name(
instruction, "out", network_id, default_to_temp=True
)
if target_scl is None:
print(f"Error: Sin destino ADD {instr_uid}")
instruction["scl"] = f"// ERROR: Add {instr_uid} sin destino"
instruction["type"] += "_error"
return True
# Formatear operandos si son variables
op1 = (
format_variable_name(in1_scl)
if in1_info and in1_info.get("type") == "variable"
else in1_scl
)
op2 = (
format_variable_name(in2_scl)
if in2_info and in2_info.get("type") == "variable"
else in2_scl
)
# Añadir paréntesis si es necesario
op1 = f"({op1})" if " " in op1 and not op1.startswith("(") else op1
op2 = f"({op2})" if " " in op2 and not op2.startswith("(") else op2
scl_core = f"{target_scl} := {op1} + {op2};"
scl_final = (
f"IF {en_scl} THEN\n {scl_core}\nEND_IF;" if en_scl != "TRUE" else scl_core
)
instruction["scl"] = scl_final
instruction["type"] = instr_type + SCL_SUFFIX
map_key_out = (network_id, instr_uid, "out")
scl_map[map_key_out] = target_scl
map_key_eno = (network_id, instr_uid, "eno")
scl_map[map_key_eno] = en_scl
return True
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para el procesador Add."""
return {'type_name': 'add', 'processor_func': process_add, 'priority': 4}

View File

@ -0,0 +1,112 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_blkmov(instruction, network_id, scl_map, access_map, data):
"""
Genera SCL usando BLKMOV directamente como nombre de función,
sin COUNT y con formato específico, según solicitud del usuario.
ADVERTENCIA: Es MUY PROBABLE que esto NO compile en TIA Portal estándar,
ya que BLKMOV no es una función SCL y MOVE_BLK requiere COUNT.
"""
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"]
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False # Ya procesado o con error
# --- Obtener Entradas ---
en_input = instruction["inputs"].get("en")
en_scl = (
get_scl_representation(en_input, network_id, scl_map, access_map)
if en_input
else "TRUE"
)
srcblk_info = instruction["inputs"].get("SRCBLK")
# ¡IMPORTANTE! Obtenemos el nombre RAW antes de formatearlo para usarlo como pide el usuario
raw_srcblk_name = srcblk_info.get("name") if srcblk_info else None
# Verificar dependencias de entrada (solo necesitamos que EN esté resuelto)
if en_scl is None:
return False # Dependencia EN no lista
if raw_srcblk_name is None:
print(f"Error: BLKMOV {instr_uid} sin información válida para SRCBLK.")
instruction["scl"] = f"// ERROR: BLKMOV {instr_uid} sin SRCBLK válido."
instruction["type"] += "_error"
return True
# --- Obtener Destinos (Salidas) ---
# RET_VAL (Usamos get_target_scl_name para manejar variables temporales si es necesario)
retval_target_scl = get_target_scl_name(
instruction, "RET_VAL", network_id, default_to_temp=True
)
if retval_target_scl is None:
print(f"Error: BLKMOV {instr_uid} sin destino claro para RET_VAL.")
instruction["scl"] = f"// ERROR: BLKMOV {instr_uid} sin destino RET_VAL"
instruction["type"] += "_error"
return True
# DSTBLK (Obtenemos el nombre RAW para usarlo como pide el usuario)
raw_dstblk_name = None
dstblk_output_list = instruction.get("outputs", {}).get("DSTBLK", [])
if dstblk_output_list and isinstance(dstblk_output_list, list) and len(dstblk_output_list) == 1:
dest_access = dstblk_output_list[0]
if dest_access.get("type") == "variable":
raw_dstblk_name = dest_access.get("name") # Nombre raw del JSON
else:
print(f"Advertencia: Destino DSTBLK de BLKMOV {instr_uid} no es una variable (Tipo: {dest_access.get('type')}).")
else:
print(f"Error: No se encontró un destino único y válido para DSTBLK en BLKMOV {instr_uid}.")
if raw_dstblk_name is None:
instruction["scl"] = f"// ERROR: BLKMOV {instr_uid} sin destino DSTBLK válido."
instruction["type"] += "_error"
return True
# --- Formateo especial para SRCBLK/DSTBLK como pidió el usuario ---
# Asume formato "DB".Variable o "Struct".Variable del JSON y lo mantiene
# (Esto anula la limpieza normal de format_variable_name para estos parámetros)
srcblk_final_str = raw_srcblk_name if raw_srcblk_name else "_ERROR_SRC_"
dstblk_final_str = raw_dstblk_name if raw_dstblk_name else "_ERROR_DST_"
# --- Generar SCL Exacto Solicitado ---
scl_core = (
f"{retval_target_scl} := BLKMOV(SRCBLK := {srcblk_final_str}, "
f"DSTBLK => {dstblk_final_str}); "
f"// ADVERTENCIA: BLKMOV usado directamente, probablemente no compile!"
)
# Añadir condición EN (usando la representación SCL obtenida para EN)
scl_final = (
f"IF {en_scl} THEN\n {scl_core}\nEND_IF;" if en_scl != "TRUE" else scl_core
)
# --- Actualizar Instrucción y Mapa SCL ---
instruction["scl"] = scl_final
instruction["type"] = instr_type + SCL_SUFFIX
# Propagar ENO (igual que EN)
map_key_eno = (network_id, instr_uid, "eno")
scl_map[map_key_eno] = en_scl
# Propagar el valor de retorno (el contenido de la variable asignada a RET_VAL)
map_key_ret_val = (network_id, instr_uid, "RET_VAL")
scl_map[map_key_ret_val] = retval_target_scl # El valor es lo que sea que se asigne
return True
# ... (Asegúrate de que esta función está registrada en processor_map como antes) ...
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para el procesador BLKMOV."""
# Asumiendo que 'BLKMOV' es el type en el JSON simplificado
return {'type_name': 'blkmov', 'processor_func': process_blkmov, 'priority': 6}

141
processors/process_call.py Normal file
View File

@ -0,0 +1,141 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_call(instruction, network_id, scl_map, access_map, data):
instr_uid = instruction["instruction_uid"]
instr_type = instruction.get("type", "") # Usar get con default
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
block_name = instruction.get("block_name", f"UnknownCall_{instr_uid}")
block_type = instruction.get("block_type") # FC, FB
instance_db = instruction.get("instance_db") # Nombre del DB de instancia (para FB)
# Formatear nombres
block_name_scl = format_variable_name(block_name)
instance_db_scl = format_variable_name(instance_db) if instance_db else None
# --- Manejo de EN ---
en_input = instruction["inputs"].get("en")
en_scl = (
get_scl_representation(en_input, network_id, scl_map, access_map)
if en_input
else "TRUE"
)
if en_scl is None:
return False # Dependencia EN no resuelta
# --- Procesar Parámetros de Entrada/Salida ---
# Necesitamos iterar sobre los pines definidos en la interfaz del bloque llamado.
# Esta información no está directamente en la instrucción 'Call' del JSON simplificado.
# ¡Limitación! Sin la interfaz del bloque llamado, solo podemos manejar EN/ENO
# y asumir una llamada sin parámetros o con parámetros conectados implícitamente.
# Solución temporal: Buscar conexiones en 'inputs' y 'outputs' que NO sean 'en'/'eno'
# y construir la llamada basándose en eso. Esto es muy heurístico.
scl_call_params = []
processed_inputs = {"en"} # Marcar 'en' como ya procesado
for pin_name, source_info in instruction.get("inputs", {}).items():
if pin_name not in processed_inputs:
param_scl = get_scl_representation(
source_info, network_id, scl_map, access_map
)
if param_scl is None:
# print(f"DEBUG: Call {instr_uid} esperando parámetro de entrada {pin_name}")
return False # Dependencia de parámetro no resuelta
# Formatear si es variable
param_scl_formatted = (
format_variable_name(param_scl)
if source_info.get("type") == "variable"
else param_scl
)
scl_call_params.append(
f"{format_variable_name(pin_name)} := {param_scl_formatted}"
)
processed_inputs.add(pin_name)
# Procesar parámetros de salida (asignaciones después de la llamada o pasados como VAR_IN_OUT/VAR_OUTPUT)
# Esto es aún más complejo. SCL normalmente asigna salidas después o usa punteros/referencias.
# Simplificación: Asumir que las salidas se manejan por asignación posterior si es necesario,
# o que son VAR_OUTPUT y se acceden como instancia.salida.
# Por ahora, no generamos asignaciones explícitas para las salidas aquí.
# --- Construcción de la Llamada SCL ---
scl_call_body = ""
param_string = ", ".join(scl_call_params)
if block_type == "FB":
if not instance_db_scl:
print(
f"Error: Llamada a FB '{block_name_scl}' (UID {instr_uid}) sin DB de instancia especificado."
)
instruction["scl"] = f"// ERROR: FB Call {block_name_scl} sin instancia"
instruction["type"] = "Call_FB_error"
return True # Procesado con error
# Llamada a FB con DB de instancia
scl_call_body = f"{instance_db_scl}({param_string});"
elif block_type == "FC":
# Llamada a FC
scl_call_body = f"{block_name_scl}({param_string});"
else:
print(
f"Advertencia: Tipo de bloque no soportado para Call UID {instr_uid}: {block_type}"
)
scl_call_body = f"// ERROR: Call a bloque tipo '{block_type}' no soportado: {block_name_scl}"
instruction["type"] = f"Call_{block_type}_error" # Marcar como error parcial
# --- Aplicar Condición EN ---
scl_final = ""
if en_scl != "TRUE":
# Indentar la llamada dentro del IF
indented_call = "\n".join([f" {line}" for line in scl_call_body.splitlines()])
scl_final = f"IF {en_scl} THEN\n{indented_call}\nEND_IF;"
else:
scl_final = scl_call_body
# --- Actualizar JSON y Mapa SCL ---
instruction["scl"] = scl_final
instruction["type"] = (
f"Call_{block_type}_scl"
if "_error" not in instruction["type"]
else instruction["type"]
)
# Actualizar scl_map con el estado ENO (igual a EN para llamadas simples sin manejo explícito de ENO)
map_key_eno = (network_id, instr_uid, "eno")
scl_map[map_key_eno] = en_scl
# Propagar valores de salida (si pudiéramos determinarlos)
# Ejemplo: Si supiéramos que hay una salida 'Out1' de tipo INT
# map_key_out1 = (network_id, instr_uid, "Out1")
# if block_type == "FB" and instance_db_scl:
# scl_map[map_key_out1] = f"{instance_db_scl}.Out1" # Acceso a salida de instancia
# else:
# # Para FCs, necesitaríamos una variable temporal o asignación explícita
# temp_out1 = generate_temp_var_name(network_id, instr_uid, "Out1")
# # Modificar scl_call_body para incluir la asignación: Out1 => temp_out1
# scl_map[map_key_out1] = temp_out1
return True
# --- Procesador de Temporizadores (TON, TOF) ---
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para las llamadas a FC y FB."""
# Esta función maneja tanto FC como FB. El despachador en x2_process.py
# usará 'call_fc' o 'call_fb' como clave basada en block_type.
return [
{'type_name': 'call_fc', 'processor_func': process_call, 'priority': 6},
{'type_name': 'call_fb', 'processor_func': process_call, 'priority': 6}
]

View File

@ -0,0 +1,79 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_coil(instruction, network_id, scl_map, access_map, data):
"""Genera la asignación para Coil. Si la entrada viene de PBox/NBox,
añade la actualización de memoria del flanco DESPUÉS de la asignación."""
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"]
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
coil_input_info = instruction["inputs"].get("in")
operand_info = instruction["inputs"].get("operand")
in_rlo_scl = get_scl_representation(coil_input_info, network_id, scl_map, access_map)
operand_scl = get_scl_representation(operand_info, network_id, scl_map, access_map)
if in_rlo_scl is None or operand_scl is None: return False
if not (operand_info and operand_info.get("type") == "variable"):
instruction["scl"] = f"// ERROR: Coil {instr_uid} operando no es variable o falta info"
instruction["type"] = instr_type + "_error"
return True
operand_scl_formatted = format_variable_name(operand_scl)
if in_rlo_scl == "(TRUE)": in_rlo_scl = "TRUE"
elif in_rlo_scl == "(FALSE)": in_rlo_scl = "FALSE"
# Generar la asignación SCL principal de la bobina
scl_assignment = f"{operand_scl_formatted} := {in_rlo_scl};"
scl_final = scl_assignment # Inicializar SCL final
# --- Lógica para añadir actualización de memoria de flancos ---
mem_update_scl_combined = None
if isinstance(coil_input_info, dict) and coil_input_info.get("type") == "connection":
source_uid = coil_input_info.get("source_instruction_uid")
source_pin = coil_input_info.get("source_pin")
# Buscar la instrucción fuente PBox/NBox
source_instruction = None
network_logic = next((net["logic"] for net in data["networks"] if net["id"] == network_id), [])
for instr in network_logic:
if instr.get("instruction_uid") == source_uid:
source_instruction = instr
break
if source_instruction:
source_type = source_instruction.get("type","").replace('_scl','').replace('_error','')
# Si la fuente es PBox o NBox y tiene el campo temporal con la actualización
if source_type in ["PBox", "NBox"] and '_edge_mem_update_scl' in source_instruction:
mem_update_scl_combined = source_instruction.get('_edge_mem_update_scl') # Obtener update+comment
if mem_update_scl_combined:
# Añadir la actualización DESPUÉS de la asignación de la bobina, USANDO \n
scl_final = f"{scl_assignment}\n{mem_update_scl_combined}"
# Marcar la instrucción PBox/NBox para que x3 no escriba su SCL (que ahora está vacío/comentario)
source_instruction['scl'] = f"// Logic moved to Coil {instr_uid}" # Actualizar PBox/NBox SCL
instruction["scl"] = scl_final
instruction["type"] = instr_type + SCL_SUFFIX
return True
# EN x2_process.py, junto a otras funciones process_xxx
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para el procesador Coil."""
return {'type_name': 'coil', 'processor_func': process_coil, 'priority': 3}

View File

@ -0,0 +1,81 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_comparison(instruction, network_id, scl_map, access_map, data):
"""
Genera la expresión SCL para Comparadores (GT, LT, GE, LE, NE).
El resultado se propaga por scl_map['out'].
"""
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"] # GT, LT, GE, LE, NE
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
# Mapa de tipos a operadores SCL
op_map = {"GT": ">", "LT": "<", "GE": ">=", "LE": "<=", "NE": "<>"}
scl_operator = op_map.get(instr_type)
if not scl_operator:
instruction["scl"] = f"// ERROR: Tipo de comparación no soportado: {instr_type}"
instruction["type"] += "_error"
return True
# Obtener operandos
in1_info = instruction["inputs"].get("in1")
in2_info = instruction["inputs"].get("in2")
in1_scl = get_scl_representation(in1_info, network_id, scl_map, access_map)
in2_scl = get_scl_representation(in2_info, network_id, scl_map, access_map)
if in1_scl is None or in2_scl is None:
return False # Dependencias no listas
# Formatear operandos si son variables
op1 = format_variable_name(in1_scl) if in1_info and in1_info.get("type") == "variable" else in1_scl
op2 = format_variable_name(in2_scl) if in2_info and in2_info.get("type") == "variable" else in2_scl
# Añadir paréntesis si contienen espacios (poco probable tras formatear)
op1 = f"({op1})" if " " in op1 and not op1.startswith("(") else op1
op2 = f"({op2})" if " " in op2 and not op2.startswith("(") else op2
comparison_scl = f"{op1} {scl_operator} {op2}"
# Guardar resultado en el mapa para 'out'
map_key_out = (network_id, instr_uid, "out")
scl_map[map_key_out] = f"({comparison_scl})" # Poner paréntesis por seguridad
# Manejar entrada 'pre'/RLO -> ENO (como en EQ)
pre_input = instruction["inputs"].get("pre") # Asumir 'pre' como en EQ
en_scl = get_scl_representation(pre_input, network_id, scl_map, access_map) if pre_input else "TRUE"
if en_scl is None:
return False # Dependencia 'pre'/'en' no lista
map_key_eno = (network_id, instr_uid, "eno")
scl_map[map_key_eno] = en_scl
instruction["scl"] = f"// Comparison {instr_type} {instr_uid}: {comparison_scl}"
instruction["type"] = instr_type + SCL_SUFFIX
return True
# --- Procesador de Matemáticas (ADD ya existe, añadir otros) ---
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para los comparadores (excepto EQ)."""
# Esta función maneja múltiples tipos de comparación.
return [
{'type_name': 'gt', 'processor_func': process_comparison, 'priority': 2}, # >
{'type_name': 'lt', 'processor_func': process_comparison, 'priority': 2}, # <
{'type_name': 'ge', 'processor_func': process_comparison, 'priority': 2}, # >=
{'type_name': 'le', 'processor_func': process_comparison, 'priority': 2}, # <=
{'type_name': 'ne', 'processor_func': process_comparison, 'priority': 2} # <>
]

View File

@ -0,0 +1,73 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_contact(instruction, network_id, scl_map, access_map, data):
"""Traduce Contact (normal o negado) a una expresión booleana SCL."""
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"]
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
# --- INICIO LEER NEGACIÓN ---
# Verificar si el pin 'operand' está marcado como negado en el JSON
is_negated = instruction.get("negated_pins", {}).get("operand", False)
# --- FIN LEER NEGACIÓN ---
# print(f"DEBUG: Intentando procesar CONTACT{' (N)' if is_negated else ''} - UID: {instr_uid} en Red: {network_id}")
in_input = instruction["inputs"].get("in")
in_rlo_scl = (
"TRUE"
if in_input is None
else get_scl_representation(in_input, network_id, scl_map, access_map)
)
operand_scl = get_scl_representation(
instruction["inputs"].get("operand"), network_id, scl_map, access_map
)
if in_rlo_scl is None or operand_scl is None:
return False
# Usar is_negated para aplicar NOT
term = f"NOT {operand_scl}" if is_negated else operand_scl
if not (term.startswith('"') and term.endswith('"')):
# Añadir paréntesis si es NOT o si contiene espacios/operadores
if is_negated or (
" " in term and not (term.startswith("(") and term.endswith(")"))
):
term = f"({term})"
new_rlo_scl = (
term
if in_rlo_scl == "TRUE"
else (
f"({in_rlo_scl}) AND {term}"
if ("AND" in in_rlo_scl or "OR" in in_rlo_scl)
and not (in_rlo_scl.startswith("(") and in_rlo_scl.endswith(")"))
else f"{in_rlo_scl} AND {term}"
)
)
map_key = (network_id, instr_uid, "out")
scl_map[map_key] = new_rlo_scl
instruction["scl"] = f"// RLO: {new_rlo_scl}"
instruction["type"] = instr_type + SCL_SUFFIX
return True
# --- process_edge_detector MODIFICADA ---
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para el procesador Contact."""
return {'type_name': 'contact', 'processor_func': process_contact, 'priority': 1}

View File

@ -0,0 +1,81 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_convert(instruction, network_id, scl_map, access_map, data):
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"]
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
en_input = instruction["inputs"].get("en")
en_scl = (
get_scl_representation(en_input, network_id, scl_map, access_map)
if en_input
else "TRUE"
)
in_info = instruction["inputs"].get("in")
in_scl = get_scl_representation(in_info, network_id, scl_map, access_map)
if en_scl is None or in_scl is None:
return False # Esperar si dependencias no listas
target_scl = get_target_scl_name(
instruction, "out", network_id, default_to_temp=True
)
if target_scl is None:
print(f"Error: Sin destino claro para CONVERT {instr_uid}")
instruction["scl"] = f"// ERROR: Convert {instr_uid} sin destino"
instruction["type"] += "_error"
return True # Procesado con error
# Formatear entrada si es variable
in_scl_formatted = (
format_variable_name(in_scl)
if in_info and in_info.get("type") == "variable"
else in_scl
)
# Determinar el tipo de destino (simplificado, necesitaría info del XML original)
# Asumimos que el tipo está en TemplateValues o inferirlo del nombre/contexto
target_type = instruction.get("template_values", {}).get(
"destType", "VARIANT"
) # Ejemplo, ajustar según XML real
conversion_func = f"{target_type}_TO_" # Necesita el tipo de origen también
# Esta parte es compleja sin saber los tipos exactos. Usaremos una conversión genérica o MOVE.
# Para una conversión real, necesitaríamos algo como:
# conversion_expr = f"CONVERT(IN := {in_scl_formatted}, OUT => {target_scl})" # Sintaxis inventada
# O usar funciones específicas: INT_TO_REAL, etc.
# Simplificación: Usar asignación directa (MOVE implícito) o función genérica si existe
# Asumiremos asignación directa por ahora.
conversion_expr = in_scl_formatted
scl_core = f"{target_scl} := {conversion_expr};"
# Añadir IF EN si es necesario
scl_final = (
f"IF {en_scl} THEN\n {scl_core}\nEND_IF;" if en_scl != "TRUE" else scl_core
)
instruction["scl"] = scl_final
instruction["type"] = instr_type + SCL_SUFFIX
map_key_out = (network_id, instr_uid, "out")
scl_map[map_key_out] = target_scl # El valor de salida es el contenido del destino
map_key_eno = (network_id, instr_uid, "eno")
scl_map[map_key_eno] = en_scl # ENO sigue a EN
return True
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para el procesador Convert."""
return {'type_name': 'convert', 'processor_func': process_convert, 'priority': 4}

View File

@ -0,0 +1,94 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_counter(instruction, network_id, scl_map, access_map, data):
"""
Genera SCL para Contadores (CTU, CTD, CTUD).
Requiere datos de instancia (DB o STAT).
"""
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"] # CTU, CTD, CTUD
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
# 1. Obtener Inputs (varía según tipo)
params = []
resolved = True
input_pins = []
if instr_type == "CTU": input_pins = ["CU", "R", "PV"]
elif instr_type == "CTD": input_pins = ["CD", "LD", "PV"]
elif instr_type == "CTUD": input_pins = ["CU", "CD", "R", "LD", "PV"]
else:
instruction["scl"] = f"// ERROR: Tipo de contador no soportado: {instr_type}"
instruction["type"] += "_error"
return True # Procesado con error
for pin in input_pins:
pin_info = instruction["inputs"].get(pin)
if pin_info is None and pin not in ["R", "LD"]: # R y LD pueden no estar conectados
print(f"Error: Falta entrada requerida '{pin}' para {instr_type} UID {instr_uid}.")
# Permitir continuar si solo faltan R o LD opcionales? Por ahora no.
instruction["scl"] = f"// ERROR: Falta entrada requerida '{pin}' para {instr_type} UID {instr_uid}."
instruction["type"] += "_error"
return True # Error
elif pin_info: # Si el pin existe en el JSON
scl_pin = get_scl_representation(pin_info, network_id, scl_map, access_map)
if scl_pin is None:
resolved = False
break # Salir si una dependencia no está lista
scl_pin_formatted = format_variable_name(scl_pin) if pin_info.get("type") == "variable" else scl_pin
params.append(f"{pin} := {scl_pin_formatted}")
if not resolved: return False
# 2. Obtener Nombre de Instancia (NECESITA MEJORA EN x1.py)
instance_name = instruction.get("instance_db")
if not instance_name:
instance_name = f"#COUNTER_INSTANCE_{instr_uid}" # Placeholder
print(f"Advertencia: No se encontró instancia para {instr_type} UID {instr_uid}. Usando placeholder '{instance_name}'. Ajustar x1.py y declarar en x3.py.")
else:
instance_name = format_variable_name(instance_name)
# 3. Generar la llamada SCL
param_string = ", ".join(params)
scl_call = f"{instance_name}({param_string}); // TODO: Declarar {instance_name} : {instr_type}; en VAR_STAT o VAR"
instruction["scl"] = scl_call
instruction["type"] = instr_type + SCL_SUFFIX
# 4. Actualizar scl_map para las salidas (QU, QD, CV)
output_pins = []
if instr_type == "CTU": output_pins = ["QU", "CV"]
elif instr_type == "CTD": output_pins = ["QD", "CV"]
elif instr_type == "CTUD": output_pins = ["QU", "QD", "CV"]
for pin in output_pins:
map_key = (network_id, instr_uid, pin)
scl_map[map_key] = f"{instance_name}.{pin}"
# Contadores no tienen ENO estándar en LAD/FBD
return True
# --- Procesador de Comparadores (EQ ya existe, añadir otros) ---
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para los contadores CTU, CTD, CTUD."""
# Asumiendo que los tipos en el JSON son CTU, CTD, CTUD
return [
{'type_name': 'ctu', 'processor_func': process_counter, 'priority': 5},
{'type_name': 'ctd', 'processor_func': process_counter, 'priority': 5},
{'type_name': 'ctud', 'processor_func': process_counter, 'priority': 5}
]

View File

@ -0,0 +1,91 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_edge_detector(instruction, network_id, scl_map, access_map, data):
"""Genera SCL para PBox (P_TRIG) o NBox (N_TRIG).
Guarda la expresión del pulso en scl_map['out'] y la actualización
de memoria en un campo temporal '_edge_mem_update_scl'.
El campo 'scl' principal se deja casi vacío/comentario.
Usa el nombre de memoria original sin renombrar.
"""
instr_uid = instruction["instruction_uid"]
instr_type_original = instruction["type"] # PBox o NBox
if instr_type_original.endswith(SCL_SUFFIX) or "_error" in instr_type_original:
return False # Ya procesado o error
# 1. Obtener CLK y MemBit original
clk_input = instruction["inputs"].get("in")
mem_bit_input = instruction["inputs"].get("bit")
clk_scl = get_scl_representation(clk_input, network_id, scl_map, access_map)
mem_bit_scl_original = get_scl_representation(mem_bit_input, network_id, scl_map, access_map) # Ej: "M19001"
# 2. Verificar dependencias y tipo de MemBit
if clk_scl is None: return False
if mem_bit_scl_original is None:
instruction["scl"] = f"// ERROR: {instr_type_original} {instr_uid} MemBit no resuelto."
instruction["type"] = instr_type_original + "_error"
return True
if not (mem_bit_input and mem_bit_input.get("type") == "variable"):
instruction["scl"] = f"// ERROR: {instr_type_original} {instr_uid} 'bit' no es variable."
instruction["type"] = instr_type_original + "_error"
return True
# 3. Formatear CLK (usa memoria original)
clk_scl_formatted = clk_scl
if clk_scl not in ["TRUE", "FALSE"] and \
(' ' in clk_scl or 'AND' in clk_scl or 'OR' in clk_scl or ':=' in clk_scl) and \
not (clk_scl.startswith('(') and clk_scl.endswith(')')):
clk_scl_formatted = f"({clk_scl})"
# 4. Generar Lógica SCL del *pulso*
result_pulse_expression = "FALSE"
scl_comment = ""
if instr_type_original == "PBox": # P_TRIG
result_pulse_expression = f"{clk_scl_formatted} AND NOT {mem_bit_scl_original}"
scl_comment = f"// P_TRIG({clk_scl_formatted})"
elif instr_type_original == "NBox": # N_TRIG
result_pulse_expression = f"NOT {clk_scl_formatted} AND {mem_bit_scl_original}"
scl_comment = f"// N_TRIG({clk_scl_formatted})"
else: # Error
instruction["scl"] = f"// ERROR: Tipo de flanco inesperado {instr_type_original}"
instruction["type"] = instr_type_original + "_error"
return True
# 5. Generar la actualización del bit de memoria
scl_mem_update = f"{mem_bit_scl_original} := {clk_scl_formatted};"
# 6. Almacenar Resultados
map_key_out = (network_id, instr_uid, "out")
scl_map[map_key_out] = result_pulse_expression # Guardar EXPRESIÓN del pulso
instruction['_edge_mem_update_scl'] = f"{scl_mem_update} {scl_comment}" # Guardar UPDATE + Comentario en campo temporal
instruction['scl'] = f"// {instr_type_original} Logic moved to consumer Coil" # Dejar SCL principal vacío/comentario
instruction["type"] = instr_type_original + SCL_SUFFIX
# 7. Propagar ENO
map_key_eno = (network_id, instr_uid, "eno")
scl_map[map_key_eno] = clk_scl
return True
# --- process_coil MODIFICADA (con \n correcto) ---
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para los detectores de flanco PBox (P_TRIG) y NBox (N_TRIG)."""
return [
{'type_name': 'pbox', 'processor_func': process_edge_detector, 'priority': 2}, # Flanco positivo
{'type_name': 'nbox', 'processor_func': process_edge_detector, 'priority': 2} # Flanco negativo
]

74
processors/process_eq.py Normal file
View File

@ -0,0 +1,74 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_eq(instruction, network_id, scl_map, access_map, data):
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"]
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
in1_info = instruction["inputs"].get("in1")
in2_info = instruction["inputs"].get("in2")
in1_scl = get_scl_representation(in1_info, network_id, scl_map, access_map)
in2_scl = get_scl_representation(in2_info, network_id, scl_map, access_map)
if in1_scl is None or in2_scl is None:
return False # Dependencias no listas
# Formatear operandos si son variables
op1 = (
format_variable_name(in1_scl)
if in1_info and in1_info.get("type") == "variable"
else in1_scl
)
op2 = (
format_variable_name(in2_scl)
if in2_info and in2_info.get("type") == "variable"
else in2_scl
)
# Añadir paréntesis si los operandos contienen espacios (poco probable después de formatear)
op1 = f"({op1})" if " " in op1 and not op1.startswith("(") else op1
op2 = f"({op2})" if " " in op2 and not op2.startswith("(") else op2
comparison_scl = f"{op1} = {op2}"
# Guardar el resultado booleano de la comparación en el mapa SCL para la salida 'out'
map_key_out = (network_id, instr_uid, "out")
scl_map[map_key_out] = comparison_scl
# Procesar la entrada 'pre' (RLO anterior) para determinar ENO
pre_input = instruction["inputs"].get("pre")
pre_scl = (
"TRUE"
if pre_input is None
else get_scl_representation(pre_input, network_id, scl_map, access_map)
)
if pre_scl is None:
return False # Dependencia 'pre' no lista
# Guardar el estado de 'pre' como ENO
map_key_eno = (network_id, instr_uid, "eno")
scl_map[map_key_eno] = pre_scl
# El SCL generado es solo un comentario indicando la condición,
# ya que la lógica se propaga a través de scl_map['out']
instruction["scl"] = f"// Comparison Eq {instr_uid}: {comparison_scl}"
instruction["type"] = instr_type + SCL_SUFFIX
return True
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para el comparador de igualdad (EQ)."""
return {'type_name': 'eq', 'processor_func': process_eq, 'priority': 2}

View File

@ -0,0 +1,82 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_math(instruction, network_id, scl_map, access_map, data):
"""
Genera SCL para operaciones matemáticas (SUB, MUL, DIV).
"""
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"] # SUB, MUL, DIV
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
# Mapa de tipos a operadores SCL
op_map = {"SUB": "-", "MUL": "*", "DIV": "/"}
scl_operator = op_map.get(instr_type)
if not scl_operator:
instruction["scl"] = f"// ERROR: Operación matemática no soportada: {instr_type}"
instruction["type"] += "_error"
return True
# Obtener EN, IN1, IN2
en_input = instruction["inputs"].get("en")
in1_info = instruction["inputs"].get("in1")
in2_info = instruction["inputs"].get("in2")
en_scl = get_scl_representation(en_input, network_id, scl_map, access_map) if en_input else "TRUE"
in1_scl = get_scl_representation(in1_info, network_id, scl_map, access_map)
in2_scl = get_scl_representation(in2_info, network_id, scl_map, access_map)
if en_scl is None or in1_scl is None or in2_scl is None:
return False # Dependencias no listas
# Obtener destino 'out'
target_scl = get_target_scl_name(instruction, "out", network_id, default_to_temp=True)
if target_scl is None:
instruction["scl"] = f"// ERROR: {instr_type} {instr_uid} sin destino 'out'."
instruction["type"] += "_error"
return True
# Formatear operandos si son variables
op1 = format_variable_name(in1_scl) if in1_info and in1_info.get("type") == "variable" else in1_scl
op2 = format_variable_name(in2_scl) if in2_info and in2_info.get("type") == "variable" else in2_scl
# Añadir paréntesis si es necesario (especialmente para expresiones)
op1 = f"({op1})" if (" " in op1 or "+" in op1 or "-" in op1 or "*" in op1 or "/" in op1) and not op1.startswith("(") else op1
op2 = f"({op2})" if (" " in op2 or "+" in op2 or "-" in op2 or "*" in op2 or "/" in op2) and not op2.startswith("(") else op2
# Generar SCL
scl_core = f"{target_scl} := {op1} {scl_operator} {op2};"
scl_final = f"IF {en_scl} THEN\n {scl_core}\nEND_IF;" if en_scl != "TRUE" else scl_core
instruction["scl"] = scl_final
instruction["type"] = instr_type + SCL_SUFFIX
# Actualizar mapa SCL
map_key_out = (network_id, instr_uid, "out")
scl_map[map_key_out] = target_scl
map_key_eno = (network_id, instr_uid, "eno")
scl_map[map_key_eno] = en_scl
return True
# --- Procesador NOT ---
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para operaciones matemáticas (excepto ADD, MOD)."""
# Esta función maneja SUB, MUL, DIV.
return [
{'type_name': 'sub', 'processor_func': process_math, 'priority': 4}, # Resta
{'type_name': 'mul', 'processor_func': process_math, 'priority': 4}, # Multiplicación
{'type_name': 'div', 'processor_func': process_math, 'priority': 4} # División
]

76
processors/process_mod.py Normal file
View File

@ -0,0 +1,76 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_mod(instruction, network_id, scl_map, access_map, data):
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"]
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
en_input = instruction["inputs"].get("en")
en_scl = (
get_scl_representation(en_input, network_id, scl_map, access_map)
if en_input
else "TRUE"
)
in1_info = instruction["inputs"].get("in1")
in2_info = instruction["inputs"].get("in2")
in1_scl = get_scl_representation(in1_info, network_id, scl_map, access_map)
in2_scl = get_scl_representation(in2_info, network_id, scl_map, access_map)
if en_scl is None or in1_scl is None or in2_scl is None:
return False
target_scl = get_target_scl_name(
instruction, "out", network_id, default_to_temp=True
)
if target_scl is None:
print(f"Error: Sin destino MOD {instr_uid}")
instruction["scl"] = f"// ERROR: Mod {instr_uid} sin destino"
instruction["type"] += "_error"
return True
# Formatear operandos si son variables
op1 = (
format_variable_name(in1_scl)
if in1_info and in1_info.get("type") == "variable"
else in1_scl
)
op2 = (
format_variable_name(in2_scl)
if in2_info and in2_info.get("type") == "variable"
else in2_scl
)
# Añadir paréntesis si es necesario (poco probable tras formatear)
op1 = f"({op1})" if " " in op1 and not op1.startswith("(") else op1
op2 = f"({op2})" if " " in op2 and not op2.startswith("(") else op2
scl_core = f"{target_scl} := {op1} MOD {op2};"
scl_final = (
f"IF {en_scl} THEN\n {scl_core}\nEND_IF;" if en_scl != "TRUE" else scl_core
)
instruction["scl"] = scl_final
instruction["type"] = instr_type + SCL_SUFFIX
map_key_out = (network_id, instr_uid, "out")
scl_map[map_key_out] = target_scl
map_key_eno = (network_id, instr_uid, "eno")
scl_map[map_key_eno] = en_scl
return True
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para la operación Modulo."""
return {'type_name': 'mod', 'processor_func': process_mod, 'priority': 4}

View File

@ -0,0 +1,84 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_move(instruction, network_id, scl_map, access_map, data):
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"]
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
en_input = instruction["inputs"].get("en")
en_scl = (
get_scl_representation(en_input, network_id, scl_map, access_map)
if en_input
else "TRUE"
)
in_info = instruction["inputs"].get("in")
in_scl = get_scl_representation(in_info, network_id, scl_map, access_map)
if en_scl is None or in_scl is None:
return False
# Intentar obtener el destino de 'out1' (o 'out' si es el estándar)
target_scl = get_target_scl_name(
instruction, "out1", network_id, default_to_temp=False
)
if target_scl is None:
target_scl = get_target_scl_name(
instruction, "out", network_id, default_to_temp=False
)
if target_scl is None:
# Si no hay destino explícito, podríamos usar una temp si la lógica lo requiere,
# pero MOVE generalmente necesita un destino claro. Marcar como advertencia/error.
print(
f"Advertencia/Error: MOVE {instr_uid} sin destino claro en 'out' o 'out1'. Se requiere destino explícito."
)
# Decidir si generar error o simplemente no hacer nada. No hacer nada es más seguro.
# instruction["scl"] = f"// ERROR: MOVE {instr_uid} sin destino"
# instruction["type"] += "_error"
# return True
return False # No procesar si no hay destino claro
# Formatear entrada si es variable
in_scl_formatted = (
format_variable_name(in_scl)
if in_info and in_info.get("type") == "variable"
else in_scl
)
scl_core = f"{target_scl} := {in_scl_formatted};"
scl_final = (
f"IF {en_scl} THEN\n {scl_core}\nEND_IF;" if en_scl != "TRUE" else scl_core
)
instruction["scl"] = scl_final
instruction["type"] = instr_type + SCL_SUFFIX
# Propagar el valor movido a través de scl_map si se usa 'out' o 'out1' como fuente
map_key_out = (network_id, instr_uid, "out") # Asumir 'out' como estándar
scl_map[map_key_out] = target_scl # El valor es lo que está en el destino
map_key_out1 = (network_id, instr_uid, "out1") # Si existe out1
scl_map[map_key_out1] = target_scl
map_key_eno = (network_id, instr_uid, "eno")
scl_map[map_key_eno] = en_scl
return True
# EN x2_process.py
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para la operación Move."""
return {'type_name': 'move', 'processor_func': process_move, 'priority': 3}

47
processors/process_not.py Normal file
View File

@ -0,0 +1,47 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_not(instruction, network_id, scl_map, access_map, data):
"""Genera la expresión SCL para la inversión lógica NOT."""
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"] # Not
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
in_info = instruction["inputs"].get("in")
in_scl = get_scl_representation(in_info, network_id, scl_map, access_map)
if in_scl is None:
return False # Dependencia no lista
# Formatear entrada (añadir paréntesis si es complejo)
in_scl_formatted = in_scl
if (" " in in_scl or "AND" in in_scl or "OR" in in_scl) and not (in_scl.startswith("(") and in_scl.endswith(")")):
in_scl_formatted = f"({in_scl})"
result_scl = f"NOT {in_scl_formatted}"
# Guardar resultado en mapa para 'out'
map_key_out = (network_id, instr_uid, "out")
scl_map[map_key_out] = result_scl
instruction["scl"] = f"// Logic NOT {instr_uid}: {result_scl}"
instruction["type"] = instr_type + SCL_SUFFIX
# NOT no tiene EN/ENO explícito en LAD, modifica el RLO
return True
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para la operación Not."""
return {'type_name': 'not', 'processor_func': process_not, 'priority': 1}

80
processors/process_o.py Normal file
View File

@ -0,0 +1,80 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_o(instruction, network_id, scl_map, access_map, data):
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"]
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
# Buscar todas las entradas 'in', 'in1', 'in2', ...
input_pins = sorted([pin for pin in instruction["inputs"] if pin.startswith("in")])
if not input_pins:
print(f"Error: O {instr_uid} sin pines de entrada (inX).")
instruction["scl"] = f"// ERROR: O {instr_uid} sin pines inX"
instruction["type"] += "_error"
return True # Procesado con error
scl_parts = []
all_resolved = True
for pin in input_pins:
in_scl = get_scl_representation(
instruction["inputs"][pin], network_id, scl_map, access_map
)
if in_scl is None:
all_resolved = False
# print(f"DEBUG: O {instr_uid} esperando pin {pin}")
break # Salir del bucle for si una entrada no está lista
# Formatear término (añadir paréntesis si es necesario)
term = in_scl
if (" " in term or "AND" in term) and not (
term.startswith("(") and term.endswith(")")
):
term = f"({term})"
scl_parts.append(term)
if not all_resolved:
return False # Esperar a que todas las entradas estén resueltas
# Construir la expresión OR
result_scl = "FALSE" # Valor por defecto si no hay entradas válidas (raro)
if scl_parts:
result_scl = " OR ".join(scl_parts)
# Simplificar si solo hay un término
if len(scl_parts) == 1:
result_scl = scl_parts[0]
# Quitar paréntesis redundantes si solo hay un término y está entre paréntesis
if result_scl.startswith("(") and result_scl.endswith(")"):
# Comprobar si los paréntesis son necesarios (contienen operadores de menor precedencia)
# Simplificación: quitar siempre si solo hay un término. Podría ser incorrecto en casos complejos.
# result_scl = result_scl[1:-1] # Comentado por seguridad
pass
# Actualizar mapa SCL y la instrucción
map_key_out = (network_id, instr_uid, "out")
scl_map[map_key_out] = result_scl
instruction["scl"] = (
f"// Logic O {instr_uid}: {result_scl}" # Comentario informativo
)
instruction["type"] = instr_type + SCL_SUFFIX
# La instrucción 'O' no tiene ENO propio, propaga el resultado por 'out'
return True
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para la operación lógica O (OR)."""
return {'type_name': 'o', 'processor_func': process_o, 'priority': 1}

View File

@ -0,0 +1,59 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_rcoil(instruction, network_id, scl_map, access_map, data ):
"""Genera SCL para Reset Coil (RCoil): IF condition THEN variable := FALSE; END_IF;"""
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"]
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False # Ya procesado o con error
# Obtener condición de entrada (RLO)
in_info = instruction["inputs"].get("in")
condition_scl = get_scl_representation(in_info, network_id, scl_map, access_map)
# Obtener operando (variable a poner a FALSE)
operand_info = instruction["inputs"].get("operand")
variable_scl = get_scl_representation(operand_info, network_id, scl_map, access_map)
# Verificar dependencias
if condition_scl is None or variable_scl is None:
return False # Dependencias no listas
# Verificar que el operando sea una variable
if not (operand_info and operand_info.get("type") == "variable"):
print(f"Error: RCoil {instr_uid} operando no es variable o falta info (Tipo: {operand_info.get('type')}).")
instruction["scl"] = f"// ERROR: RCoil {instr_uid} operando no es variable."
instruction["type"] += "_error"
return True # Procesado con error
# Formatear nombre de variable
variable_name_formatted = format_variable_name(variable_scl)
# Generar SCL
scl_core = f"{variable_name_formatted} := FALSE;"
scl_final = (
f"IF {condition_scl} THEN\n {scl_core}\nEND_IF;" if condition_scl != "TRUE" else scl_core
)
# Actualizar instrucción
instruction["scl"] = scl_final
instruction["type"] = instr_type + SCL_SUFFIX
# RCoil no genera salida 'out' ni 'eno' significativas para propagar
return True
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para la bobina Reset (RCoil)."""
return {'type_name': 'rcoil', 'processor_func': process_rcoil, 'priority': 3}

View File

@ -0,0 +1,59 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_scoil(instruction, network_id, scl_map, access_map, data):
"""Genera SCL para Set Coil (SCoil): IF condition THEN variable := TRUE; END_IF;"""
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"]
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False # Ya procesado o con error
# Obtener condición de entrada (RLO)
in_info = instruction["inputs"].get("in")
condition_scl = get_scl_representation(in_info, network_id, scl_map, access_map)
# Obtener operando (variable a poner a TRUE)
operand_info = instruction["inputs"].get("operand")
variable_scl = get_scl_representation(operand_info, network_id, scl_map, access_map)
# Verificar dependencias
if condition_scl is None or variable_scl is None:
return False # Dependencias no listas
# Verificar que el operando sea una variable
if not (operand_info and operand_info.get("type") == "variable"):
print(f"Error: SCoil {instr_uid} operando no es variable o falta info (Tipo: {operand_info.get('type')}).")
instruction["scl"] = f"// ERROR: SCoil {instr_uid} operando no es variable."
instruction["type"] += "_error"
return True # Procesado con error
# Formatear nombre de variable
variable_name_formatted = format_variable_name(variable_scl)
# Generar SCL
scl_core = f"{variable_name_formatted} := TRUE;"
scl_final = (
f"IF {condition_scl} THEN\n {scl_core}\nEND_IF;" if condition_scl != "TRUE" else scl_core
)
# Actualizar instrucción
instruction["scl"] = scl_final
instruction["type"] = instr_type + SCL_SUFFIX
# SCoil no genera salida 'out' ni 'eno' significativas para propagar
return True
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para la bobina Set (SCoil)."""
return {'type_name': 'scoil', 'processor_func': process_scoil, 'priority': 3}

73
processors/process_sd.py Normal file
View File

@ -0,0 +1,73 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_sd(instruction, network_id, scl_map, access_map, data):
"""
Genera SCL para Temporizador On-Delay (Sd -> TON).
Requiere datos de instancia (DB o STAT/TEMP).
"""
instr_uid = instruction["instruction_uid"]
instr_type = "Sd" # Tipo original LAD
if instruction["type"].endswith(SCL_SUFFIX) or "_error" in instruction["type"]:
return False
# 1. Obtener Inputs: s (start), tv (time value)
# El pin 'r' (reset) no tiene equivalente directo en TON, se ignora aquí.
s_info = instruction["inputs"].get("s")
tv_info = instruction["inputs"].get("tv")
timer_instance_info = instruction["inputs"].get("timer") # Esperando que x1 lo extraiga
scl_s = get_scl_representation(s_info, network_id, scl_map, access_map)
scl_tv = get_scl_representation(tv_info, network_id, scl_map, access_map)
scl_instance_name = get_scl_representation(timer_instance_info, network_id, scl_map, access_map)
if scl_s is None or scl_tv is None:
return False # Dependencias no listas
# 2. Validar y obtener Nombre de Instancia
instance_name = None
if timer_instance_info and timer_instance_info.get("type") == "variable":
instance_name = scl_instance_name
elif timer_instance_info:
print(f"Advertencia: Pin 'timer' de {instr_type} UID {instr_uid} conectado a algo inesperado: {timer_instance_info.get('type')}")
instance_name = f"#TON_INSTANCE_{instr_uid}"
else:
instance_name = f"#TON_INSTANCE_{instr_uid}"
print(f"Advertencia: No se encontró conexión al pin 'timer' para {instr_type} UID {instr_uid}. Usando placeholder '{instance_name}'. ¡Revisar x1.py y XML!")
# 3. Formatear entradas si son variables
scl_s_formatted = format_variable_name(scl_s) if s_info and s_info.get("type") == "variable" else scl_s
scl_tv_formatted = format_variable_name(scl_tv) if tv_info and tv_info.get("type") == "variable" else scl_tv
# 4. Generar la llamada SCL (TON usa IN, PT, Q, ET)
scl_call = f"{instance_name}(IN := {scl_s_formatted}, PT := {scl_tv_formatted}); // TODO: Declarar {instance_name} : TON; en VAR_STAT o VAR"
instruction["scl"] = scl_call
instruction["type"] = instr_type + SCL_SUFFIX
# 5. Actualizar scl_map para las salidas Q y RT (mapeado a ET de TON)
map_key_q = (network_id, instr_uid, "q")
scl_map[map_key_q] = f"{instance_name}.Q"
map_key_rt = (network_id, instr_uid, "rt")
scl_map[map_key_rt] = f"{instance_name}.ET"
return True
# --- NUEVO: Procesador de Agrupación (Refinado) ---
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para el temporizador On-Delay (Sd -> TON)."""
# Asumiendo que el tipo en el JSON es 'Sd'
return {'type_name': 'sd', 'processor_func': process_sd, 'priority': 5}

91
processors/process_se.py Normal file
View File

@ -0,0 +1,91 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_se(instruction, network_id, scl_map, access_map, data):
"""
Genera SCL para Temporizador de Pulso (Se -> TP).
Requiere datos de instancia (DB o STAT/TEMP).
"""
instr_uid = instruction["instruction_uid"]
instr_type = "Se" # Tipo original LAD
if instruction["type"].endswith(SCL_SUFFIX) or "_error" in instruction["type"]:
return False
# 1. Obtener Inputs: s (start), tv (time value)
# El pin 'r' (reset) no tiene equivalente directo en TP, se ignora aquí.
s_info = instruction["inputs"].get("s")
tv_info = instruction["inputs"].get("tv")
timer_instance_info = instruction["inputs"].get("timer") # Esperando que x1 lo extraiga
scl_s = get_scl_representation(s_info, network_id, scl_map, access_map)
scl_tv = get_scl_representation(tv_info, network_id, scl_map, access_map)
# Obtenemos el nombre de la variable instancia, crucial!
scl_instance_name = get_scl_representation(timer_instance_info, network_id, scl_map, access_map)
if scl_s is None or scl_tv is None:
return False # Dependencias no listas
# 2. Validar y obtener Nombre de Instancia
instance_name = None
if timer_instance_info and timer_instance_info.get("type") == "variable":
instance_name = scl_instance_name # Ya debería estar formateado por get_scl_repr
elif timer_instance_info: # Si está conectado pero no es variable directa? Raro.
print(f"Advertencia: Pin 'timer' de {instr_type} UID {instr_uid} conectado a algo inesperado: {timer_instance_info.get('type')}")
instance_name = f"#TP_INSTANCE_{instr_uid}" # Usar placeholder
else: # Si no hay pin 'timer' conectado (no debería pasar si x1 funciona)
instance_name = f"#TP_INSTANCE_{instr_uid}" # Usar placeholder
print(f"Advertencia: No se encontró conexión al pin 'timer' para {instr_type} UID {instr_uid}. Usando placeholder '{instance_name}'. ¡Revisar x1.py y XML!")
# 3. Formatear entradas si son variables (aunque get_scl_representation ya debería hacerlo)
scl_s_formatted = format_variable_name(scl_s) if s_info and s_info.get("type") == "variable" else scl_s
scl_tv_formatted = format_variable_name(scl_tv) if tv_info and tv_info.get("type") == "variable" else scl_tv
# 4. Generar la llamada SCL (TP usa IN, PT, Q, ET)
scl_call = f"{instance_name}(IN := {scl_s_formatted}, PT := {scl_tv_formatted}); // TODO: Declarar {instance_name} : TP; en VAR_STAT o VAR"
instruction["scl"] = scl_call
instruction["type"] = instr_type + SCL_SUFFIX
# 5. Actualizar scl_map usando los nombres de pin ORIGINALES mapeados si existen
output_pin_mapping_reverse = {v: k for k, v in instruction.get("_output_pin_mapping", {}).items()} # Necesitaríamos guardar el mapeo en x1
q_original_pin = "q" # Default
rt_original_pin = "rt" # Default
# Intentar encontrar los pines originales si x1 guardó el mapeo (MEJORA NECESARIA en x1)
# Por ahora, para SdCoil que mapeaba out->q, usaremos 'out' directamente
if instruction.get("type") == "Se_scl" and instruction.get("original_type") == "SdCoil": # Necesitamos guardar original_type en x1
q_original_pin = "out"
# rt no existe en SdCoil
map_key_q = (network_id, instr_uid, q_original_pin)
scl_map[map_key_q] = f"{instance_name}.Q"
if rt_original_pin: # Solo añadir rt si corresponde
map_key_rt = (network_id, instr_uid, rt_original_pin)
scl_map[map_key_rt] = f"{instance_name}.ET"
return True
# --- Procesador para Sd (On-Delay Timer -> TON SCL) ---
# --- Function code ends ---
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para el temporizador de Pulso (Se -> TP) y maneja SdCoil."""
# Asumiendo que el tipo en el JSON es 'Se'
# Y que SdCoil también se mapea aquí según el análisis previo
return [
{'type_name': 'se', 'processor_func': process_se, 'priority': 5},
{'type_name': 'sdcoil', 'processor_func': process_se, 'priority': 5} # SdCoil también se procesa como TP
]

View File

@ -0,0 +1,70 @@
# -*- coding: utf-8 -*-
from .processor_utils import get_scl_representation, format_variable_name,get_target_scl_name
# TODO: Import necessary functions from processor_utils
# Example: from .processor_utils import get_scl_representation, format_variable_name
# Or: import processors.processor_utils as utils
# TODO: Define constants if needed (e.g., SCL_SUFFIX) or import them
SCL_SUFFIX = "_scl"
# --- Function code starts ---
def process_timer(instruction, network_id, scl_map, access_map, data):
"""
Genera SCL para Temporizadores (TON, TOF).
Requiere datos de instancia (DB o STAT).
"""
instr_uid = instruction["instruction_uid"]
instr_type = instruction["type"] # Será "TON" o "TOF"
if instr_type.endswith(SCL_SUFFIX) or "_error" in instr_type:
return False
# 1. Obtener Inputs
in_info = instruction["inputs"].get("IN") # Entrada booleana
pt_info = instruction["inputs"].get("PT") # Preset Time (Tipo TIME)
scl_in = get_scl_representation(in_info, network_id, scl_map, access_map)
scl_pt = get_scl_representation(pt_info, network_id, scl_map, access_map)
if scl_in is None or scl_pt is None:
return False # Dependencias no listas
# 2. Obtener Nombre de Instancia (NECESITA MEJORA EN x1.py)
instance_name = instruction.get("instance_db") # Reutilizar campo si x1 lo llena
if not instance_name:
# Generar placeholder si x1 no extrajo la instancia para Part
instance_name = f"#TIMER_INSTANCE_{instr_uid}" # Placeholder para VAR_TEMP o VAR_STAT
print(f"Advertencia: No se encontró instancia para {instr_type} UID {instr_uid}. Usando placeholder '{instance_name}'. Ajustar x1.py y declarar en x3.py.")
else:
instance_name = format_variable_name(instance_name) # Limpiar si viene de x1
# 3. Formatear entradas si son variables
scl_in_formatted = format_variable_name(scl_in) if in_info and in_info.get("type") == "variable" else scl_in
scl_pt_formatted = format_variable_name(scl_pt) if pt_info and pt_info.get("type") == "variable" else scl_pt
# 4. Generar la llamada SCL
# Nota: Las salidas Q y ET se acceden directamente desde la instancia.
scl_call = f"{instance_name}(IN := {scl_in_formatted}, PT := {scl_pt_formatted}); // TODO: Declarar {instance_name} : {instr_type}; en VAR_STAT o VAR"
instruction["scl"] = scl_call
instruction["type"] = instr_type + SCL_SUFFIX
# 5. Actualizar scl_map para las salidas Q y ET
map_key_q = (network_id, instr_uid, "Q")
scl_map[map_key_q] = f"{instance_name}.Q"
map_key_et = (network_id, instr_uid, "ET")
scl_map[map_key_et] = f"{instance_name}.ET"
# TON/TOF no tienen un pin ENO estándar en LAD/FBD que se mapee directamente
return True
# --- Processor Information Function ---
def get_processor_info():
"""Devuelve la información para los temporizadores TON y TOF (si se usan genéricamente)."""
# Esta función manejaría tipos TON/TOF si aparecieran directamente en el JSON
# y no fueran manejados por process_sd/process_se (que es lo más común desde LAD).
# Incluir por si acaso o si la conversión inicial genera TON/TOF directamente.
return [
{'type_name': 'ton', 'processor_func': process_timer, 'priority': 5},
{'type_name': 'tof', 'processor_func': process_timer, 'priority': 5}
]

File diff suppressed because it is too large Load Diff