#!/usr/bin/env python
# -*- coding: utf-8 -*-
#This file is part of PLCOpenEditor, a library implementing an IEC 61131-3 editor
#based on the plcopen standard.
#
#Copyright (C) 2007: Edouard TISSERANT and Laurent BESSARD
#
#See COPYING file for copyrights details.
#
#This library is free software; you can redistribute it and/or
#modify it under the terms of the GNU General Public
#License as published by the Free Software Foundation; either
#version 2.1 of the License, or (at your option) any later version.
#
#This library is distributed in the hope that it will be useful,
#but WITHOUT ANY WARRANTY; without even the implied warranty of
#MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
#General Public License for more details.
#
#You should have received a copy of the GNU General Public
#License along with this library; if not, write to the Free Software
#Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
from xmlclass import *
from structures import *
from types import *
import os, re
"""
Dictionary that makes the relation between var names in plcopen and displayed values
"""
VarTypes = {"Local" : "localVars", "Temp" : "tempVars", "Input" : "inputVars",
"Output" : "outputVars", "InOut" : "inOutVars", "External" : "externalVars",
"Global" : "globalVars", "Access" : "accessVars"}
searchResultVarTypes = {
"inputVars": "var_input",
"outputVars": "var_output",
"inOutVars": "var_inout"
}
"""
Define in which order var types must be displayed
"""
VarOrder = ["Local","Temp","Input","Output","InOut","External","Global","Access"]
"""
Define which action qualifier must be associated with a duration
"""
QualifierList = {"N" : False, "R" : False, "S" : False, "L" : True, "D" : True,
"P" : False, "P0" : False, "P1" : False, "SD" : True, "DS" : True, "SL" : True}
FILTER_ADDRESS_MODEL = "(%%[IQM](?:[XBWDL])?)(%s)((?:\.[0-9]+)*)"
def update_address(address, address_model, new_leading):
result = address_model.match(address)
if result is None:
return address
groups = result.groups()
return groups[0] + new_leading + groups[2]
def _init_and_compare(function, v1, v2):
if v1 is None:
return v2
if v2 is not None:
return function(v1, v2)
return v1
"""
Helper class for bounding_box calculation
"""
class rect:
def __init__(self, x=None, y=None, width=None, height=None):
self.x_min = x
self.x_max = None
self.y_min = y
self.y_max = None
if width is not None and x is not None:
self.x_max = x + width
if height is not None and y is not None:
self.y_max = y + height
def update(self, x, y):
self.x_min = _init_and_compare(min, self.x_min, x)
self.x_max = _init_and_compare(max, self.x_max, x)
self.y_min = _init_and_compare(min, self.y_min, y)
self.y_max = _init_and_compare(max, self.y_max, y)
def union(self, rect):
self.x_min = _init_and_compare(min, self.x_min, rect.x_min)
self.x_max = _init_and_compare(max, self.x_max, rect.x_max)
self.y_min = _init_and_compare(min, self.y_min, rect.y_min)
self.y_max = _init_and_compare(max, self.y_max, rect.y_max)
def bounding_box(self):
width = height = None
if self.x_min is not None and self.x_max is not None:
width = self.x_max - self.x_min
if self.y_min is not None and self.y_max is not None:
height = self.y_max - self.y_min
return self.x_min, self.y_min, width, height
def TextLenInRowColumn(text):
if text == "":
return (0, 0)
lines = text.split("\n")
return len(lines) - 1, len(lines[-1])
def TestTextElement(text, criteria):
lines = text.splitlines()
if not criteria["case_sensitive"]:
text = text.upper()
test_result = []
result = criteria["pattern"].search(text)
while result is not None:
start = TextLenInRowColumn(text[:result.start()])
end = TextLenInRowColumn(text[:result.end() - 1])
test_result.append((start, end, "\n".join(lines[start[0]:end[0] + 1])))
result = criteria["pattern"].search(text, result.end())
return test_result
PLCOpenClasses = GenerateClassesFromXSD(os.path.join(os.path.split(__file__)[0], "tc6_xml_v201.xsd"))
ElementNameToClass = {}
cls = PLCOpenClasses.get("formattedText", None)
if cls:
def updateElementName(self, old_name, new_name):
index = self.text.find(old_name)
while index != -1:
if index > 0 and (self.text[index - 1].isalnum() or self.text[index - 1] == "_"):
index = self.text.find(old_name, index + len(old_name))
elif index < len(self.text) - len(old_name) and (self.text[index + len(old_name)].isalnum() or self.text[index + len(old_name)] == "_"):
index = self.text.find(old_name, index + len(old_name))
else:
self.text = self.text[:index] + new_name + self.text[index + len(old_name):]
index = self.text.find(old_name, index + len(new_name))
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
startpos = 0
result = address_model.search(self.text, startpos)
while result is not None:
groups = result.groups()
new_address = groups[0] + new_leading + groups[2]
self.text = self.text[:result.start()] + new_address + self.text[result.end():]
startpos = result.start() + len(new_address)
result = address_model.search(self.text, startpos)
setattr(cls, "updateElementAddress", updateElementAddress)
def Search(self, criteria, parent_infos):
return [(tuple(parent_infos),) + result for result in TestTextElement(self.gettext(), criteria)]
setattr(cls, "Search", Search)
cls = PLCOpenClasses.get("project", None)
if cls:
cls.singleLineAttributes = False
cls.EnumeratedDataTypeValues = {}
cls.CustomDataTypeRange = {}
cls.CustomTypeHierarchy = {}
cls.ElementUsingTree = {}
cls.CustomBlockTypes = []
def setname(self, name):
self.contentHeader.setname(name)
setattr(cls, "setname", setname)
def getname(self):
return self.contentHeader.getname()
setattr(cls, "getname", getname)
def getfileHeader(self):
fileheader = {}
fileheader["companyName"] = self.fileHeader.getcompanyName()
if self.fileHeader.getcompanyURL():
fileheader["companyURL"] = self.fileHeader.getcompanyURL()
fileheader["productName"] = self.fileHeader.getproductName()
fileheader["productVersion"] = self.fileHeader.getproductVersion()
if self.fileHeader.getproductRelease():
fileheader["productRelease"] = self.fileHeader.getproductRelease()
fileheader["creationDateTime"] = self.fileHeader.getcreationDateTime()
if self.fileHeader.getcontentDescription():
fileheader["contentDescription"] = self.fileHeader.getcontentDescription()
return fileheader
setattr(cls, "getfileHeader", getfileHeader)
def setfileHeader(self, fileheader):
self.fileHeader.setcompanyName(fileheader["companyName"])
if fileheader.has_key("companyURL"):
self.fileHeader.setcompanyURL(fileheader["companyURL"])
self.fileHeader.setproductName(fileheader["productName"])
self.fileHeader.setproductVersion(fileheader["productVersion"])
if fileheader.has_key("productRelease"):
self.fileHeader.setproductRelease(fileheader["productRelease"])
self.fileHeader.setcreationDateTime(fileheader["creationDateTime"])
if fileheader.has_key("contentDescription"):
self.fileHeader.setcontentDescription(fileheader["contentDescription"])
setattr(cls, "setfileHeader", setfileHeader)
def getcontentHeader(self):
contentheader = {}
contentheader["projectName"] = self.contentHeader.getname()
if self.contentHeader.getversion():
contentheader["projectVersion"] = self.contentHeader.getversion()
if self.contentHeader.getmodificationDateTime():
contentheader["modificationDateTime"] = self.contentHeader.getmodificationDateTime()
if self.contentHeader.getorganization():
contentheader["organization"] = self.contentHeader.getorganization()
if self.contentHeader.getauthor():
contentheader["authorName"] = self.contentHeader.getauthor()
if self.contentHeader.getlanguage():
contentheader["language"] = self.contentHeader.getlanguage()
contentheader["pageSize"] = self.contentHeader.getpageSize()
contentheader["scaling"] = self.contentHeader.getscaling()
return contentheader
setattr(cls, "getcontentHeader", getcontentHeader)
def setcontentHeader(self, contentheader):
self.contentHeader.setname(contentheader["projectName"])
if contentheader.has_key("projectVersion"):
self.contentHeader.setversion(contentheader["projectVersion"])
if contentheader.has_key("modificationDateTime"):
self.contentHeader.setmodificationDateTime(contentheader["modificationDateTime"])
if contentheader.has_key("organization"):
self.contentHeader.setorganization(contentheader["organization"])
if contentheader.has_key("authorName"):
self.contentHeader.setauthor(contentheader["authorName"])
if contentheader.has_key("language"):
self.contentHeader.setlanguage(contentheader["language"])
self.contentHeader.setpageSize(*contentheader["pageSize"])
self.contentHeader.setscaling(contentheader["scaling"])
setattr(cls, "setcontentHeader", setcontentHeader)
def getdataTypes(self):
return self.types.getdataTypeElements()
setattr(cls, "getdataTypes", getdataTypes)
def getdataType(self, name):
return self.types.getdataTypeElement(name)
setattr(cls, "getdataType", getdataType)
def appenddataType(self, name):
if self.CustomTypeHierarchy.has_key(name):
raise ValueError, "\"%s\" Data Type already exists !!!"%name
self.types.appenddataTypeElement(name)
self.AddCustomDataType(self.getdataType(name))
setattr(cls, "appenddataType", appenddataType)
def insertdataType(self, index, datatype):
self.types.insertdataTypeElement(index, datatype)
self.AddCustomDataType(datatype)
setattr(cls, "insertdataType", insertdataType)
def removedataType(self, name):
self.types.removedataTypeElement(name)
self.RefreshDataTypeHierarchy()
self.RefreshElementUsingTree()
setattr(cls, "removedataType", removedataType)
def getpous(self):
return self.types.getpouElements()
setattr(cls, "getpous", getpous)
def getpou(self, name):
return self.types.getpouElement(name)
setattr(cls, "getpou", getpou)
def appendpou(self, name, pou_type, body_type):
self.types.appendpouElement(name, pou_type, body_type)
self.AddCustomBlockType(self.getpou(name))
setattr(cls, "appendpou", appendpou)
def insertpou(self, index, pou):
self.types.insertpouElement(index, pou)
self.AddCustomBlockType(pou)
setattr(cls, "insertpou", insertpou)
def removepou(self, name):
self.types.removepouElement(name)
self.RefreshCustomBlockTypes()
self.RefreshElementUsingTree()
setattr(cls, "removepou", removepou)
def getconfigurations(self):
configurations = self.instances.configurations.getconfiguration()
if configurations:
return configurations
return []
setattr(cls, "getconfigurations", getconfigurations)
def getconfiguration(self, name):
for configuration in self.instances.configurations.getconfiguration():
if configuration.getname() == name:
return configuration
return None
setattr(cls, "getconfiguration", getconfiguration)
def addconfiguration(self, name):
for configuration in self.instances.configurations.getconfiguration():
if configuration.getname() == name:
raise ValueError, _("\"%s\" configuration already exists !!!")%name
new_configuration = PLCOpenClasses["configurations_configuration"]()
new_configuration.setname(name)
self.instances.configurations.appendconfiguration(new_configuration)
setattr(cls, "addconfiguration", addconfiguration)
def removeconfiguration(self, name):
found = False
for idx, configuration in enumerate(self.instances.configurations.getconfiguration()):
if configuration.getname() == name:
self.instances.configurations.removeconfiguration(idx)
found = True
break
if not found:
raise ValueError, ("\"%s\" configuration doesn't exist !!!")%name
setattr(cls, "removeconfiguration", removeconfiguration)
def getconfigurationResource(self, config_name, name):
configuration = self.getconfiguration(config_name)
if configuration:
for resource in configuration.getresource():
if resource.getname() == name:
return resource
return None
setattr(cls, "getconfigurationResource", getconfigurationResource)
def addconfigurationResource(self, config_name, name):
configuration = self.getconfiguration(config_name)
if configuration:
for resource in configuration.getresource():
if resource.getname() == name:
raise ValueError, _("\"%s\" resource already exists in \"%s\" configuration !!!")%(name, config_name)
new_resource = PLCOpenClasses["configuration_resource"]()
new_resource.setname(name)
configuration.appendresource(new_resource)
setattr(cls, "addconfigurationResource", addconfigurationResource)
def removeconfigurationResource(self, config_name, name):
configuration = self.getconfiguration(config_name)
if configuration:
found = False
for idx, resource in enumerate(configuration.getresource()):
if resource.getname() == name:
configuration.removeresource(idx)
found = True
break
if not found:
raise ValueError, _("\"%s\" resource doesn't exist in \"%s\" configuration !!!")%(name, config_name)
setattr(cls, "removeconfigurationResource", removeconfigurationResource)
def updateElementName(self, old_name, new_name):
for datatype in self.types.getdataTypeElements():
datatype_content = datatype.baseType.getcontent()
if datatype_content["name"] == "derived" and datatype_content["value"].getname() == old_name:
datatype_content["value"].setname(new_name)
elif datatype_content["name"] in ["array", "subrangeSigned", "subrangeUnsigned"]:
basetype_content = datatype_content["value"].baseType.getcontent()
if basetype_content["name"] == "derived" and basetype_content["value"].getname() == old_name:
basetype_content["value"].setname(new_name)
elif datatype_content["name"] == "struct":
for element in datatype_content["value"].getvariable():
element_type = element.type.getcontent()
if element_type["name"] == "derived" and element_type["value"].getname() == old_name:
element_type["value"].setname(new_name)
for pou in self.types.getpouElements():
pou.updateElementName(old_name, new_name)
for configuration in self.instances.configurations.getconfiguration():
configuration.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, old_leading, new_leading):
address_model = re.compile(FILTER_ADDRESS_MODEL % old_leading)
for pou in self.types.getpouElements():
pou.updateElementAddress(address_model, new_leading)
for configuration in self.instances.configurations.getconfiguration():
configuration.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def removeVariableByAddress(self, address):
for pou in self.types.getpouElements():
pou.removeVariableByAddress(address)
for configuration in self.instances.configurations.getconfiguration():
configuration.removeVariableByAddress(address)
setattr(cls, "removeVariableByAddress", removeVariableByAddress)
def removeVariableByFilter(self, leading):
address_model = re.compile(FILTER_ADDRESS_MODEL % leading)
for pou in self.types.getpouElements():
pou.removeVariableByFilter(address_model)
for configuration in self.instances.configurations.getconfiguration():
configuration.removeVariableByFilter(address_model)
setattr(cls, "removeVariableByFilter", removeVariableByFilter)
def RefreshDataTypeHierarchy(self):
self.EnumeratedDataTypeValues = {}
self.CustomDataTypeRange = {}
self.CustomTypeHierarchy = {}
for datatype in self.getdataTypes():
self.AddCustomDataType(datatype)
setattr(cls, "RefreshDataTypeHierarchy", RefreshDataTypeHierarchy)
def AddCustomDataType(self, datatype):
name = datatype.getname()
basetype_content = datatype.getbaseType().getcontent()
if basetype_content["value"] is None:
self.CustomTypeHierarchy[name] = basetype_content["name"]
elif basetype_content["name"] in ["string", "wstring"]:
self.CustomTypeHierarchy[name] = basetype_content["name"].upper()
elif basetype_content["name"] == "derived":
self.CustomTypeHierarchy[name] = basetype_content["value"].getname()
elif basetype_content["name"] in ["subrangeSigned", "subrangeUnsigned"]:
range = (basetype_content["value"].range.getlower(),
basetype_content["value"].range.getupper())
self.CustomDataTypeRange[name] = range
base_type = basetype_content["value"].baseType.getcontent()
if base_type["value"] is None:
self.CustomTypeHierarchy[name] = base_type["name"]
else:
self.CustomTypeHierarchy[name] = base_type["value"].getname()
else:
if basetype_content["name"] == "enum":
values = []
for value in basetype_content["value"].values.getvalue():
values.append(value.getname())
self.EnumeratedDataTypeValues[name] = values
self.CustomTypeHierarchy[name] = "ANY_DERIVED"
setattr(cls, "AddCustomDataType", AddCustomDataType)
# Update Block types with user-defined pou added
def RefreshCustomBlockTypes(self):
# Reset the tree of user-defined pou cross-use
self.CustomBlockTypes = []
for pou in self.getpous():
self.AddCustomBlockType(pou)
setattr(cls, "RefreshCustomBlockTypes", RefreshCustomBlockTypes)
def AddCustomBlockType(self, pou):
pou_name = pou.getname()
pou_type = pou.getpouType()
block_infos = {"name" : pou_name, "type" : pou_type, "extensible" : False,
"inputs" : [], "outputs" : [], "comment" : "",
"generate" : generate_block, "initialise" : initialise_block}
if pou.getinterface():
return_type = pou.interface.getreturnType()
if return_type:
var_type = return_type.getcontent()
if var_type["name"] == "derived":
block_infos["outputs"].append(("", var_type["value"].getname(), "none"))
elif var_type["name"] in ["string", "wstring"]:
block_infos["outputs"].append(("", var_type["name"].upper(), "none"))
else:
block_infos["outputs"].append(("", var_type["name"], "none"))
for type, varlist in pou.getvars():
if type == "InOut":
for var in varlist.getvariable():
var_type = var.type.getcontent()
if var_type["name"] == "derived":
block_infos["inputs"].append((var.getname(), var_type["value"].getname(), "none"))
block_infos["outputs"].append((var.getname(), var_type["value"].getname(), "none"))
elif var_type["name"] in ["string", "wstring"]:
block_infos["inputs"].append((var.getname(), var_type["name"].upper(), "none"))
block_infos["outputs"].append((var.getname(), var_type["name"].upper(), "none"))
else:
block_infos["inputs"].append((var.getname(), var_type["name"], "none"))
block_infos["outputs"].append((var.getname(), var_type["name"], "none"))
elif type == "Input":
for var in varlist.getvariable():
var_type = var.type.getcontent()
if var_type["name"] == "derived":
block_infos["inputs"].append((var.getname(), var_type["value"].getname(), "none"))
elif var_type["name"] in ["string", "wstring"]:
block_infos["inputs"].append((var.getname(), var_type["name"].upper(), "none"))
else:
block_infos["inputs"].append((var.getname(), var_type["name"], "none"))
elif type == "Output":
for var in varlist.getvariable():
var_type = var.type.getcontent()
if var_type["name"] == "derived":
block_infos["outputs"].append((var.getname(), var_type["value"].getname(), "none"))
elif var_type["name"] in ["string", "wstring"]:
block_infos["outputs"].append((var.getname(), var_type["name"].upper(), "none"))
else:
block_infos["outputs"].append((var.getname(), var_type["name"], "none"))
if pou.getbodyType() in ["FBD","LD","SFC"]:
for instance in pou.getinstances():
if isinstance(instance, PLCOpenClasses.get("commonObjects_comment", None)):
block_infos["comment"] = instance.getcontentText()
self.CustomBlockTypes.append(block_infos)
setattr(cls, "AddCustomBlockType", AddCustomBlockType)
def RefreshElementUsingTree(self):
# Reset the tree of user-defined element cross-use
self.ElementUsingTree = {}
pous = self.getpous()
datatypes = self.getdataTypes()
# Reference all the user-defined elementu names and initialize the tree of
# user-defined elemnt cross-use
elementnames = [datatype.getname() for datatype in datatypes] + \
[pou.getname() for pou in pous]
for name in elementnames:
self.ElementUsingTree[name] = []
# Analyze each datatype
for datatype in datatypes:
name = datatype.getname()
basetype_content = datatype.baseType.getcontent()
if basetype_content["name"] == "derived":
typename = basetype_content["value"].getname()
if name in self.ElementUsingTree[typename]:
self.ElementUsingTree[typename].append(name)
elif basetype_content["name"] in ["subrangeSigned", "subrangeUnsigned", "array"]:
base_type = basetype_content["value"].baseType.getcontent()
if base_type["name"] == "derived":
typename = base_type["value"].getname()
if self.ElementUsingTree.has_key(typename) and name not in self.ElementUsingTree[typename]:
self.ElementUsingTree[typename].append(name)
elif basetype_content["name"] == "struct":
for element in basetype_content["value"].getvariable():
type_content = element.type.getcontent()
if type_content["name"] == "derived":
typename = type_content["value"].getname()
if self.ElementUsingTree.has_key(typename) and name not in self.ElementUsingTree[typename]:
self.ElementUsingTree[typename].append(name)
# Analyze each pou
for pou in pous:
name = pou.getname()
if pou.interface:
# Extract variables from every varLists
for type, varlist in pou.getvars():
for var in varlist.getvariable():
vartype_content = var.gettype().getcontent()
if vartype_content["name"] == "derived":
typename = vartype_content["value"].getname()
if self.ElementUsingTree.has_key(typename) and name not in self.ElementUsingTree[typename]:
self.ElementUsingTree[typename].append(name)
setattr(cls, "RefreshElementUsingTree", RefreshElementUsingTree)
def GetParentType(self, type):
if self.CustomTypeHierarchy.has_key(type):
return self.CustomTypeHierarchy[type]
elif TypeHierarchy.has_key(type):
return TypeHierarchy[type]
return None
setattr(cls, "GetParentType", GetParentType)
def GetBaseType(self, type):
parent_type = self.GetParentType(type)
if parent_type is not None:
if parent_type.startswith("ANY"):
return type
else:
return self.GetBaseType(parent_type)
return None
setattr(cls, "GetBaseType", GetBaseType)
def GetSubrangeBaseTypes(self, exclude):
derived = []
for type in self.CustomTypeHierarchy.keys():
for base_type in DataTypeRange.keys():
if self.IsOfType(type, base_type) and not self.IsOfType(type, exclude):
derived.append(type)
break
return DataTypeRange.keys() + derived
setattr(cls, "GetSubrangeBaseTypes", GetSubrangeBaseTypes)
"""
returns true if the given data type is the same that "reference" meta-type or one of its types.
"""
def IsOfType(self, type, reference):
if reference is None:
return True
elif type == reference:
return True
else:
parent_type = self.GetParentType(type)
if parent_type is not None:
return self.IsOfType(parent_type, reference)
return False
setattr(cls, "IsOfType", IsOfType)
# Return if pou given by name is used by another pou
def ElementIsUsed(self, name):
if self.ElementUsingTree.has_key(name):
return len(self.ElementUsingTree[name]) > 0
return False
setattr(cls, "ElementIsUsed", ElementIsUsed)
def DataTypeIsDerived(self, name):
return name in self.CustomTypeHierarchy.values()
setattr(cls, "DataTypeIsDerived", DataTypeIsDerived)
# Return if pou given by name is directly or undirectly used by the reference pou
def ElementIsUsedBy(self, name, reference):
if self.ElementUsingTree.has_key(name):
list = self.ElementUsingTree[name]
# Test if pou is directly used by reference
if reference in list:
return True
else:
# Test if pou is undirectly used by reference, by testing if pous
# that directly use pou is directly or undirectly used by reference
used = False
for element in list:
used |= self.ElementIsUsedBy(element, reference)
return used
return False
setattr(cls, "ElementIsUsedBy", ElementIsUsedBy)
def GetDataTypeRange(self, type):
if self.CustomDataTypeRange.has_key(type):
return self.CustomDataTypeRange[type]
elif DataTypeRange.has_key(type):
return DataTypeRange[type]
else:
parent_type = self.GetParentType(type)
if parent_type is not None:
return self.GetDataTypeRange(parent_type)
return None
setattr(cls, "GetDataTypeRange", GetDataTypeRange)
def GetEnumeratedDataTypeValues(self, type = None):
if type is None:
all_values = []
for values in self.EnumeratedDataTypeValues.values():
all_values.extend(values)
return all_values
elif self.EnumeratedDataTypeValues.has_key(type):
return self.EnumeratedDataTypeValues[type]
return []
setattr(cls, "GetEnumeratedDataTypeValues", GetEnumeratedDataTypeValues)
# Function that returns the block definition associated to the block type given
def GetCustomBlockType(self, type, inputs = None):
for customblocktype in self.CustomBlockTypes:
if inputs is not None and inputs != "undefined":
customblock_inputs = tuple([var_type for name, var_type, modifier in customblocktype["inputs"]])
same_inputs = inputs == customblock_inputs
else:
same_inputs = True
if customblocktype["name"] == type and same_inputs:
return customblocktype
return None
setattr(cls, "GetCustomBlockType", GetCustomBlockType)
# Return Block types checking for recursion
def GetCustomBlockTypes(self, exclude = "", onlyfunctions = False):
type = None
if exclude != "":
pou = self.getpou(exclude)
if pou is not None:
type = pou.getpouType()
customblocktypes = []
for customblocktype in self.CustomBlockTypes:
if customblocktype["type"] != "program" and customblocktype["name"] != exclude and not self.ElementIsUsedBy(exclude, customblocktype["name"]) and not (onlyfunctions and customblocktype["type"] != "function"):
customblocktypes.append(customblocktype)
return customblocktypes
setattr(cls, "GetCustomBlockTypes", GetCustomBlockTypes)
# Return Function Block types checking for recursion
def GetCustomFunctionBlockTypes(self, exclude = ""):
customblocktypes = []
for customblocktype in self.CustomBlockTypes:
if customblocktype["type"] == "functionBlock" and customblocktype["name"] != exclude and not self.ElementIsUsedBy(exclude, customblocktype["name"]):
customblocktypes.append(customblocktype["name"])
return customblocktypes
setattr(cls, "GetCustomFunctionBlockTypes", GetCustomFunctionBlockTypes)
# Return Block types checking for recursion
def GetCustomBlockResource(self):
customblocktypes = []
for customblocktype in self.CustomBlockTypes:
if customblocktype["type"] == "program":
customblocktypes.append(customblocktype["name"])
return customblocktypes
setattr(cls, "GetCustomBlockResource", GetCustomBlockResource)
# Return Data Types checking for recursion
def GetCustomDataTypes(self, exclude = "", only_locatable = False):
customdatatypes = []
for customdatatype in self.getdataTypes():
if not only_locatable or self.IsLocatableType(customdatatype):
customdatatype_name = customdatatype.getname()
if customdatatype_name != exclude and not self.ElementIsUsedBy(exclude, customdatatype_name):
customdatatypes.append(customdatatype_name)
return customdatatypes
setattr(cls, "GetCustomDataTypes", GetCustomDataTypes)
# Return if Data Type can be used for located variables
def IsLocatableType(self, datatype):
basetype_content = datatype.baseType.getcontent()
if basetype_content["name"] in ["enum", "struct"]:
return False
elif basetype_content["name"] == "derived":
base_type = self.getdataType(basetype_content["value"].getname())
if base_type is not None:
return self.IsLocatableType(base_type)
elif basetype_content["name"] == "array":
array_base_type = basetype_content["value"].baseType.getcontent()
if array_base_type["value"] is not None and array_base_type["name"] not in ["string", "wstring"]:
base_type = self.getdataType(array_base_type["value"].getname())
if base_type is not None:
return self.IsLocatableType(base_type)
return True
setattr(cls, "IsLocatableType", IsLocatableType)
def Search(self, criteria, parent_infos=[]):
result = self.types.Search(criteria, parent_infos)
for configuration in self.instances.configurations.getconfiguration():
result.extend(configuration.Search(criteria, parent_infos))
return result
setattr(cls, "Search", Search)
cls = PLCOpenClasses.get("project_fileHeader", None)
if cls:
cls.singleLineAttributes = False
cls = PLCOpenClasses.get("project_contentHeader", None)
if cls:
cls.singleLineAttributes = False
def setpageSize(self, width, height):
self.coordinateInfo.setpageSize(width, height)
setattr(cls, "setpageSize", setpageSize)
def getpageSize(self):
return self.coordinateInfo.getpageSize()
setattr(cls, "getpageSize", getpageSize)
def setscaling(self, scaling):
for language, (x, y) in scaling.items():
self.coordinateInfo.setscaling(language, x, y)
setattr(cls, "setscaling", setscaling)
def getscaling(self):
scaling = {}
scaling["FBD"] = self.coordinateInfo.getscaling("FBD")
scaling["LD"] = self.coordinateInfo.getscaling("LD")
scaling["SFC"] = self.coordinateInfo.getscaling("SFC")
return scaling
setattr(cls, "getscaling", getscaling)
cls = PLCOpenClasses.get("contentHeader_coordinateInfo", None)
if cls:
def setpageSize(self, width, height):
if width == 0 and height == 0:
self.deletepageSize()
else:
if self.pageSize is None:
self.addpageSize()
self.pageSize.setx(width)
self.pageSize.sety(height)
setattr(cls, "setpageSize", setpageSize)
def getpageSize(self):
if self.pageSize is not None:
return self.pageSize.getx(), self.pageSize.gety()
return 0, 0
setattr(cls, "getpageSize", getpageSize)
def setscaling(self, language, x, y):
if language == "FBD":
self.fbd.scaling.setx(x)
self.fbd.scaling.sety(y)
elif language == "LD":
self.ld.scaling.setx(x)
self.ld.scaling.sety(y)
elif language == "SFC":
self.sfc.scaling.setx(x)
self.sfc.scaling.sety(y)
setattr(cls, "setscaling", setscaling)
def getscaling(self, language):
if language == "FBD":
return self.fbd.scaling.getx(), self.fbd.scaling.gety()
elif language == "LD":
return self.ld.scaling.getx(), self.ld.scaling.gety()
elif language == "SFC":
return self.sfc.scaling.getx(), self.sfc.scaling.gety()
return 0, 0
setattr(cls, "getscaling", getscaling)
def _Search(attributes, criteria, parent_infos):
search_result = []
for attr, value in attributes:
if value is not None:
search_result.extend([(tuple(parent_infos + [attr]),) + result for result in TestTextElement(value, criteria)])
return search_result
def _updateConfigurationResourceElementName(self, old_name, new_name):
for varlist in self.getglobalVars():
for var in varlist.getvariable():
var_address = var.getaddress()
if var_address is not None:
if var_address == old_name:
var.setaddress(new_name)
if var.getname() == old_name:
var.setname(new_name)
def _updateConfigurationResourceElementAddress(self, address_model, new_leading):
for varlist in self.getglobalVars():
for var in varlist.getvariable():
var_address = var.getaddress()
if var_address is not None:
var.setaddress(update_address(var_address, address_model, new_leading))
def _removeConfigurationResourceVariableByAddress(self, address):
for varlist in self.getglobalVars():
variables = varlist.getvariable()
for i in xrange(len(variables)-1, -1, -1):
if variables[i].getaddress() == address:
variables.pop(i)
def _removeConfigurationResourceVariableByFilter(self, address_model):
for varlist in self.getglobalVars():
variables = varlist.getvariable()
for i in xrange(len(variables)-1, -1, -1):
var_address = variables[i].getaddress()
if var_address is not None:
result = address_model.match(var_address)
if result is not None:
variables.pop(i)
def _SearchInConfigurationResource(self, criteria, parent_infos=[]):
search_result = []
for result in TestTextElement(self.getname(), criteria):
search_result.append((tuple(parent_infos + ["name"]),) + result)
var_number = 0
for varlist in self.getglobalVars():
variable_type = searchResultVarTypes.get("globalVars", "var_local")
variables = varlist.getvariable()
for modifier, has_modifier in [("constant", varlist.getconstant()),
("retain", varlist.getretain()),
("non_retain", varlist.getnonretain())]:
if has_modifier:
for result in TestTextElement(modifier, criteria):
search_result.append((tuple(parent_infos + [variable_type, (var_number, var_number + len(variables)), modifier]),) + result)
break
for variable in variables:
search_result.extend(variable.Search(criteria, parent_infos + [variable_type, var_number]))
var_number += 1
return search_result
cls = PLCOpenClasses.get("configurations_configuration", None)
if cls:
def updateElementName(self, old_name, new_name):
_updateConfigurationResourceElementName(self, old_name, new_name)
for resource in self.getresource():
resource.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
_updateConfigurationResourceElementAddress(self, address_model, new_leading)
for resource in self.getresource():
resource.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
setattr(cls, "removeVariableByAddress", _removeConfigurationResourceVariableByAddress)
setattr(cls, "removeVariableByFilter", _removeConfigurationResourceVariableByFilter)
def Search(self, criteria, parent_infos=[]):
search_result = []
parent_infos = parent_infos + ["C::%s" % self.getname()]
filter = criteria["filter"]
if filter == "all" or "configuration" in filter:
search_result = _SearchInConfigurationResource(self, criteria, parent_infos)
for resource in self.getresource():
search_result.extend(resource.Search(criteria, parent_infos))
return search_result
setattr(cls, "Search", Search)
cls = PLCOpenClasses.get("configuration_resource", None)
if cls:
def updateElementName(self, old_name, new_name):
_updateConfigurationResourceElementName(self, old_name, new_name)
for instance in self.getpouInstance():
instance.updateElementName(old_name, new_name)
for task in self.gettask():
task.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
_updateConfigurationResourceElementAddress(self, address_model, new_leading)
for task in self.gettask():
task.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
setattr(cls, "removeVariableByAddress", _removeConfigurationResourceVariableByAddress)
setattr(cls, "removeVariableByFilter", _removeConfigurationResourceVariableByFilter)
def Search(self, criteria, parent_infos=[]):
parent_infos = parent_infos[:-1] + ["R::%s::%s" % (parent_infos[-1].split("::")[1], self.getname())]
search_result = _SearchInConfigurationResource(self, criteria, parent_infos)
task_number = 0
instance_number = 0
for task in self.gettask():
results = TestTextElement(task.getname(), criteria)
for result in results:
search_result.append((tuple(parent_infos + ["task", task_number, "name"]),) + result)
search_result.extend(task.Search(criteria, parent_infos + ["task", task_number]))
task_number += 1
for instance in task.getpouInstance():
search_result.extend(task.Search(criteria, parent_infos + ["instance", instance_number]))
for result in results:
search_result.append((tuple(parent_infos + ["instance", instance_number, "task"]),) + result)
instance_number += 1
for instance in self.getpouInstance():
search_result.extend(instance.Search(criteria, parent_infos + ["instance", instance_number]))
instance_number += 1
return search_result
setattr(cls, "Search", Search)
cls = PLCOpenClasses.get("resource_task", None)
if cls:
def compatibility(self, tree):
if tree.hasAttribute("interval"):
interval = GetAttributeValue(tree._attrs["interval"])
result = time_model.match(interval)
if result is not None:
values = result.groups()
time_values = [int(v) for v in values[:2]]
seconds = float(values[2])
time_values.extend([int(seconds), int((seconds % 1) * 1000000)])
text = "t#"
if time_values[0] != 0:
text += "%dh"%time_values[0]
if time_values[1] != 0:
text += "%dm"%time_values[1]
if time_values[2] != 0:
text += "%ds"%time_values[2]
if time_values[3] != 0:
if time_values[3] % 1000 != 0:
text += "%.3fms"%(float(time_values[3]) / 1000)
else:
text += "%dms"%(time_values[3] / 1000)
NodeSetAttr(tree, "interval", text)
setattr(cls, "compatibility", compatibility)
def updateElementName(self, old_name, new_name):
if self.single == old_name:
self.single = new_name
if self.interval == old_name:
self.interval = new_name
for instance in self.getpouInstance():
instance.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
if self.single is not None:
self.single = update_address(self.single, address_model, new_leading)
if self.interval is not None:
self.interval = update_address(self.interval, address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def Search(self, criteria, parent_infos=[]):
return _Search([("single", self.getsingle()),
("interval", self.getinterval()),
("priority", str(self.getpriority()))],
criteria, parent_infos)
setattr(cls, "Search", Search)
cls = PLCOpenClasses.get("pouInstance", None)
if cls:
def compatibility(self, tree):
if tree.hasAttribute("type"):
NodeRenameAttr(tree, "type", "typeName")
setattr(cls, "compatibility", compatibility)
def updateElementName(self, old_name, new_name):
if self.typeName == old_name:
self.typeName = new_name
setattr(cls, "updateElementName", updateElementName)
def Search(self, criteria, parent_infos=[]):
return _Search([("name", self.getname()),
("type", self.gettypeName())],
criteria, parent_infos)
setattr(cls, "Search", Search)
cls = PLCOpenClasses.get("varListPlain_variable", None)
if cls:
def gettypeAsText(self):
vartype_content = self.gettype().getcontent()
# Variable type is a user data type
if vartype_content["name"] == "derived":
return vartype_content["value"].getname()
# Variable type is a string type
elif vartype_content["name"] in ["string", "wstring"]:
return vartype_content["name"].upper()
# Variable type is an array
elif vartype_content["name"] == "array":
base_type = vartype_content["value"].baseType.getcontent()
# Array derived directly from a user defined type
if base_type["name"] == "derived":
basetype_name = base_type["value"].getname()
# Array derived directly from a string type
elif base_type["name"] in ["string", "wstring"]:
basetype_name = base_type["name"].upper()
# Array derived directly from an elementary type
else:
basetype_name = base_type["name"]
return "ARRAY [%s] OF %s" % (",".join(map(lambda x : "%s..%s" % (x.getlower(), x.getupper()), vartype_content["value"].getdimension())), basetype_name)
# Variable type is an elementary type
return vartype_content["name"]
setattr(cls, "gettypeAsText", gettypeAsText)
def Search(self, criteria, parent_infos=[]):
search_result = _Search([("name", self.getname()),
("type", self.gettypeAsText()),
("address", self.getaddress())],
criteria, parent_infos)
initial = self.getinitialValue()
if initial is not None:
search_result.extend(_Search([("initial", initial.getvalue())], criteria, parent_infos))
doc = self.getdocumentation()
if doc is not None:
search_result.extend(doc.Search(criterias, parent_infos + ["documentation"]))
return search_result
setattr(cls, "Search", Search)
cls = PLCOpenClasses.get("project_types", None)
if cls:
def getdataTypeElements(self):
return self.dataTypes.getdataType()
setattr(cls, "getdataTypeElements", getdataTypeElements)
def getdataTypeElement(self, name):
elements = self.dataTypes.getdataType()
for element in elements:
if element.getname() == name:
return element
return None
setattr(cls, "getdataTypeElement", getdataTypeElement)
def appenddataTypeElement(self, name):
new_datatype = PLCOpenClasses["dataTypes_dataType"]()
new_datatype.setname(name)
new_datatype.baseType.setcontent({"name" : "BOOL", "value" : None})
self.dataTypes.appenddataType(new_datatype)
setattr(cls, "appenddataTypeElement", appenddataTypeElement)
def insertdataTypeElement(self, index, dataType):
self.dataTypes.insertdataType(index, dataType)
setattr(cls, "insertdataTypeElement", insertdataTypeElement)
def removedataTypeElement(self, name):
found = False
for idx, element in enumerate(self.dataTypes.getdataType()):
if element.getname() == name:
self.dataTypes.removedataType(idx)
found = True
break
if not found:
raise ValueError, _("\"%s\" Data Type doesn't exist !!!")%name
setattr(cls, "removedataTypeElement", removedataTypeElement)
def getpouElements(self):
return self.pous.getpou()
setattr(cls, "getpouElements", getpouElements)
def getpouElement(self, name):
elements = self.pous.getpou()
for element in elements:
if element.getname() == name:
return element
return None
setattr(cls, "getpouElement", getpouElement)
def appendpouElement(self, name, pou_type, body_type):
for element in self.pous.getpou():
if element.getname() == name:
raise ValueError, _("\"%s\" POU already exists !!!")%name
new_pou = PLCOpenClasses["pous_pou"]()
new_pou.setname(name)
new_pou.setpouType(pou_type)
new_pou.appendbody(PLCOpenClasses["body"]())
new_pou.setbodyType(body_type)
self.pous.appendpou(new_pou)
setattr(cls, "appendpouElement", appendpouElement)
def insertpouElement(self, index, pou):
self.pous.insertpou(index, pou)
setattr(cls, "insertpouElement", insertpouElement)
def removepouElement(self, name):
found = False
for idx, element in enumerate(self.pous.getpou()):
if element.getname() == name:
self.pous.removepou(idx)
found = True
break
if not found:
raise ValueError, _("\"%s\" POU doesn't exist !!!")%name
setattr(cls, "removepouElement", removepouElement)
def Search(self, criteria, parent_infos=[]):
search_result = []
filter = criteria["filter"]
#if filter == "all" or "datatype" in filter:
# for datatype in self.dataTypes.getdataType():
# search_result.extend(datatype.Search(criteria, parent_infos))
for pou in self.pous.getpou():
search_result.extend(pou.Search(criteria, parent_infos))
return search_result
setattr(cls, "Search", Search)
cls = PLCOpenClasses.get("pous_pou", None)
if cls:
def setbodyType(self, type):
if len(self.body) > 0:
if type == "IL":
self.body[0].setcontent({"name" : "IL", "value" : PLCOpenClasses["formattedText"]()})
elif type == "ST":
self.body[0].setcontent({"name" : "ST", "value" : PLCOpenClasses["formattedText"]()})
elif type == "LD":
self.body[0].setcontent({"name" : "LD", "value" : PLCOpenClasses["body_LD"]()})
elif type == "FBD":
self.body[0].setcontent({"name" : "FBD", "value" : PLCOpenClasses["body_FBD"]()})
elif type == "SFC":
self.body[0].setcontent({"name" : "SFC", "value" : PLCOpenClasses["body_SFC"]()})
else:
raise ValueError, "%s isn't a valid body type!"%type
setattr(cls, "setbodyType", setbodyType)
def getbodyType(self):
if len(self.body) > 0:
return self.body[0].getcontent()["name"]
setattr(cls, "getbodyType", getbodyType)
def resetexecutionOrder(self):
if len(self.body) > 0:
self.body[0].resetexecutionOrder()
setattr(cls, "resetexecutionOrder", resetexecutionOrder)
def compileexecutionOrder(self):
if len(self.body) > 0:
self.body[0].compileexecutionOrder()
setattr(cls, "compileexecutionOrder", compileexecutionOrder)
def setelementExecutionOrder(self, instance, new_executionOrder):
if len(self.body) > 0:
self.body[0].setelementExecutionOrder(instance, new_executionOrder)
setattr(cls, "setelementExecutionOrder", setelementExecutionOrder)
def addinstance(self, name, instance):
if len(self.body) > 0:
self.body[0].appendcontentInstance(name, instance)
setattr(cls, "addinstance", addinstance)
def getinstances(self):
if len(self.body) > 0:
return self.body[0].getcontentInstances()
return []
setattr(cls, "getinstances", getinstances)
def getinstance(self, id):
if len(self.body) > 0:
return self.body[0].getcontentInstance(id)
return None
setattr(cls, "getinstance", getinstance)
def getrandomInstance(self, exclude):
if len(self.body) > 0:
return self.body[0].getcontentRandomInstance(exclude)
return None
setattr(cls, "getrandomInstance", getrandomInstance)
def getinstanceByName(self, name):
if len(self.body) > 0:
return self.body[0].getcontentInstanceByName(name)
return None
setattr(cls, "getinstanceByName", getinstanceByName)
def removeinstance(self, id):
if len(self.body) > 0:
self.body[0].removecontentInstance(id)
setattr(cls, "removeinstance", removeinstance)
def settext(self, text):
if len(self.body) > 0:
self.body[0].settext(text)
setattr(cls, "settext", settext)
def gettext(self):
if len(self.body) > 0:
return self.body[0].gettext()
return ""
setattr(cls, "gettext", gettext)
def getvars(self):
vars = []
if self.interface is not None:
reverse_types = {}
for name, value in VarTypes.items():
reverse_types[value] = name
for varlist in self.interface.getcontent():
vars.append((reverse_types[varlist["name"]], varlist["value"]))
return vars
setattr(cls, "getvars", getvars)
def setvars(self, vars):
if self.interface is None:
self.interface = PLCOpenClasses["pou_interface"]()
self.interface.setcontent([])
for vartype, varlist in vars:
self.interface.appendcontent({"name" : VarTypes[vartype], "value" : varlist})
setattr(cls, "setvars", setvars)
def addpouVar(self, type, name, location="", description=""):
if self.interface is None:
self.interface = PLCOpenClasses["pou_interface"]()
content = self.interface.getcontent()
if len(content) == 0 or content[-1]["name"] != "localVars":
content.append({"name" : "localVars", "value" : PLCOpenClasses["interface_localVars"]()})
else:
varlist = content[-1]["value"]
variables = varlist.getvariable()
if varlist.getconstant() or varlist.getretain() or len(variables) > 0 and variables[0].getaddress():
content.append({"name" : "localVars", "value" : PLCOpenClasses["interface_localVars"]()})
var = PLCOpenClasses["varListPlain_variable"]()
var.setname(name)
var_type = PLCOpenClasses["dataType"]()
if type in [x for x,y in TypeHierarchy_list if not x.startswith("ANY")]:
if type == "STRING":
var_type.setcontent({"name" : "string", "value" : PLCOpenClasses["elementaryTypes_string"]()})
elif type == "WSTRING":
var_type.setcontent({"name" : "wstring", "value" : PLCOpenClasses["elementaryTypes_wstring"]()})
else:
var_type.setcontent({"name" : type, "value" : None})
else:
derived_type = PLCOpenClasses["derivedTypes_derived"]()
derived_type.setname(type)
var_type.setcontent({"name" : "derived", "value" : derived_type})
var.settype(var_type)
if location != "":
var.setaddress(location)
if description != "":
ft = PLCOpenClasses["formattedText"]()
ft.settext(description)
var.setdocumentation(ft)
content[-1]["value"].appendvariable(var)
setattr(cls, "addpouVar", addpouVar)
def changepouVar(self, old_type, old_name, new_type, new_name):
if self.interface is not None:
content = self.interface.getcontent()
for varlist in content:
variables = varlist["value"].getvariable()
for var in variables:
if var.getname() == old_name:
vartype_content = var.gettype().getcontent()
if vartype_content["name"] == "derived" and vartype_content["value"].getname() == old_type:
var.setname(new_name)
vartype_content["value"].setname(new_type)
return
setattr(cls, "changepouVar", changepouVar)
def removepouVar(self, type, name):
if self.interface is not None:
content = self.interface.getcontent()
for varlist in content:
variables = varlist["value"].getvariable()
for var in variables:
if var.getname() == name:
vartype_content = var.gettype().getcontent()
if vartype_content["name"] == "derived" and vartype_content["value"].getname() == type:
variables.remove(var)
break
if len(varlist["value"].getvariable()) == 0:
content.remove(varlist)
break
setattr(cls, "removepouVar", removepouVar)
def hasblock(self, name):
if self.getbodyType() in ["FBD", "LD", "SFC"]:
for instance in self.getinstances():
if isinstance(instance, PLCOpenClasses["fbdObjects_block"]) and instance.getinstanceName() == name:
return True
if self.transitions:
for transition in self.transitions.gettransition():
result = transition.hasblock(name)
if result:
return result
if self.actions:
for action in self.actions.getaction():
result = action.hasblock(name)
if result:
return result
return False
setattr(cls, "hasblock", hasblock)
def addtransition(self, name, type):
if not self.transitions:
self.addtransitions()
self.transitions.settransition([])
transition = PLCOpenClasses["transitions_transition"]()
transition.setname(name)
transition.setbodyType(type)
if type == "ST":
transition.settext(":= ;")
elif type == "IL":
transition.settext("\tST\t%s"%name)
self.transitions.appendtransition(transition)
setattr(cls, "addtransition", addtransition)
def gettransition(self, name):
if self.transitions:
for transition in self.transitions.gettransition():
if transition.getname() == name:
return transition
return None
setattr(cls, "gettransition", gettransition)
def gettransitionList(self):
if self.transitions:
return self.transitions.gettransition()
return []
setattr(cls, "gettransitionList", gettransitionList)
def removetransition(self, name):
if self.transitions:
transitions = self.transitions.gettransition()
i = 0
removed = False
while i < len(transitions) and not removed:
if transitions[i].getname() == name:
transitions.pop(i)
removed = True
i += 1
if not removed:
raise ValueError, _("Transition with name %s doesn't exist!")%name
setattr(cls, "removetransition", removetransition)
def addaction(self, name, type):
if not self.actions:
self.addactions()
self.actions.setaction([])
action = PLCOpenClasses["actions_action"]()
action.setname(name)
action.setbodyType(type)
self.actions.appendaction(action)
setattr(cls, "addaction", addaction)
def getaction(self, name):
if self.actions:
for action in self.actions.getaction():
if action.getname() == name:
return action
return None
setattr(cls, "getaction", getaction)
def getactionList(self):
if self.actions:
return self.actions.getaction()
return []
setattr(cls, "getactionList", getactionList)
def removeaction(self, name):
if self.actions:
actions = self.actions.getaction()
i = 0
removed = False
while i < len(actions) and not removed:
if actions[i].getname() == name:
actions.pop(i)
removed = True
i += 1
if not removed:
raise ValueError, _("Action with name %s doesn't exist!")%name
setattr(cls, "removeaction", removeaction)
def updateElementName(self, old_name, new_name):
if self.interface:
for content in self.interface.getcontent():
for var in content["value"].getvariable():
var_address = var.getaddress()
if var_address is not None:
if var_address == old_name:
var.setaddress(new_name)
if var.getname() == old_name:
var.setname(new_name)
var_type_content = var.gettype().getcontent()
if var_type_content["name"] == "derived":
if var_type_content["value"].getname() == old_name:
var_type_content["value"].setname(new_name)
self.body[0].updateElementName(old_name, new_name)
for action in self.getactionList():
action.updateElementName(old_name, new_name)
for transition in self.gettransitionList():
transition.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
if self.interface:
for content in self.interface.getcontent():
for var in content["value"].getvariable():
var_address = var.getaddress()
if var_address is not None:
var.setaddress(update_address(var_address, address_model, new_leading))
self.body[0].updateElementAddress(address_model, new_leading)
for action in self.getactionList():
action.updateElementAddress(address_model, new_leading)
for transition in self.gettransitionList():
transition.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def removeVariableByAddress(self, address):
if self.interface:
for content in self.interface.getcontent():
variables = content["value"].getvariable()
for i in xrange(len(variables)-1, -1, -1):
if variables[i].getaddress() == address:
variables.pop(i)
setattr(cls, "removeVariableByAddress", removeVariableByAddress)
def removeVariableByFilter(self, address_model):
if self.interface:
for content in self.interface.getcontent():
variables = content["value"].getvariable()
for i in xrange(len(variables)-1, -1, -1):
var_address = variables[i].getaddress()
if var_address is not None:
result = address_model.match(var_address)
if result is not None:
variables.pop(i)
setattr(cls, "removeVariableByFilter", removeVariableByFilter)
def Search(self, criteria, parent_infos=[]):
search_result = []
filter = criteria["filter"]
if filter == "all" or self.getpouType() in filter:
parent_infos = parent_infos + ["P::%s" % self.getname()]
for result in TestTextElement(self.getname(), criteria):
search_result.append((tuple(parent_infos + ["name"]),) + result)
if self.interface is not None:
var_number = 0
for content in self.interface.getcontent():
variable_type = searchResultVarTypes.get(content["value"], "var_local")
variables = content["value"].getvariable()
for modifier, has_modifier in [("constant", content["value"].getconstant()),
("retain", content["value"].getretain()),
("non_retain", content["value"].getnonretain())]:
if has_modifier:
for result in TestTextElement(modifier, criteria):
search_result.append((tuple(parent_infos + [variable_type, (var_number, var_number + len(variables)), modifier]),) + result)
break
for variable in variables:
search_result.extend(variable.Search(criteria, parent_infos + [variable_type, var_number]))
var_number += 1
if len(self.body) > 0:
search_result.extend(self.body[0].Search(criteria, parent_infos))
for action in self.getactionList():
search_result.extend(action.Search(criteria, parent_infos))
for transition in self.gettransitionList():
search_result.extend(transition.Search(criteria, parent_infos))
return search_result
setattr(cls, "Search", Search)
def setbodyType(self, type):
if type == "IL":
self.body.setcontent({"name" : "IL", "value" : PLCOpenClasses["formattedText"]()})
elif type == "ST":
self.body.setcontent({"name" : "ST", "value" : PLCOpenClasses["formattedText"]()})
elif type == "LD":
self.body.setcontent({"name" : "LD", "value" : PLCOpenClasses["body_LD"]()})
elif type == "FBD":
self.body.setcontent({"name" : "FBD", "value" : PLCOpenClasses["body_FBD"]()})
elif type == "SFC":
self.body.setcontent({"name" : "SFC", "value" : PLCOpenClasses["body_SFC"]()})
else:
raise ValueError, "%s isn't a valid body type!"%type
def getbodyType(self):
return self.body.getcontent()["name"]
def resetexecutionOrder(self):
self.body.resetexecutionOrder()
def compileexecutionOrder(self):
self.body.compileexecutionOrder()
def setelementExecutionOrder(self, instance, new_executionOrder):
self.body.setelementExecutionOrder(instance, new_executionOrder)
def addinstance(self, name, instance):
self.body.appendcontentInstance(name, instance)
def getinstances(self):
return self.body.getcontentInstances()
def getinstance(self, id):
return self.body.getcontentInstance(id)
def getrandomInstance(self, exclude):
return self.body.getcontentRandomInstance(exclude)
def getinstanceByName(self, name):
return self.body.getcontentInstanceByName(name)
def removeinstance(self, id):
self.body.removecontentInstance(id)
def settext(self, text):
self.body.settext(text)
def gettext(self):
return self.body.gettext()
cls = PLCOpenClasses.get("transitions_transition", None)
if cls:
setattr(cls, "setbodyType", setbodyType)
setattr(cls, "getbodyType", getbodyType)
setattr(cls, "resetexecutionOrder", resetexecutionOrder)
setattr(cls, "compileexecutionOrder", compileexecutionOrder)
setattr(cls, "setelementExecutionOrder", setelementExecutionOrder)
setattr(cls, "addinstance", addinstance)
setattr(cls, "getinstances", getinstances)
setattr(cls, "getinstance", getinstance)
setattr(cls, "getrandomInstance", getrandomInstance)
setattr(cls, "getinstanceByName", getinstanceByName)
setattr(cls, "removeinstance", removeinstance)
setattr(cls, "settext", settext)
setattr(cls, "gettext", gettext)
def updateElementName(self, old_name, new_name):
self.body.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
self.body.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def hasblock(self, name):
if self.getbodyType() in ["FBD", "LD", "SFC"]:
for instance in self.getinstances():
if isinstance(instance, PLCOpenClasses["fbdObjects_block"]) and instance.getinstanceName() == name:
return True
return False
setattr(cls, "hasblock", hasblock)
def Search(self, criteria, parent_infos):
search_result = []
parent_infos = parent_infos[:-1] + ["T::%s::%s" % (parent_infos[-1].split("::")[1], self.getname())]
for result in TestTextElement(self.getname(), criteria):
search_result.append((tuple(parent_infos + ["name"]),) + result)
search_result.extend(self.body.Search(criteria, parent_infos))
return search_result
setattr(cls, "Search", Search)
cls = PLCOpenClasses.get("actions_action", None)
if cls:
setattr(cls, "setbodyType", setbodyType)
setattr(cls, "getbodyType", getbodyType)
setattr(cls, "resetexecutionOrder", resetexecutionOrder)
setattr(cls, "compileexecutionOrder", compileexecutionOrder)
setattr(cls, "setelementExecutionOrder", setelementExecutionOrder)
setattr(cls, "addinstance", addinstance)
setattr(cls, "getinstances", getinstances)
setattr(cls, "getinstance", getinstance)
setattr(cls, "getrandomInstance", getrandomInstance)
setattr(cls, "getinstanceByName", getinstanceByName)
setattr(cls, "removeinstance", removeinstance)
setattr(cls, "settext", settext)
setattr(cls, "gettext", gettext)
def updateElementName(self, old_name, new_name):
self.body.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
self.body.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def hasblock(self, name):
if self.getbodyType() in ["FBD", "LD", "SFC"]:
for instance in self.getinstances():
if isinstance(instance, PLCOpenClasses["fbdObjects_block"]) and instance.getinstanceName() == name:
return True
return False
setattr(cls, "hasblock", hasblock)
def Search(self, criteria, parent_infos):
search_result = []
parent_infos = parent_infos[:-1] + ["A::%s::%s" % (parent_infos[-1].split("::")[1], self.getname())]
for result in TestTextElement(self.getname(), criteria):
search_result.append((tuple(parent_infos + ["name"]),) + result)
search_result.extend(self.body.Search(criteria, parent_infos))
return search_result
setattr(cls, "Search", Search)
cls = PLCOpenClasses.get("body", None)
if cls:
cls.currentExecutionOrderId = 0
def resetcurrentExecutionOrderId(self):
object.__setattr__(self, "currentExecutionOrderId", 0)
setattr(cls, "resetcurrentExecutionOrderId", resetcurrentExecutionOrderId)
def getnewExecutionOrderId(self):
object.__setattr__(self, "currentExecutionOrderId", self.currentExecutionOrderId + 1)
return self.currentExecutionOrderId
setattr(cls, "getnewExecutionOrderId", getnewExecutionOrderId)
def resetexecutionOrder(self):
if self.content["name"] == "FBD":
for element in self.content["value"].getcontent():
if not isinstance(element["value"], (PLCOpenClasses.get("commonObjects_comment", None),
PLCOpenClasses.get("commonObjects_connector", None),
PLCOpenClasses.get("commonObjects_continuation", None))):
element["value"].setexecutionOrderId(0)
else:
raise TypeError, _("Can only generate execution order on FBD networks!")
setattr(cls, "resetexecutionOrder", resetexecutionOrder)
def compileexecutionOrder(self):
if self.content["name"] == "FBD":
self.resetexecutionOrder()
self.resetcurrentExecutionOrderId()
for element in self.content["value"].getcontent():
if isinstance(element["value"], PLCOpenClasses.get("fbdObjects_outVariable", None)) and element["value"].getexecutionOrderId() == 0:
connections = element["value"].connectionPointIn.getconnections()
if connections and len(connections) == 1:
self.compileelementExecutionOrder(connections[0])
element["value"].setexecutionOrderId(self.getnewExecutionOrderId())
else:
raise TypeError, _("Can only generate execution order on FBD networks!")
setattr(cls, "compileexecutionOrder", compileexecutionOrder)
def compileelementExecutionOrder(self, link):
if self.content["name"] == "FBD":
localid = link.getrefLocalId()
instance = self.getcontentInstance(localid)
if isinstance(instance, PLCOpenClasses.get("fbdObjects_block", None)) and instance.getexecutionOrderId() == 0:
for variable in instance.inputVariables.getvariable():
connections = variable.connectionPointIn.getconnections()
if connections and len(connections) == 1:
self.compileelementExecutionOrder(connections[0])
instance.setexecutionOrderId(self.getnewExecutionOrderId())
elif isinstance(instance, PLCOpenClasses.get("commonObjects_continuation", None)) and instance.getexecutionOrderId() == 0:
name = instance.getname()
for tmp_instance in self.getcontentInstances():
if isinstance(tmp_instance, PLCOpenClasses.get("commonObjects_connector", None)) and tmp_instance.getname() == name and tmp_instance.getexecutionOrderId() == 0:
connections = tmp_instance.connectionPointIn.getconnections()
if connections and len(connections) == 1:
self.compileelementExecutionOrder(connections[0])
else:
raise TypeError, _("Can only generate execution order on FBD networks!")
setattr(cls, "compileelementExecutionOrder", compileelementExecutionOrder)
def setelementExecutionOrder(self, instance, new_executionOrder):
if self.content["name"] == "FBD":
old_executionOrder = instance.getexecutionOrderId()
if old_executionOrder is not None and old_executionOrder != 0 and new_executionOrder != 0:
for element in self.content["value"].getcontent():
if element["value"] != instance and not isinstance(element["value"], PLCOpenClasses.get("commonObjects_comment", None)):
element_executionOrder = element["value"].getexecutionOrderId()
if old_executionOrder <= element_executionOrder <= new_executionOrder:
element["value"].setexecutionOrderId(element_executionOrder - 1)
if new_executionOrder <= element_executionOrder <= old_executionOrder:
element["value"].setexecutionOrderId(element_executionOrder + 1)
instance.setexecutionOrderId(new_executionOrder)
else:
raise TypeError, _("Can only generate execution order on FBD networks!")
setattr(cls, "setelementExecutionOrder", setelementExecutionOrder)
def appendcontentInstance(self, name, instance):
if self.content["name"] in ["LD","FBD","SFC"]:
self.content["value"].appendcontent({"name" : name, "value" : instance})
else:
raise TypeError, _("%s body don't have instances!")%self.content["name"]
setattr(cls, "appendcontentInstance", appendcontentInstance)
def getcontentInstances(self):
if self.content["name"] in ["LD","FBD","SFC"]:
instances = []
for element in self.content["value"].getcontent():
instances.append(element["value"])
return instances
else:
raise TypeError, _("%s body don't have instances!")%self.content["name"]
setattr(cls, "getcontentInstances", getcontentInstances)
def getcontentInstance(self, id):
if self.content["name"] in ["LD","FBD","SFC"]:
for element in self.content["value"].getcontent():
if element["value"].getlocalId() == id:
return element["value"]
return None
else:
raise TypeError, _("%s body don't have instances!")%self.content["name"]
setattr(cls, "getcontentInstance", getcontentInstance)
def getcontentRandomInstance(self, exclude):
if self.content["name"] in ["LD","FBD","SFC"]:
for element in self.content["value"].getcontent():
if element["value"].getlocalId() not in exclude:
return element["value"]
return None
else:
raise TypeError, _("%s body don't have instances!")%self.content["name"]
setattr(cls, "getcontentRandomInstance", getcontentRandomInstance)
def getcontentInstanceByName(self, name):
if self.content["name"] in ["LD","FBD","SFC"]:
for element in self.content["value"].getcontent():
if isinstance(element["value"], PLCOpenClasses.get("fbdObjects_block", None)) and element["value"].getinstanceName() == name:
return element["value"]
else:
raise TypeError, _("%s body don't have instances!")%self.content["name"]
setattr(cls, "getcontentInstanceByName", getcontentInstanceByName)
def removecontentInstance(self, id):
if self.content["name"] in ["LD","FBD","SFC"]:
i = 0
removed = False
elements = self.content["value"].getcontent()
while i < len(elements) and not removed:
if elements[i]["value"].getlocalId() == id:
self.content["value"].removecontent(i)
removed = True
i += 1
if not removed:
raise ValueError, _("Instance with id %d doesn't exist!")%id
else:
raise TypeError, "%s body don't have instances!"%self.content["name"]
setattr(cls, "removecontentInstance", removecontentInstance)
def settext(self, text):
if self.content["name"] in ["IL","ST"]:
self.content["value"].settext(text)
else:
raise TypeError, _("%s body don't have text!")%self.content["name"]
setattr(cls, "settext", settext)
def gettext(self):
if self.content["name"] in ["IL","ST"]:
return self.content["value"].gettext()
else:
raise TypeError, _("%s body don't have text!")%self.content["name"]
setattr(cls, "gettext", gettext)
def updateElementName(self, old_name, new_name):
if self.content["name"] in ["IL", "ST"]:
self.content["value"].updateElementName(old_name, new_name)
else:
for element in self.content["value"].getcontent():
element["value"].updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
if self.content["name"] in ["IL", "ST"]:
self.content["value"].updateElementAddress(address_model, new_leading)
else:
for element in self.content["value"].getcontent():
element["value"].updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def Search(self, criteria, parent_infos=[]):
if self.content["name"] in ["IL", "ST"]:
search_result = self.content["value"].Search(criteria, parent_infos + ["body", 0])
else:
search_result = []
for element in self.content["value"].getcontent():
search_result.extend(element["value"].Search(criteria, parent_infos))
return search_result
setattr(cls, "Search", Search)
def getx(self):
return self.position.getx()
def gety(self):
return self.position.gety()
def setx(self, x):
self.position.setx(x)
def sety(self, y):
self.position.sety(y)
def _getBoundingBox(self):
return rect(self.getx(), self.gety(), self.getwidth(), self.getheight())
def _getConnectionsBoundingBox(connectionPointIn):
bbox = rect()
connections = connectionPointIn.getconnections()
if connections is not None:
for connection in connections:
for x, y in connection.getpoints():
bbox.update(x, y)
return bbox
def _getBoundingBoxSingle(self):
bbox = _getBoundingBox(self)
if self.connectionPointIn is not None:
bbox.union(_getConnectionsBoundingBox(self.connectionPointIn))
return bbox
def _getBoundingBoxMultiple(self):
bbox = _getBoundingBox(self)
for connectionPointIn in self.getconnectionPointIn():
bbox.union(_getConnectionsBoundingBox(connectionPointIn))
return bbox
def _filterConnections(connectionPointIn, localId, connections):
in_connections = connectionPointIn.getconnections()
if in_connections is not None:
to_delete = []
for i, connection in enumerate(in_connections):
connected = connection.getrefLocalId()
if not connections.has_key((localId, connected)) and \
not connections.has_key((connected, localId)):
to_delete.append(i)
to_delete.reverse()
for i in to_delete:
connectionPointIn.removeconnection(i)
def _filterConnectionsSingle(self, connections):
if self.connectionPointIn is not None:
_filterConnections(self.connectionPointIn, self.localId, connections)
def _filterConnectionsMultiple(self, connections):
for connectionPointIn in self.getconnectionPointIn():
_filterConnections(connectionPointIn, self.localId, connections)
def _getconnectionsdefinition(instance, connections_end):
id = instance.getlocalId()
return dict([((id, end), True) for end in connections_end])
def _updateConnectionsId(connectionPointIn, translation):
connections_end = []
connections = connectionPointIn.getconnections()
if connections is not None:
for connection in connections:
refLocalId = connection.getrefLocalId()
new_reflocalId = translation.get(refLocalId, refLocalId)
connection.setrefLocalId(new_reflocalId)
connections_end.append(new_reflocalId)
return connections_end
def _updateConnectionsIdSingle(self, translation):
connections_end = []
if self.connectionPointIn is not None:
connections_end = _updateConnectionsId(self.connectionPointIn, translation)
return _getconnectionsdefinition(self, connections_end)
def _updateConnectionsIdMultiple(self, translation):
connections_end = []
for connectionPointIn in self.getconnectionPointIn():
connections_end.extend(_updateConnectionsId(connectionPointIn, translation))
return _getconnectionsdefinition(self, connections_end)
def _translate(self, dx, dy):
self.setx(self.getx() + dx)
self.sety(self.gety() + dy)
def _translateConnections(connectionPointIn, dx, dy):
connections = connectionPointIn.getconnections()
if connections is not None:
for connection in connections:
for position in connection.getposition():
position.setx(position.getx() + dx)
position.sety(position.gety() + dy)
def _translateSingle(self, dx, dy):
_translate(self, dx, dy)
if self.connectionPointIn is not None:
_translateConnections(self.connectionPointIn, dx, dy)
def _translateMultiple(self, dx, dy):
_translate(self, dx, dy)
for connectionPointIn in self.getconnectionPointIn():
_translateConnections(connectionPointIn, dx, dy)
def _updateElementName(self, old_name, new_name):
pass
def _updateElementAddress(self, address_model, new_leading):
pass
def _SearchInElement(self, criteria, parent_infos=[]):
return []
_connectionsFunctions = {
"bbox": {"none": _getBoundingBox,
"single": _getBoundingBoxSingle,
"multiple": _getBoundingBoxMultiple},
"translate": {"none": _translate,
"single": _translateSingle,
"multiple": _translateMultiple},
"filter": {"none": lambda self, connections: None,
"single": _filterConnectionsSingle,
"multiple": _filterConnectionsMultiple},
"update": {"none": lambda self, translation: {},
"single": _updateConnectionsIdSingle,
"multiple": _updateConnectionsIdMultiple},
}
def _initElementClass(name, classname, connectionPointInType="none"):
ElementNameToClass[name] = classname
cls = PLCOpenClasses.get(classname, None)
if cls:
setattr(cls, "getx", getx)
setattr(cls, "gety", gety)
setattr(cls, "setx", setx)
setattr(cls, "sety", sety)
setattr(cls, "updateElementName", _updateElementName)
setattr(cls, "updateElementAddress", _updateElementAddress)
setattr(cls, "getBoundingBox", _connectionsFunctions["bbox"][connectionPointInType])
setattr(cls, "translate", _connectionsFunctions["translate"][connectionPointInType])
setattr(cls, "filterConnections", _connectionsFunctions["filter"][connectionPointInType])
setattr(cls, "updateConnectionsId", _connectionsFunctions["update"][connectionPointInType])
setattr(cls, "Search", _SearchInElement)
return cls
def _getexecutionOrder(instance, specific_values):
executionOrder = instance.getexecutionOrderId()
if executionOrder is None:
executionOrder = 0
specific_values["executionOrder"] = executionOrder
def _getdefaultmodifiers(instance, infos):
infos["negated"] = instance.getnegated()
infos["edge"] = instance.getedge()
def _getinputmodifiers(instance, infos):
infos["negated"] = instance.getnegatedIn()
infos["edge"] = instance.getedgeIn()
def _getoutputmodifiers(instance, infos):
infos["negated"] = instance.getnegatedOut()
infos["edge"] = instance.getedgeOut()
MODIFIERS_FUNCTIONS = {"default": _getdefaultmodifiers,
"input": _getinputmodifiers,
"output": _getoutputmodifiers}
def _getconnectioninfos(instance, connection, links=False, modifiers=None, parameter=False):
infos = {"position": connection.getrelPositionXY()}
if parameter:
infos["name"] = instance.getformalParameter()
MODIFIERS_FUNCTIONS.get(modifiers, lambda x, y: None)(instance, infos)
if links:
infos["links"] = []
connections = connection.getconnections()
if connections is not None:
for link in connections:
dic = {"refLocalId": link.getrefLocalId(),
"points": link.getpoints(),
"formalParameter": link.getformalParameter()}
infos["links"].append(dic)
return infos
def _getelementinfos(instance):
return {"id": instance.getlocalId(),
"x": instance.getx(),
"y": instance.gety(),
"height": instance.getheight(),
"width": instance.getwidth(),
"specific_values": {},
"inputs": [],
"outputs": []}
def _getvariableinfosFunction(type, input, output):
def getvariableinfos(self):
infos = _getelementinfos(self)
infos["type"] = type
specific_values = infos["specific_values"]
specific_values["name"] = self.getexpression()
_getexecutionOrder(self, specific_values)
if input and output:
infos["inputs"].append(_getconnectioninfos(self, self.connectionPointIn, True, "input"))
infos["outputs"].append(_getconnectioninfos(self, self.connectionPointOut, False, "output"))
elif input:
infos["inputs"].append(_getconnectioninfos(self, self.connectionPointIn, True, "default"))
elif output:
infos["outputs"].append(_getconnectioninfos(self, self.connectionPointOut, False, "default"))
return infos
return getvariableinfos
def _getconnectorinfosFunction(type):
def getvariableinfos(self):
infos = _getelementinfos(self)
infos["type"] = type
infos["specific_values"]["name"] = self.getname()
if type == "connector":
infos["inputs"].append(_getconnectioninfos(self, self.connectionPointIn, True))
elif type == "continuation":
infos["outputs"].append(_getconnectioninfos(self, self.connectionPointOut))
return infos
return getvariableinfos
def _getpowerrailinfosFunction(type):
def getpowerrailinfos(self):
infos = _getelementinfos(self)
infos["type"] = type
if type == "rightPowerRail":
for connectionPointIn in self.getconnectionPointIn():
infos["inputs"].append(_getconnectioninfos(self, connectionPointIn, True))
infos["specific_values"]["connectors"] = len(infos["inputs"])
elif type == "leftPowerRail":
for connectionPointOut in self.getconnectionPointOut():
infos["outputs"].append(_getconnectioninfos(self, connectionPointOut))
infos["specific_values"]["connectors"] = len(infos["outputs"])
return infos
return getpowerrailinfos
def _getldelementinfosFunction(type):
def getldelementinfos(self):
infos = _getelementinfos(self)
infos["type"] = type
specific_values = infos["specific_values"]
specific_values["name"] = self.getvariable()
_getexecutionOrder(self, specific_values)
specific_values["negated"] = self.getnegated()
specific_values["edge"] = self.getedge()
if type == "coil":
specific_values["storage"] = self.getstorage()
infos["inputs"].append(_getconnectioninfos(self, self.connectionPointIn, True))
infos["outputs"].append(_getconnectioninfos(self, self.connectionPointOut))
return infos
return getldelementinfos
DIVERGENCE_TYPES = {(True, True): "simultaneousDivergence",
(True, False): "selectionDivergence",
(False, True): "simultaneousConvergence",
(False, False): "selectionConvergence"}
def _getdivergenceinfosFunction(divergence, simultaneous):
def getdivergenceinfos(self):
infos = _getelementinfos(self)
infos["type"] = DIVERGENCE_TYPES[(divergence, simultaneous)]
if divergence:
infos["inputs"].append(_getconnectioninfos(self, self.connectionPointIn, True))
for connectionPointOut in self.getconnectionPointOut():
infos["outputs"].append(_getconnectioninfos(self, connectionPointOut))
infos["specific_values"]["connectors"] = len(infos["outputs"])
else:
for connectionPointIn in self.getconnectionPointIn():
infos["inputs"].append(_getconnectioninfos(self, connectionPointIn, True))
infos["outputs"].append(_getconnectioninfos(self, self.connectionPointOut))
infos["specific_values"]["connectors"] = len(infos["inputs"])
return infos
return getdivergenceinfos
cls = _initElementClass("comment", "commonObjects_comment")
if cls:
def getinfos(self):
infos = _getelementinfos(self)
infos["type"] = "comment"
infos["specific_values"]["content"] = self.getcontentText()
return infos
setattr(cls, "getinfos", getinfos)
def setcontentText(self, text):
self.content.settext(text)
setattr(cls, "setcontentText", setcontentText)
def getcontentText(self):
return self.content.gettext()
setattr(cls, "getcontentText", getcontentText)
def updateElementName(self, old_name, new_name):
self.content.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
self.content.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def Search(self, criteria, parent_infos=[]):
return self.content.Search(criteria, parent_infos + ["comment", self.getlocalId(), "content"])
setattr(cls, "Search", Search)
cls = _initElementClass("block", "fbdObjects_block")
if cls:
def getBoundingBox(self):
bbox = _getBoundingBox(self)
for input in self.inputVariables.getvariable():
bbox.union(_getConnectionsBoundingBox(input.connectionPointIn))
return bbox
setattr(cls, "getBoundingBox", getBoundingBox)
def getinfos(self):
infos = _getelementinfos(self)
infos["type"] = self.gettypeName()
specific_values = infos["specific_values"]
specific_values["name"] = self.getinstanceName()
_getexecutionOrder(self, specific_values)
for variable in self.inputVariables.getvariable():
infos["inputs"].append(_getconnectioninfos(variable, variable.connectionPointIn, True, "default", True))
for variable in self.outputVariables.getvariable():
infos["outputs"].append(_getconnectioninfos(variable, variable.connectionPointOut, False, "default", True))
return infos
setattr(cls, "getinfos", getinfos)
def updateElementName(self, old_name, new_name):
if self.typeName == old_name:
self.typeName = new_name
setattr(cls, "updateElementName", updateElementName)
def filterConnections(self, connections):
for input in self.inputVariables.getvariable():
_filterConnections(input.connectionPointIn, self.localId, connections)
setattr(cls, "filterConnections", filterConnections)
def updateConnectionsId(self, translation):
connections_end = []
for input in self.inputVariables.getvariable():
connections_end.extend(_updateConnectionsId(input.connectionPointIn, translation))
return _getconnectionsdefinition(self, connections_end)
setattr(cls, "updateConnectionsId", updateConnectionsId)
def translate(self, dx, dy):
_translate(self, dx, dy)
for input in self.inputVariables.getvariable():
_translateConnections(input.connectionPointIn, dx, dy)
setattr(cls, "translate", translate)
def Search(self, criteria, parent_infos=[]):
parent_infos = parent_infos + ["block", self.getlocalId()]
search_result = _Search([("name", self.getinstanceName()),
("type", self.gettypeName())],
criteria, parent_infos)
for i, variable in enumerate(self.inputVariables.getvariable()):
for result in TestTextElement(variable.getformalParameter(), criteria):
search_result.append((tuple(parent_infos + ["input", i]),) + result)
for i, variable in enumerate(self.outputVariables.getvariable()):
for result in TestTextElement(variable.getformalParameter(), criteria):
search_result.append((tuple(parent_infos + ["output", i]),) + result)
return search_result
setattr(cls, "Search", Search)
cls = _initElementClass("leftPowerRail", "ldObjects_leftPowerRail")
if cls:
setattr(cls, "getinfos", _getpowerrailinfosFunction("leftPowerRail"))
cls = _initElementClass("rightPowerRail", "ldObjects_rightPowerRail", "multiple")
if cls:
setattr(cls, "getinfos", _getpowerrailinfosFunction("rightPowerRail"))
cls = _initElementClass("contact", "ldObjects_contact", "single")
if cls:
setattr(cls, "getinfos", _getldelementinfosFunction("contact"))
def updateElementName(self, old_name, new_name):
if self.variable == old_name:
self.variable = new_name
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
self.variable = update_address(self.variable, address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def Search(self, criteria, parent_infos=[]):
return _Search([("reference", self.getvariable())], criteria, parent_infos + ["contact", self.getlocalId()])
setattr(cls, "Search", Search)
cls = _initElementClass("coil", "ldObjects_coil", "single")
if cls:
setattr(cls, "getinfos", _getldelementinfosFunction("coil"))
def updateElementName(self, old_name, new_name):
if self.variable == old_name:
self.variable = new_name
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
self.variable = update_address(self.variable, address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def Search(self, criteria, parent_infos=[]):
return _Search([("reference", self.getvariable())], criteria, parent_infos + ["coil", self.getlocalId()])
setattr(cls, "Search", Search)
cls = _initElementClass("step", "sfcObjects_step", "single")
if cls:
def getinfos(self):
infos = _getelementinfos(self)
infos["type"] = "step"
specific_values = infos["specific_values"]
specific_values["name"] = self.getname()
specific_values["initial"] = self.getinitialStep()
if self.connectionPointIn:
infos["inputs"].append(_getconnectioninfos(self, self.connectionPointIn, True))
if self.connectionPointOut:
infos["outputs"].append(_getconnectioninfos(self, self.connectionPointOut))
if self.connectionPointOutAction:
specific_values["action"] = _getconnectioninfos(self, self.connectionPointOutAction)
return infos
setattr(cls, "getinfos", getinfos)
def Search(self, criteria, parent_infos=[]):
return _Search([("name", self.getname())], criteria, parent_infos + ["step", self.getlocalId()])
setattr(cls, "Search", Search)
cls = PLCOpenClasses.get("transition_condition", None)
if cls:
def compatibility(self, tree):
connections = []
for child in tree.childNodes:
if child.nodeName == "connection":
connections.append(child)
if len(connections) > 0:
node = CreateNode("connectionPointIn")
relPosition = CreateNode("relPosition")
NodeSetAttr(relPosition, "x", "0")
NodeSetAttr(relPosition, "y", "0")
node.childNodes.append(relPosition)
node.childNodes.extend(connections)
tree.childNodes = [node]
setattr(cls, "compatibility", compatibility)
cls = _initElementClass("transition", "sfcObjects_transition", "single")
if cls:
def getinfos(self):
infos = _getelementinfos(self)
infos["type"] = "transition"
specific_values = infos["specific_values"]
priority = self.getpriority()
if priority is None:
priority = 0
specific_values["priority"] = priority
condition = self.getconditionContent()
specific_values["condition_type"] = condition["type"]
if specific_values["condition_type"] == "connection":
specific_values["connection"] = _getconnectioninfos(self, condition["value"], True)
else:
specific_values["condition"] = condition["value"]
infos["inputs"].append(_getconnectioninfos(self, self.connectionPointIn, True))
infos["outputs"].append(_getconnectioninfos(self, self.connectionPointOut))
return infos
setattr(cls, "getinfos", getinfos)
def setconditionContent(self, type, value):
if not self.condition:
self.addcondition()
if type == "reference":
condition = PLCOpenClasses["condition_reference"]()
condition.setname(value)
elif type == "inline":
condition = PLCOpenClasses["condition_inline"]()
condition.setcontent({"name" : "ST", "value" : PLCOpenClasses["formattedText"]()})
condition.settext(value)
elif type == "connection":
type = "connectionPointIn"
condition = PLCOpenClasses["connectionPointIn"]()
self.condition.setcontent({"name" : type, "value" : condition})
setattr(cls, "setconditionContent", setconditionContent)
def getconditionContent(self):
if self.condition:
content = self.condition.getcontent()
values = {"type" : content["name"]}
if values["type"] == "reference":
values["value"] = content["value"].getname()
elif values["type"] == "inline":
values["value"] = content["value"].gettext()
elif values["type"] == "connectionPointIn":
values["type"] = "connection"
values["value"] = content["value"]
return values
return ""
setattr(cls, "getconditionContent", getconditionContent)
def updateElementName(self, old_name, new_name):
if self.condition:
content = self.condition.getcontent()
if content["name"] == "reference":
if content["value"].getname() == old_name:
content["value"].setname(new_name)
elif content["name"] == "inline":
content["value"].updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
if self.condition:
content = self.condition.getcontent()
if content["name"] == "reference":
content["value"].setname(update_address(content["value"].getname(), address_model, new_leading))
elif content["name"] == "inline":
content["value"].updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def getconnections(self):
if self.condition:
content = self.condition.getcontent()
if content["name"] == "connectionPointIn":
return content["value"].getconnections()
setattr(cls, "getconnections", getconnections)
def Search(self, criteria, parent_infos=[]):
parent_infos = parent_infos + ["transition", self.getlocalId()]
search_result = []
content = self.condition.getcontent()
if content["name"] == "reference":
search_result.extend(_Search([("reference", content["value"].getname())], criteria, parent_infos))
elif content["name"] == "inline":
search_result.extend(content["value"].Search(criteria, parent_infos + ["inline"]))
return search_result
setattr(cls, "Search", Search)
cls = _initElementClass("selectionDivergence", "sfcObjects_selectionDivergence", "single")
if cls:
setattr(cls, "getinfos", _getdivergenceinfosFunction(True, False))
cls = _initElementClass("selectionConvergence", "sfcObjects_selectionConvergence", "multiple")
if cls:
setattr(cls, "getinfos", _getdivergenceinfosFunction(False, False))
cls = _initElementClass("simultaneousDivergence", "sfcObjects_simultaneousDivergence", "single")
if cls:
setattr(cls, "getinfos", _getdivergenceinfosFunction(True, True))
cls = _initElementClass("simultaneousConvergence", "sfcObjects_simultaneousConvergence", "multiple")
if cls:
setattr(cls, "getinfos", _getdivergenceinfosFunction(False, True))
cls = _initElementClass("jumpStep", "sfcObjects_jumpStep", "single")
if cls:
def getinfos(self):
infos = _getelementinfos(self)
infos["type"] = "jump"
infos["specific_values"]["target"] = self.gettargetName()
infos["inputs"].append(_getconnectioninfos(self, self.connectionPointIn, True))
return infos
setattr(cls, "getinfos", getinfos)
def Search(self, criteria, parent_infos):
return _Search([("target", self.gettargetName())], criteria, parent_infos + ["jump", self.getlocalId()])
setattr(cls, "Search", Search)
cls = PLCOpenClasses.get("actionBlock_action", None)
if cls:
def compatibility(self, tree):
relPosition = reduce(lambda x, y: x | (y.nodeName == "relPosition"), tree.childNodes, False)
if not tree.hasAttribute("localId"):
NodeSetAttr(tree, "localId", "0")
if not relPosition:
node = CreateNode("relPosition")
NodeSetAttr(node, "x", "0")
NodeSetAttr(node, "y", "0")
tree.childNodes.insert(0, node)
setattr(cls, "compatibility", compatibility)
def setreferenceName(self, name):
if self.reference:
self.reference.setname(name)
setattr(cls, "setreferenceName", setreferenceName)
def getreferenceName(self):
if self.reference:
return self.reference.getname()
return None
setattr(cls, "getreferenceName", getreferenceName)
def setinlineContent(self, content):
if self.inline:
self.inline.setcontent({"name" : "ST", "value" : PLCOpenClasses["formattedText"]()})
self.inline.settext(content)
setattr(cls, "setinlineContent", setinlineContent)
def getinlineContent(self):
if self.inline:
return self.inline.gettext()
return None
setattr(cls, "getinlineContent", getinlineContent)
def updateElementName(self, old_name, new_name):
if self.reference and self.reference.getname() == old_name:
self.reference.setname(new_name)
if self.inline:
self.inline.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
if self.reference:
self.reference.setname(update_address(self.reference.getname(), address_model, new_leading))
if self.inline:
self.inline.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def Search(self, criteria, parent_infos=[]):
qualifier = self.getqualifier()
if qualifier is None:
qualifier = "N"
return _Search([("inline", self.getinlineContent()),
("reference", self.getreferenceName()),
("qualifier", qualifier),
("duration", self.getduration()),
("indicator", self.getindicator())],
criteria, parent_infos)
setattr(cls, "Search", Search)
cls = _initElementClass("actionBlock", "commonObjects_actionBlock", "single")
if cls:
def compatibility(self, tree):
for child in tree.childNodes[:]:
if child.nodeName == "connectionPointOut":
tree.childNodes.remove(child)
setattr(cls, "compatibility", compatibility)
def getinfos(self):
infos = _getelementinfos(self)
infos["type"] = "actionBlock"
infos["specific_values"]["actions"] = self.getactions()
infos["inputs"].append(_getconnectioninfos(self, self.connectionPointIn, True))
return infos
setattr(cls, "getinfos", getinfos)
def setactions(self, actions):
self.action = []
for params in actions:
action = PLCOpenClasses["actionBlock_action"]()
action.setqualifier(params["qualifier"])
if params["type"] == "reference":
action.addreference()
action.setreferenceName(params["value"])
else:
action.addinline()
action.setinlineContent(params["value"])
if params.has_key("duration"):
action.setduration(params["duration"])
if params.has_key("indicator"):
action.setindicator(params["indicator"])
self.action.append(action)
setattr(cls, "setactions", setactions)
def getactions(self):
actions = []
for action in self.action:
params = {}
params["qualifier"] = action.getqualifier()
if params["qualifier"] is None:
params["qualifier"] = "N"
if action.getreference():
params["type"] = "reference"
params["value"] = action.getreferenceName()
elif action.getinline():
params["type"] = "inline"
params["value"] = action.getinlineContent()
duration = action.getduration()
if duration:
params["duration"] = duration
indicator = action.getindicator()
if indicator:
params["indicator"] = indicator
actions.append(params)
return actions
setattr(cls, "getactions", getactions)
def updateElementName(self, old_name, new_name):
for action in self.action:
action.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
for action in self.action:
action.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def Search(self, criteria, parent_infos=[]):
parent_infos = parent_infos + ["action_block", self.getlocalId()]
search_result = []
for idx, action in enumerate(self.action):
search_result.extend(action.Search(criteria, parent_infos + ["action", idx]))
return search_result
setattr(cls, "Search", Search)
def _SearchInIOVariable(self, criteria, parent_infos=[]):
return _Search([("expression", self.getexpression())], criteria, parent_infos + ["io_variable", self.getlocalId()])
cls = _initElementClass("inVariable", "fbdObjects_inVariable")
if cls:
setattr(cls, "getinfos", _getvariableinfosFunction("input", False, True))
def updateElementName(self, old_name, new_name):
if self.expression == old_name:
self.expression = new_name
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
self.expression = update_address(self.expression, address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
setattr(cls, "Search", _SearchInIOVariable)
cls = _initElementClass("outVariable", "fbdObjects_outVariable", "single")
if cls:
setattr(cls, "getinfos", _getvariableinfosFunction("output", True, False))
def updateElementName(self, old_name, new_name):
if self.expression == old_name:
self.expression = new_name
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
self.expression = update_address(self.expression, address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
setattr(cls, "Search", _SearchInIOVariable)
cls = _initElementClass("inOutVariable", "fbdObjects_inOutVariable", "single")
if cls:
setattr(cls, "getinfos", _getvariableinfosFunction("inout", True, True))
def updateElementName(self, old_name, new_name):
if self.expression == old_name:
self.expression = new_name
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
self.expression = update_address(self.expression, address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
setattr(cls, "Search", _SearchInIOVariable)
def _SearchInConnector(self, criteria, parent_infos=[]):
return _Search([("name", self.getname())], criteria, parent_infos + ["connector", self.getlocalId()])
cls = _initElementClass("continuation", "commonObjects_continuation")
if cls:
setattr(cls, "getinfos", _getconnectorinfosFunction("continuation"))
setattr(cls, "Search", _SearchInConnector)
cls = _initElementClass("connector", "commonObjects_connector", "single")
if cls:
setattr(cls, "getinfos", _getconnectorinfosFunction("connector"))
setattr(cls, "Search", _SearchInConnector)
cls = PLCOpenClasses.get("connection", None)
if cls:
def setpoints(self, points):
self.position = []
for point in points:
position = PLCOpenClasses["position"]()
position.setx(point.x)
position.sety(point.y)
self.position.append(position)
setattr(cls, "setpoints", setpoints)
def getpoints(self):
points = []
for position in self.position:
points.append((position.getx(),position.gety()))
return points
setattr(cls, "getpoints", getpoints)
cls = PLCOpenClasses.get("connectionPointIn", None)
if cls:
def setrelPositionXY(self, x, y):
self.relPosition = PLCOpenClasses["position"]()
self.relPosition.setx(x)
self.relPosition.sety(y)
setattr(cls, "setrelPositionXY", setrelPositionXY)
def getrelPositionXY(self):
if self.relPosition:
return self.relPosition.getx(), self.relPosition.gety()
else:
return self.relPosition
setattr(cls, "getrelPositionXY", getrelPositionXY)
def addconnection(self):
if not self.content:
self.content = {"name" : "connection", "value" : [PLCOpenClasses["connection"]()]}
else:
self.content["value"].append(PLCOpenClasses["connection"]())
setattr(cls, "addconnection", addconnection)
def removeconnection(self, idx):
if self.content:
self.content["value"].pop(idx)
if len(self.content["value"]) == 0:
self.content = None
setattr(cls, "removeconnection", removeconnection)
def removeconnections(self):
if self.content:
self.content = None
setattr(cls, "removeconnections", removeconnections)
def getconnections(self):
if self.content:
return self.content["value"]
setattr(cls, "getconnections", getconnections)
def setconnectionId(self, idx, id):
if self.content:
self.content["value"][idx].setrefLocalId(id)
setattr(cls, "setconnectionId", setconnectionId)
def getconnectionId(self, idx):
if self.content:
return self.content["value"][idx].getrefLocalId()
return None
setattr(cls, "getconnectionId", getconnectionId)
def setconnectionPoints(self, idx, points):
if self.content:
self.content["value"][idx].setpoints(points)
setattr(cls, "setconnectionPoints", setconnectionPoints)
def getconnectionPoints(self, idx):
if self.content:
return self.content["value"][idx].getpoints()
return None
setattr(cls, "getconnectionPoints", getconnectionPoints)
def setconnectionParameter(self, idx, parameter):
if self.content:
self.content["value"][idx].setformalParameter(parameter)
setattr(cls, "setconnectionParameter", setconnectionParameter)
def getconnectionParameter(self, idx):
if self.content:
return self.content["value"][idx].getformalParameter()
return None
setattr(cls, "getconnectionParameter", getconnectionParameter)
cls = PLCOpenClasses.get("connectionPointOut", None)
if cls:
def setrelPositionXY(self, x, y):
self.relPosition = PLCOpenClasses["position"]()
self.relPosition.setx(x)
self.relPosition.sety(y)
setattr(cls, "setrelPositionXY", setrelPositionXY)
def getrelPositionXY(self):
if self.relPosition:
return self.relPosition.getx(), self.relPosition.gety()
return self.relPosition
setattr(cls, "getrelPositionXY", getrelPositionXY)
cls = PLCOpenClasses.get("value", None)
if cls:
def setvalue(self, value):
value = value.strip()
if value.startswith("[") and value.endswith("]"):
arrayValue = PLCOpenClasses["value_arrayValue"]()
self.content = {"name" : "arrayValue", "value" : arrayValue}
elif value.startswith("(") and value.endswith(")"):
structValue = PLCOpenClasses["value_structValue"]()
self.content = {"name" : "structValue", "value" : structValue}
else:
simpleValue = PLCOpenClasses["value_simpleValue"]()
self.content = {"name" : "simpleValue", "value": simpleValue}
self.content["value"].setvalue(value)
setattr(cls, "setvalue", setvalue)
def getvalue(self):
return self.content["value"].getvalue()
setattr(cls, "getvalue", getvalue)
def extractValues(values):
items = values.split(",")
i = 1
while i < len(items):
opened = items[i - 1].count("(") + items[i - 1].count("[")
closed = items[i - 1].count(")") + items[i - 1].count("]")
if opened > closed:
items[i - 1] = ','.join([items[i - 1], items.pop(i)])
elif opened == closed:
i += 1
else:
raise ValueError, _("\"%s\" is an invalid value!")%value
return items
cls = PLCOpenClasses.get("value_arrayValue", None)
if cls:
arrayValue_model = re.compile("([0-9]*)\((.*)\)$")
def setvalue(self, value):
self.value = []
for item in extractValues(value[1:-1]):
item = item.strip()
element = PLCOpenClasses["arrayValue_value"]()
result = arrayValue_model.match(item)
if result is not None:
groups = result.groups()
element.setrepetitionValue(groups[0])
element.setvalue(groups[1].strip())
else:
element.setvalue(item)
self.value.append(element)
setattr(cls, "setvalue", setvalue)
def getvalue(self):
values = []
for element in self.value:
repetition = element.getrepetitionValue()
if repetition is not None and int(repetition) > 1:
value = element.getvalue()
if value is None:
value = ""
values.append("%s(%s)"%(repetition, value))
else:
values.append(element.getvalue())
return "[%s]"%", ".join(values)
setattr(cls, "getvalue", getvalue)
cls = PLCOpenClasses.get("value_structValue", None)
if cls:
structValue_model = re.compile("(.*):=(.*)")
def setvalue(self, value):
self.value = []
for item in extractValues(value[1:-1]):
result = structValue_model.match(item)
if result is not None:
groups = result.groups()
element = PLCOpenClasses["structValue_value"]()
element.setmember(groups[0].strip())
element.setvalue(groups[1].strip())
self.value.append(element)
setattr(cls, "setvalue", setvalue)
def getvalue(self):
values = []
for element in self.value:
values.append("%s := %s"%(element.getmember(), element.getvalue()))
return "(%s)"%", ".join(values)
setattr(cls, "getvalue", getvalue)