350 lines
15 KiB
Python
350 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
SIO Module Boilerplate Model
|
||
============================
|
||
|
||
Model for SIO (Safety Input/Output) modules with safety-enabled functionality.
|
||
Supports IP address configuration and comment updates for safety I/O channels.
|
||
"""
|
||
|
||
from typing import Dict, Optional, TYPE_CHECKING
|
||
import xml.etree.ElementTree as ET
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
import os
|
||
|
||
if TYPE_CHECKING:
|
||
from excel_data_processor import ModuleData
|
||
|
||
|
||
@dataclass
|
||
class SIOModuleConfig:
|
||
"""Configuration for a SIO module instance."""
|
||
name: str # Module name (e.g., "SIO1")
|
||
ip_address: str = "123.124.125.15"
|
||
parent_module: str = "SLOT2_EN4TR"
|
||
parent_port_id: str = "2"
|
||
inhibited: bool = False
|
||
major_fault: bool = False
|
||
safety_network: str = "16#0000_4c14_03e7_33a8"
|
||
safety_enabled: bool = True
|
||
standard_input_names: Optional[Dict[int, str]] = None # For standard data connection inputs
|
||
standard_output_names: Optional[Dict[int, str]] = None # For standard data connection outputs
|
||
safety_input_names: Optional[Dict[int, str]] = None # For SI connection (DATA[0].0-7)
|
||
safety_output_names: Optional[Dict[int, str]] = None # For SO connection (DATA[1].0-7)
|
||
|
||
|
||
class SIOModuleGenerator:
|
||
"""Generator for SIO module XML with safety I/O support."""
|
||
|
||
def __init__(self, config: SIOModuleConfig):
|
||
self.config = config
|
||
self.boilerplate_path = os.path.join("boilerplate", "SIO_Module.L5X")
|
||
self.tree = None
|
||
self.root = None
|
||
|
||
def load_boilerplate(self):
|
||
"""Load the SIO boilerplate template."""
|
||
if not os.path.exists(self.boilerplate_path):
|
||
raise FileNotFoundError(f"Boilerplate file not found: {self.boilerplate_path}")
|
||
|
||
self.tree = ET.parse(self.boilerplate_path)
|
||
self.root = self.tree.getroot()
|
||
|
||
def update_module_name(self):
|
||
"""Update the module name throughout the XML."""
|
||
# Update in root attributes
|
||
self.root.set("TargetName", self.config.name)
|
||
|
||
# Update Module element
|
||
module = self.root.find(".//Module[@Use='Target']")
|
||
if module is not None:
|
||
module.set("Name", self.config.name)
|
||
|
||
def update_ip_address(self):
|
||
"""Update the IP address in the Ethernet port."""
|
||
port = self.root.find(".//Port[@Type='Ethernet']")
|
||
if port is not None:
|
||
port.set("Address", self.config.ip_address)
|
||
|
||
def update_parent_module(self):
|
||
"""Update parent module references."""
|
||
module = self.root.find(".//Module[@Use='Target']")
|
||
if module is not None:
|
||
module.set("ParentModule", self.config.parent_module)
|
||
module.set("ParentModPortId", self.config.parent_port_id)
|
||
|
||
def update_safety_settings(self):
|
||
"""Update safety-related settings."""
|
||
module = self.root.find(".//Module[@Use='Target']")
|
||
if module is not None:
|
||
module.set("SafetyNetwork", self.config.safety_network)
|
||
module.set("SafetyEnabled", str(self.config.safety_enabled).lower())
|
||
module.set("Inhibited", str(self.config.inhibited).lower())
|
||
module.set("MajorFault", str(self.config.major_fault).lower())
|
||
|
||
def update_comments(self):
|
||
"""Update comments for different connection types."""
|
||
# Update standard data connection input comments (if any)
|
||
if self.config.standard_input_names:
|
||
input_comments = self.root.find(".//Connection[@Name='_200424912C822C83']/InputTag/Comments")
|
||
if input_comments is not None:
|
||
# Clear existing comments
|
||
input_comments.clear()
|
||
# Add new comments for standard inputs
|
||
for index, name in self.config.standard_input_names.items():
|
||
comment = ET.SubElement(input_comments, "Comment")
|
||
comment.set("Operand", f".Data[{index}]")
|
||
comment.text = name
|
||
|
||
# Update standard data connection output comments (if any)
|
||
if self.config.standard_output_names:
|
||
output_comments = self.root.find(".//Connection[@Name='_200424912C822C83']/OutputTag/Comments")
|
||
if output_comments is not None:
|
||
# Clear existing comments
|
||
output_comments.clear()
|
||
# Add new comments for standard outputs
|
||
for index, name in self.config.standard_output_names.items():
|
||
comment = ET.SubElement(output_comments, "Comment")
|
||
comment.set("Operand", f".Data[{index}]")
|
||
comment.text = name
|
||
|
||
# Update safety input comments (SI connection - DATA[0].0 through DATA[0].7)
|
||
if self.config.safety_input_names:
|
||
safety_input_comments = self.root.find(".//Connection[@Name='_200424962CC22C87']/InputTag/Comments")
|
||
if safety_input_comments is not None:
|
||
# Clear existing comments
|
||
safety_input_comments.clear()
|
||
# Add new comments for safety inputs (SI0-SI7)
|
||
for index, name in self.config.safety_input_names.items():
|
||
if 0 <= index <= 7: # SI0-SI7 valid range
|
||
comment = ET.SubElement(safety_input_comments, "Comment")
|
||
comment.set("Operand", f".DATA[0].{index}")
|
||
comment.text = name
|
||
|
||
# Update safety output comments (SO connection - DATA[1].0 through DATA[1].7)
|
||
if self.config.safety_output_names:
|
||
safety_output_comments = self.root.find(".//Connection[@Name='_200424962C862CC2']/OutputTag/Comments")
|
||
if safety_output_comments is not None:
|
||
# Clear existing comments
|
||
safety_output_comments.clear()
|
||
# Add new comments for safety outputs (SO0-SO7, but SO5-SO7 might not be used)
|
||
for index, name in self.config.safety_output_names.items():
|
||
if 0 <= index <= 7: # SO0-SO7 valid range
|
||
comment = ET.SubElement(safety_output_comments, "Comment")
|
||
comment.set("Operand", f".DATA[1].{index}")
|
||
comment.text = name
|
||
|
||
def update_export_date(self):
|
||
"""Update the export date to current time."""
|
||
export_date = datetime.now().strftime("%a %b %d %H:%M:%S %Y")
|
||
self.root.set("ExportDate", export_date)
|
||
|
||
def apply_updates(self):
|
||
"""Apply all updates to the boilerplate."""
|
||
self.update_module_name()
|
||
self.update_ip_address()
|
||
self.update_parent_module()
|
||
self.update_safety_settings()
|
||
self.update_comments()
|
||
self.update_export_date()
|
||
|
||
def save(self, output_path: str):
|
||
"""Save the updated module to file."""
|
||
if self.tree is None:
|
||
raise RuntimeError("No boilerplate loaded. Call load_boilerplate() first.")
|
||
|
||
# Save with proper formatting
|
||
self.tree.write(output_path, encoding='UTF-8', xml_declaration=True)
|
||
|
||
def get_xml_string(self) -> str:
|
||
"""Get the XML as a string."""
|
||
if self.tree is None:
|
||
raise RuntimeError("No boilerplate loaded. Call load_boilerplate() first.")
|
||
|
||
return ET.tostring(self.root, encoding='unicode')
|
||
|
||
# ------------------------------------------------------------------
|
||
# Convenience helpers for EnhancedMCMGenerator refactor
|
||
# ------------------------------------------------------------------
|
||
|
||
@staticmethod
|
||
def _extract_comment_dictionaries(module_data: 'ModuleData') -> tuple[
|
||
Dict[int, str], Dict[int, str], Dict[int, str], Dict[int, str]
|
||
]:
|
||
"""Translate the raw Excel `ModuleData` into the comment dictionaries
|
||
expected by SIOModuleGenerator.update_comments().
|
||
|
||
The logic follows SIO channel constraints:
|
||
- SI0–SI7: Safety Input channels
|
||
- SO0–SO4: Safety Output channels
|
||
- IO0–IO3: IO-Link channels
|
||
"""
|
||
|
||
standard_input_names: Dict[int, str] = {}
|
||
standard_output_names: Dict[int, str] = {}
|
||
safety_input_names: Dict[int, str] = {}
|
||
safety_output_names: Dict[int, str] = {}
|
||
|
||
for io_mapping in module_data.io_mappings:
|
||
if not io_mapping.description:
|
||
continue
|
||
|
||
# Handle SPARE entries explicitly
|
||
comment = (
|
||
"SPARE" if io_mapping.description.upper() == "SPARE" else io_mapping.description
|
||
)
|
||
|
||
io_path = io_mapping.io_path
|
||
if not io_path or ":" not in io_path:
|
||
# Skip malformed IO_PATH strings
|
||
continue
|
||
|
||
path_parts = io_path.split(":", 1)[1]
|
||
if "." not in path_parts:
|
||
continue
|
||
|
||
channel, terminal = path_parts.split(".", 1)
|
||
channel_upper = channel.upper()
|
||
|
||
# Parse terminal to get channel number
|
||
terminal_upper = terminal.upper()
|
||
|
||
if channel_upper == "SI":
|
||
# Safety Input: SI0-SI7
|
||
if terminal_upper.startswith("SI") and len(terminal_upper) >= 3:
|
||
try:
|
||
index = int(terminal_upper[2:]) # Extract number from SI0, SI1, etc.
|
||
if 0 <= index <= 7: # Valid SI range
|
||
safety_input_names[index] = comment
|
||
except ValueError:
|
||
continue
|
||
|
||
elif channel_upper == "SO":
|
||
# Safety Output: SO0-SO4 (per memory constraints)
|
||
if terminal_upper.startswith("SO") and len(terminal_upper) >= 3:
|
||
try:
|
||
index = int(terminal_upper[2:]) # Extract number from SO0, SO1, etc.
|
||
if 0 <= index <= 4: # Valid SO range per constraints
|
||
safety_output_names[index] = comment
|
||
except ValueError:
|
||
continue
|
||
|
||
elif channel_upper == "IO":
|
||
# IO-Link channels: IO0-IO3 (per memory constraints)
|
||
if terminal_upper.startswith("IO") and len(terminal_upper) >= 3:
|
||
try:
|
||
index = int(terminal_upper[2:]) # Extract number from IO0, IO1, etc.
|
||
if 0 <= index <= 3: # Valid IO-Link range per constraints
|
||
# IO-Link channels could go to standard input/output depending on signal type
|
||
if io_mapping.signal.upper() == "I":
|
||
standard_input_names[index] = comment
|
||
elif io_mapping.signal.upper() == "O":
|
||
standard_output_names[index] = comment
|
||
except ValueError:
|
||
continue
|
||
|
||
# Any other terminal formats are ignored
|
||
|
||
return (
|
||
standard_input_names,
|
||
standard_output_names,
|
||
safety_input_names,
|
||
safety_output_names,
|
||
)
|
||
|
||
@classmethod
|
||
def from_excel(
|
||
cls,
|
||
module_data: 'ModuleData',
|
||
*,
|
||
ip_address: str = "",
|
||
parent_module: str = "SLOT2_EN4TR",
|
||
parent_port_id: str = "2",
|
||
) -> 'SIOModuleGenerator':
|
||
"""Factory that builds a fully-configured generator directly from
|
||
ExcelDataProcessor.ModuleData.
|
||
|
||
It returns an *instance* (already loaded and updated) so callers can
|
||
access .root or save it immediately.
|
||
"""
|
||
|
||
from excel_data_processor import ModuleData # local import to avoid cycle at top level
|
||
|
||
if not isinstance(module_data, ModuleData):
|
||
raise TypeError("module_data must be an Excel ModuleData instance")
|
||
|
||
(
|
||
standard_input_names,
|
||
standard_output_names,
|
||
safety_input_names,
|
||
safety_output_names,
|
||
) = cls._extract_comment_dictionaries(module_data)
|
||
|
||
config = create_sio_module(
|
||
name=module_data.tagname,
|
||
ip_address=ip_address or module_data.ip_address or "123.124.125.15",
|
||
parent_module=parent_module,
|
||
parent_port_id=parent_port_id,
|
||
standard_input_names=standard_input_names if standard_input_names else None,
|
||
standard_output_names=standard_output_names if standard_output_names else None,
|
||
safety_input_names=safety_input_names if safety_input_names else None,
|
||
safety_output_names=safety_output_names if safety_output_names else None,
|
||
)
|
||
|
||
generator = cls(config)
|
||
generator.load_boilerplate()
|
||
generator.apply_updates()
|
||
return generator
|
||
|
||
|
||
def create_sio_module(name: str, ip_address: str = "123.124.125.15",
|
||
parent_module: str = "SLOT2_EN4TR", parent_port_id: str = "2",
|
||
standard_input_names: Optional[Dict[int, str]] = None,
|
||
standard_output_names: Optional[Dict[int, str]] = None,
|
||
safety_input_names: Optional[Dict[int, str]] = None,
|
||
safety_output_names: Optional[Dict[int, str]] = None) -> SIOModuleConfig:
|
||
"""Factory function to create a SIO module configuration."""
|
||
return SIOModuleConfig(
|
||
name=name,
|
||
ip_address=ip_address,
|
||
parent_module=parent_module,
|
||
parent_port_id=parent_port_id,
|
||
standard_input_names=standard_input_names,
|
||
standard_output_names=standard_output_names,
|
||
safety_input_names=safety_input_names,
|
||
safety_output_names=safety_output_names
|
||
)
|
||
|
||
|
||
# Example usage
|
||
if __name__ == "__main__":
|
||
# Example: Create a SIO module with safety I/O
|
||
config = create_sio_module(
|
||
name="SIO1",
|
||
ip_address="123.124.125.15",
|
||
safety_input_names={
|
||
0: "Emergency Stop 1",
|
||
1: "Emergency Stop 2",
|
||
2: "Safety Gate 1",
|
||
3: "Safety Gate 2",
|
||
4: "Light Curtain",
|
||
5: "Safety Mat",
|
||
6: "Reset Button",
|
||
7: "Enable Switch"
|
||
},
|
||
safety_output_names={
|
||
0: "Safety Relay 1",
|
||
1: "Safety Relay 2",
|
||
2: "Warning Light",
|
||
3: "Safety Valve",
|
||
4: "Brake Release"
|
||
}
|
||
)
|
||
|
||
generator = SIOModuleGenerator(config)
|
||
generator.load_boilerplate()
|
||
generator.apply_updates()
|
||
generator.save("generated/SIO1.L5X")
|
||
|
||
print(f"Generated SIO module: {config.name}") |