122 lines
4.9 KiB
Python
122 lines
4.9 KiB
Python
# processors/process_coil.py
|
|
import sympy
|
|
from .processor_utils import (
|
|
get_sympy_representation,
|
|
sympy_expr_to_scl,
|
|
get_target_scl_name,
|
|
format_variable_name,
|
|
)
|
|
from .symbol_manager import SymbolManager, extract_plc_variable_name
|
|
|
|
SCL_SUFFIX = "_sympy_processed"
|
|
|
|
|
|
def process_coil(instruction, network_id, sympy_map, symbol_manager, data):
|
|
"""Genera la asignación SCL para Coil, simplificando la entrada SymPy."""
|
|
instr_uid = instruction["instruction_uid"]
|
|
instr_type_original = instruction.get("type", "Coil")
|
|
|
|
if instr_type_original.endswith(SCL_SUFFIX) or "_error" in instr_type_original:
|
|
return False
|
|
|
|
# Get input expression from SymPy map
|
|
coil_input_info = instruction["inputs"].get("in")
|
|
sympy_expr_in = get_sympy_representation(
|
|
coil_input_info, network_id, sympy_map, symbol_manager
|
|
)
|
|
|
|
# Get target variable SCL name
|
|
target_scl_name = get_target_scl_name(
|
|
instruction, "operand", network_id, default_to_temp=False
|
|
) # Coil must have explicit target
|
|
|
|
# Check dependencies
|
|
if sympy_expr_in is None:
|
|
# print(f"DEBUG Coil {instr_uid}: Input dependency not ready.")
|
|
return False
|
|
if target_scl_name is None:
|
|
print(f"Error: Coil {instr_uid} operando no es variable o falta info.")
|
|
instruction["scl"] = f"// ERROR: Coil {instr_uid} operando no es variable."
|
|
instruction["type"] = instr_type_original + "_error"
|
|
return True # Processed with error
|
|
|
|
# *** Perform Simplification ***
|
|
try:
|
|
# simplified_expr = sympy.simplify_logic(sympy_expr_in, force=False)
|
|
# simplified_expr = sympy_expr_in
|
|
simplified_expr = sympy.logic.boolalg.to_dnf(sympy_expr_in, simplify=True)
|
|
except Exception as e:
|
|
print(f"Error during SymPy simplification for Coil {instr_uid}: {e}")
|
|
simplified_expr = sympy_expr_in # Fallback to original expression
|
|
|
|
# *** Convert simplified expression back to SCL string ***
|
|
condition_scl = sympy_expr_to_scl(simplified_expr, symbol_manager)
|
|
|
|
# Generate the final SCL assignment
|
|
scl_assignment = f"{target_scl_name} := {condition_scl};"
|
|
scl_final = scl_assignment
|
|
|
|
# --- Handle Edge Detector Memory Update (FIXED: Avoid Duplication) ---
|
|
# Check if input comes from PBox/NBox and incorporate memory update logic
|
|
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")
|
|
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:
|
|
# Check for the original type before suffix was added
|
|
orig_source_type = (
|
|
source_instruction.get("type", "")
|
|
.replace(SCL_SUFFIX, "")
|
|
.replace("_error", "")
|
|
)
|
|
if (
|
|
orig_source_type in ["PBox", "NBox"]
|
|
and "_edge_mem_update_scl" in source_instruction
|
|
):
|
|
mem_update_scl_combined = source_instruction.get("_edge_mem_update_scl")
|
|
if mem_update_scl_combined:
|
|
# FIXED: Extract just the memory assignment part (before comment)
|
|
mem_assignment_part = mem_update_scl_combined.split(" //")[
|
|
0
|
|
].strip()
|
|
comment_part = (
|
|
" //" + mem_update_scl_combined.split(" //", 1)[1]
|
|
if " //" in mem_update_scl_combined
|
|
else ""
|
|
)
|
|
|
|
# Add memory update BEFORE the coil assignment (correct P_TRIG order)
|
|
scl_final = f"{mem_assignment_part}{comment_part}\n{scl_assignment}"
|
|
# Mark the source as handled
|
|
source_instruction["scl"] = (
|
|
f"// Edge Logic handled by Coil {instr_uid}"
|
|
)
|
|
|
|
# Update instruction
|
|
instruction["scl"] = scl_final
|
|
instruction["type"] = instr_type_original + SCL_SUFFIX
|
|
|
|
# *** ENHANCED: Store the Coil's input expression in sympy_map for potential reuse ***
|
|
# Some instructions (like subsequent Contacts) might need to read the Coil's logic state
|
|
map_key_out = (network_id, instr_uid, "out")
|
|
sympy_map[map_key_out] = simplified_expr # Store the simplified boolean expression
|
|
|
|
return True
|
|
|
|
|
|
# --- 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}
|