beremiz: comparison plcopen/plcopen.py

equal deleted inserted replaced

-:31e63e25b4cc
+:64beb9e9c749
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 from xmlclass import *
 from types import *
-import os, re
+import os
+import re
 from lxml import etree
 from collections import OrderedDict
 import util.paths as paths
 """
 Dictionary that makes the relation between var names in plcopen and displayed values
 """
-VarTypes = {"Local" : "localVars", "Temp" : "tempVars", "Input" : "inputVars",
+VarTypes = {
-"Output" : "outputVars", "InOut" : "inOutVars", "External" : "externalVars",
+"Local":    "localVars",
-"Global" : "globalVars", "Access" : "accessVars"}
+"Temp":     "tempVars",
+"Input":    "inputVars",
+"Output":   "outputVars",
+"InOut":    "inOutVars",
+"External": "externalVars",
+"Global":   "globalVars",
+"Access":   "accessVars"
+}
 searchResultVarTypes = {
-"inputVars": "var_input",
+"inputVars":  "var_input",
 "outputVars": "var_output",
-"inOutVars": "var_inout"
+"inOutVars":  "var_inout"
 }
 """
 Define in which order var types must be displayed
 """
-VarOrder = ["Local","Temp","Input","Output","InOut","External","Global","Access"]
+VarOrder = ["Local", "Temp", "Input", "Output", "InOut", "External", "Global", "Access"]
 """
 Define which action qualifier must be associated with a duration
 """
-QualifierList = OrderedDict([("N", False), ("R", False), ("S", False),
+QualifierList = OrderedDict([
-("L", True), ("D", True), ("P", False), ("P0", False),
+("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:
+"""
+Helper class for bounding_box calculation
+"""
 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 CompilePattern(criteria):
 flag = 0 if criteria["case_sensitive"] else re.IGNORECASE
 find_pattern = criteria["find_pattern"]
 if not criteria["regular_expression"]:
 find_pattern = re.escape(find_pattern)
 criteria["pattern"] = re.compile(find_pattern, flag)
 def TestTextElement(text, criteria):
 lines = text.splitlines()
 test_result = []
 result = criteria["pattern"].search(text)
 while result is not None:
-prev_pos=result.endpos
+prev_pos = result.endpos
 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())
-if result is not None and prev_pos==result.endpos:
+if result is not None and prev_pos == result.endpos:
 break
 return test_result
 def TextMatched(str1, str2):
 return str1 and str2 and (str1.upper() == str2.upper())
 PLCOpenParser = GenerateParserFromXSD(paths.AbsNeighbourFile(__file__, "tc6_xml_v201.xsd"))
-PLCOpen_XPath = lambda xpath: etree.XPath(xpath, namespaces=PLCOpenParser.NSMAP)
+def PLCOpen_XPath(xpath):
+return etree.XPath(xpath, namespaces=PLCOpenParser.NSMAP)
 LOAD_POU_PROJECT_TEMPLATE = """
 <project xmlns:ns1="http://www.plcopen.org/xml/tc6_0201"
 xmlns:xhtml="http://www.w3.org/1999/xhtml"
 xmlns:xsd="http://www.w3.org/2001/XMLSchema"
 xmlns="http://www.plcopen.org/xml/tc6_0201">
 <fileHeader companyName="" productName="" productVersion=""
 creationDateTime="1970-01-01T00:00:00"/>
 <contentHeader name="paste_project">
 <coordinateInfo>
 <fbd><scaling x="0" y="0"/></fbd>
 <ld><scaling x="0" y="0"/></ld>
 <configurations/>
 </instances>
 </project>
 """
 def LOAD_POU_INSTANCES_PROJECT_TEMPLATE(body_type):
 return LOAD_POU_PROJECT_TEMPLATE % """
 <pou name="paste_pou" pouType="program">
 <body>
 <%(body_type)s>%%s</%(body_type)s>
 </body>
 </pou>""" % locals()
 PLCOpen_v1_file = open(paths.AbsNeighbourFile(__file__, "TC6_XML_V10_B.xsd"))
 PLCOpen_v1_xml = PLCOpen_v1_file.read()
 PLCOpen_v1_file.close()
 PLCOpen_v1_xml = PLCOpen_v1_xml.replace(
 "http://www.plcopen.org/xml/tc6.xsd",
 "http://www.plcopen.org/xml/tc6_0201")
 PLCOpen_v1_xsd = etree.XMLSchema(etree.fromstring(PLCOpen_v1_xml))
 # XPath for file compatibility process
 ProjectResourcesXPath = PLCOpen_XPath("ppx:instances/ppx:configurations/ppx:configuration/ppx:resource")
 ResourceInstancesXpath = PLCOpen_XPath("ppx:pouInstance | ppx:task/ppx:pouInstance")
 TransitionsConditionXPath = PLCOpen_XPath("ppx:types/ppx:pous/ppx:pou/ppx:body/*/ppx:transition/ppx:condition")
 ConditionConnectionsXPath = PLCOpen_XPath("ppx:connection")
 ActionBlocksXPath = PLCOpen_XPath("ppx:types/ppx:pous/ppx:pou/ppx:body/*/ppx:actionBlock")
 ActionBlocksConnectionPointOutXPath = PLCOpen_XPath("ppx:connectionPointOut")
 def LoadProjectXML(project_xml):
 project_xml = project_xml.replace(
 "http://www.plcopen.org/xml/tc6.xsd",
 "http://www.plcopen.org/xml/tc6_0201")
 for cre, repl in [
 (re.compile("(?<!<xhtml:p>)(?:<!\[CDATA\[)"), "<xhtml:p><![CDATA["),
 (re.compile("(?:]]>)(?!</xhtml:p>)"), "]]></xhtml:p>")]:
 project_xml = cre.sub(repl, project_xml)
 try:
 tree, error = PLCOpenParser.LoadXMLString(project_xml)
 if error is None:
 return tree, None
 if PLCOpen_v1_xsd.validate(tree):
 # Make file compatible with PLCOpen v2
 # Update resource interval value
 for resource in ProjectResourcesXPath(tree):
 for task in resource.gettask():
 interval = task.get("interval")
 if interval is not None:
 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]
+text += "%dh" % time_values[0]
 if time_values[1] != 0:
-text += "%dm"%time_values[1]
+text += "%dm" % time_values[1]
 if time_values[2] != 0:
-text += "%ds"%time_values[2]
+text += "%ds" % time_values[2]
 if time_values[3] != 0:
 if time_values[3] % 1000 != 0:
-text += "%.3fms"%(float(time_values[3]) / 1000)
+text += "%.3fms" % (float(time_values[3]) / 1000)
 else:
-text += "%dms"%(time_values[3] / 1000)
+text += "%dms" % (time_values[3] / 1000)
 task.set("interval", text)
 # Update resources pou instance attributes
 for pouInstance in ResourceInstancesXpath(resource):
 type_name = pouInstance.attrib.pop("type")
 if type_name is not None:
 pouInstance.set("typeName", type_name)
 # Update transitions condition
 for transition_condition in TransitionsConditionXPath(tree):
 connections = ConditionConnectionsXPath(transition_condition)
 if len(connections) > 0:
 connectionPointIn = PLCOpenParser.CreateElement("connectionPointIn", "condition")
 transition_condition.setcontent(connectionPointIn)
 connectionPointIn.setrelPositionXY(0, 0)
 for connection in connections:
 connectionPointIn.append(connection)
 # Update actionBlocks
 for actionBlock in ActionBlocksXPath(tree):
 for connectionPointOut in ActionBlocksConnectionPointOutXPath(actionBlock):
 actionBlock.remove(connectionPointOut)
 for action in actionBlock.getaction():
 action.set("localId", "0")
 relPosition = PLCOpenParser.CreateElement("relPosition", "action")
 relPosition.set("x", "0")
 relPosition.set("y", "0")
 action.setrelPosition(relPosition)
 return tree, None
 return tree, error
 except Exception, e:
 return None, e.message
 def LoadProject(filepath):
 project_file = open(filepath)
 project_xml = project_file.read()
 project_file.close()
 return LoadProjectXML(project_xml)
 project_pou_xpath = PLCOpen_XPath("/ppx:project/ppx:types/ppx:pous/ppx:pou")
 def LoadPou(xml_string):
 root, error = LoadProjectXML(LOAD_POU_PROJECT_TEMPLATE % xml_string)
 return project_pou_xpath(root)[0], error
 project_pou_instances_xpath = {
 body_type: PLCOpen_XPath(
 "/ppx:project/ppx:types/ppx:pous/ppx:pou[@name='paste_pou']/ppx:body/ppx:%s/*" % body_type)
 for body_type in ["FBD", "LD", "SFC"]}
 def LoadPouInstances(xml_string, body_type):
 root, error = LoadProjectXML(
 LOAD_POU_INSTANCES_PROJECT_TEMPLATE(body_type) % xml_string)
 return project_pou_instances_xpath[body_type](root), error
 def SaveProject(project, filepath):
 project_file = open(filepath, 'w')
 project_file.write(etree.tostring(
 project,
 pretty_print=True,
 xml_declaration=True,
 encoding='utf-8'))
 project_file.close()
 cls = PLCOpenParser.GetElementClass("formattedText")
 if cls:
 def updateElementName(self, old_name, new_name):
 text = self.getanyText()
 pattern = re.compile('\\b' + old_name + '\\b', re.IGNORECASE)
 text = pattern.sub(new_name, text)
 self.setanyText(text)
 setattr(cls, "updateElementName", updateElementName)
 def updateElementAddress(self, address_model, new_leading):
 text = self.getanyText()
 startpos = 0
 result = address_model.search(text, startpos)
 while result is not None:
 text = text[:result.start()] + new_address + text[result.end():]
 startpos = result.start() + len(new_address)
 result = address_model.search(text, startpos)
 self.setanyText(text)
 setattr(cls, "updateElementAddress", updateElementAddress)
 def hasblock(self, block_type):
 text = self.getanyText()
 pattern = re.compile('\\b' + block_type + '\\b', re.IGNORECASE)
 return pattern.search(text) is not None
 setattr(cls, "hasblock", hasblock)
 def Search(self, criteria, parent_infos):
 return [(tuple(parent_infos),) + result for result in TestTextElement(self.getanyText(), criteria)]
 setattr(cls, "Search", Search)
 cls = PLCOpenParser.GetElementClass("project")
 if cls:
 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_obj = self.fileHeader
 return {
 attr: value if value is not None else ""
 for attr, value in [
 ("productRelease", fileheader_obj.getproductRelease()),
 ("creationDateTime", fileheader_obj.getcreationDateTime()),
 ("contentDescription", fileheader_obj.getcontentDescription())]
 }
 setattr(cls, "getfileHeader", getfileHeader)
 def setfileHeader(self, fileheader):
 fileheader_obj = self.fileHeader
 for attr in ["companyName", "companyURL", "productName",
 "productVersion", "productRelease", "creationDateTime",
 "contentDescription"]:
 value = fileheader.get(attr)
 if value is not None:
 setattr(fileheader_obj, attr, value)
 setattr(cls, "setfileHeader", setfileHeader)
 def getcontentHeader(self):
 contentheader_obj = self.contentHeader
 contentheader = {
 attr: value if value is not None else ""
 for attr, value in [
 }
 contentheader["pageSize"] = self.contentHeader.getpageSize()
 contentheader["scaling"] = self.contentHeader.getscaling()
 return contentheader
 setattr(cls, "getcontentHeader", getcontentHeader)
 def setcontentHeader(self, contentheader):
 contentheader_obj = self.contentHeader
 for attr, value in contentheader.iteritems():
 func = {"projectName": contentheader_obj.setname,
 "projectVersion": contentheader_obj.setversion,
 if func is not None:
 func(value)
 elif attr in ["modificationDateTime", "organization", "language"]:
 setattr(contentheader_obj, attr, value)
 setattr(cls, "setcontentHeader", setcontentHeader)
 def gettypeElementFunc(element_type):
 elements_xpath = PLCOpen_XPath(
 "ppx:types/ppx:%(element_type)ss/ppx:%(element_type)s[@name=$name]" % locals())
 def gettypeElement(self, name):
 elements = elements_xpath(self, name=name)
 if len(elements) == 1:
 return elements[0]
 return None
 return gettypeElement
 datatypes_xpath = PLCOpen_XPath("ppx:types/ppx:dataTypes/ppx:dataType")
 filtered_datatypes_xpath = PLCOpen_XPath(
 "ppx:types/ppx:dataTypes/ppx:dataType[@name!=$exclude]")
 def getdataTypes(self, exclude=None):
 if exclude is not None:
 return filtered_datatypes_xpath(self, exclude=exclude)
 return datatypes_xpath(self)
 setattr(cls, "getdataTypes", getdataTypes)
 setattr(cls, "getdataType", gettypeElementFunc("dataType"))
 def appenddataType(self, name):
 if self.getdataType(name) is not None:
-raise ValueError, "\"%s\" Data Type already exists !!!"%name
+raise ValueError("\"%s\" Data Type already exists !!!" % name)
 self.types.appenddataTypeElement(name)
 setattr(cls, "appenddataType", appenddataType)
 def insertdataType(self, index, datatype):
 self.types.insertdataTypeElement(index, datatype)
 setattr(cls, "insertdataType", insertdataType)
 def removedataType(self, name):
 self.types.removedataTypeElement(name)
 setattr(cls, "removedataType", removedataType)
 def getpous(self, exclude=None, filter=[]):
 return self.xpath(
 "ppx:types/ppx:pous/ppx:pou%s%s" %
 (("[@name!='%s']" % exclude) if exclude is not None else '',
 ("[%s]" % " or ".join(
 map(lambda x: "@pouType='%s'" % x, filter)))
 if len(filter) > 0 else ""),
 namespaces=PLCOpenParser.NSMAP)
 setattr(cls, "getpous", getpous)
 setattr(cls, "getpou", gettypeElementFunc("pou"))
 def appendpou(self, name, pou_type, body_type):
 self.types.appendpouElement(name, pou_type, body_type)
 setattr(cls, "appendpou", appendpou)
 def insertpou(self, index, pou):
 self.types.insertpouElement(index, pou)
 setattr(cls, "insertpou", insertpou)
 def removepou(self, name):
 self.types.removepouElement(name)
 setattr(cls, "removepou", removepou)
 configurations_xpath = PLCOpen_XPath(
 "ppx:instances/ppx:configurations/ppx:configuration")
 def getconfigurations(self):
 return configurations_xpath(self)
 setattr(cls, "getconfigurations", getconfigurations)
 configuration_xpath = PLCOpen_XPath(
 "ppx:instances/ppx:configurations/ppx:configuration[@name=$name]")
 def getconfiguration(self, name):
 configurations = configuration_xpath(self, name=name)
 if len(configurations) == 1:
 return configurations[0]
 return None
 setattr(cls, "getconfiguration", getconfiguration)
 def addconfiguration(self, name):
 if self.getconfiguration(name) is not None:
-raise ValueError, _("\"%s\" configuration already exists !!!") % name
+raise ValueError(_("\"%s\" configuration already exists !!!") % name)
 new_configuration = PLCOpenParser.CreateElement("configuration", "configurations")
 new_configuration.setname(name)
 self.instances.configurations.appendconfiguration(new_configuration)
 setattr(cls, "addconfiguration", addconfiguration)
 def removeconfiguration(self, name):
 configuration = self.getconfiguration(name)
 if configuration is None:
-raise ValueError, ("\"%s\" configuration doesn't exist !!!") % name
+raise ValueError(_("\"%s\" configuration doesn't exist !!!") % name)
 self.instances.configurations.remove(configuration)
 setattr(cls, "removeconfiguration", removeconfiguration)
 resources_xpath = PLCOpen_XPath(
 "ppx:instances/ppx:configurations/ppx:configuration[@name=$configname]/ppx:resource[@name=$name]")
 def getconfigurationResource(self, config_name, name):
 resources = resources_xpath(self, configname=config_name, name=name)
 if len(resources) == 1:
 return resources[0]
 return None
 setattr(cls, "getconfigurationResource", getconfigurationResource)
 def addconfigurationResource(self, config_name, name):
 if self.getconfigurationResource(config_name, name) is not None:
-msg = _("\"{a1}\" resource already exists in \"{a2}\" configuration !!!").format(a1 = name, a2 = config_name)
+raise ValueError(
-raise ValueError, msg
+_("\"{a1}\" resource already exists in \"{a2}\" configuration !!!").
+format(a1=name, a2=config_name))
 configuration = self.getconfiguration(config_name)
 if configuration is not None:
 new_resource = PLCOpenParser.CreateElement("resource", "configuration")
 new_resource.setname(name)
 configuration.appendresource(new_resource)
 resource = self.getconfigurationResource(config_name, name)
 if resource is not None:
 configuration.remove(resource)
 found = True
 if not found:
-msg = _("\"{a1}\" resource doesn't exist in \"{a2}\" configuration !!!").format(a1 = name, a2 = config_name)
+raise ValueError(
-raise ValueError, msg
+_("\"{a1}\" resource doesn't exist in \"{a2}\" configuration !!!").
+format(a1=name, a2=config_name))
 setattr(cls, "removeconfigurationResource", removeconfigurationResource)
 def updateElementName(self, old_name, new_name):
 for datatype in self.getdataTypes():
 datatype.updateElementName(old_name, new_name)
 configuration.removeVariableByFilter(address_model)
 setattr(cls, "removeVariableByFilter", removeVariableByFilter)
 enumerated_values_xpath = PLCOpen_XPath(
 "ppx:types/ppx:dataTypes/ppx:dataType/ppx:baseType/ppx:enum/ppx:values/ppx:value")
 def GetEnumeratedDataTypeValues(self):
 return [value.getname() for value in enumerated_values_xpath(self)]
 setattr(cls, "GetEnumeratedDataTypeValues", GetEnumeratedDataTypeValues)
 def Search(self, criteria, parent_infos=[]):
 return result
 setattr(cls, "Search", Search)
 cls = PLCOpenParser.GetElementClass("contentHeader", "project")
 if cls:
 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")
 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)
 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 TextMatched(var_address, old_name):
 var.setaddress(new_name)
 if TextMatched(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.remove(variables[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.remove(variables[i])
 def _SearchInConfigurationResource(self, criteria, parent_infos=[]):
 search_result = _Search([("name", self.getname())], criteria, parent_infos)
 var_number = 0
 for varlist in self.getglobalVars():
 for variable in variables:
 search_result.extend(variable.Search(criteria, parent_infos + [variable_type, var_number]))
 var_number += 1
 return search_result
 cls = PLCOpenParser.GetElementClass("configuration", "configurations")
 if cls:
 def addglobalVar(self, var_type, name, location="", description=""):
 globalvars = self.getglobalVars()
 if len(globalvars) == 0:
 globalvars.append(PLCOpenParser.CreateElement("varList"))
 var = PLCOpenParser.CreateElement("variable", "varListPlain")
 ft = PLCOpenParser.CreateElement("documentation", "variable")
 ft.setanyText(description)
 var.setdocumentation(ft)
 globalvars[-1].appendvariable(var)
 setattr(cls, "addglobalVar", addglobalVar)
 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)
 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 = PLCOpenParser.GetElementClass("resource", "configuration")
 if cls:
 def updateElementName(self, old_name, new_name):
 _updateConfigurationResourceElementName(self, old_name, new_name)
 for instance in self.getpouInstance():
 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)
 if TextMatched(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 = PLCOpenParser.GetElementClass("variable", "varListPlain")
 return vartype_content_name.upper()
 # Variable type is an array
 elif vartype_content_name == "array":
 base_type = vartype_content.baseType.getcontent()
 base_type_name = base_type.getLocalTag()
 # Array derived directly from a user defined type
 if base_type_name == "derived":
 basetype_name = base_type.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.getdimension())), basetype_name)
+return "ARRAY [%s] OF %s" % (",".join(map(lambda x: "%s..%s" % (x.getlower(), x.getupper()), vartype_content.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()),
 ("location", self.getaddress())],
 criteria, parent_infos)
 initial = self.getinitialValue()
 if initial is not None:
 cls = PLCOpenParser.GetElementClass("types", "project")
 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 TextMatched(element.getname(), name):
 return element
 new_datatype = PLCOpenParser.CreateElement("dataType", "dataTypes")
 self.dataTypes.appenddataType(new_datatype)
 new_datatype.setname(name)
 new_datatype.baseType.setcontent(PLCOpenParser.CreateElement("BOOL", "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 element in self.dataTypes.getdataType():
 if TextMatched(element.getname(), name):
 self.dataTypes.remove(element)
 found = True
 break
 if not found:
-raise ValueError, _("\"%s\" Data Type doesn't exist !!!")%name
+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 TextMatched(element.getname(), name):
 return element
 setattr(cls, "getpouElement", getpouElement)
 def appendpouElement(self, name, pou_type, body_type):
 for element in self.pous.getpou():
 if TextMatched(element.getname(), name):
-raise ValueError, _("\"%s\" POU already exists !!!")%name
+raise ValueError(_("\"%s\" POU already exists !!!") % name)
 new_pou = PLCOpenParser.CreateElement("pou", "pous")
 self.pous.appendpou(new_pou)
 new_pou.setname(name)
 new_pou.setpouType(pou_type)
 new_pou.appendbody(PLCOpenParser.CreateElement("body", "pou"))
 new_pou.setbodyType(body_type)
 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 element in self.pous.getpou():
 if TextMatched(element.getname(), name):
 self.pous.remove(element)
 found = True
 break
 if not found:
-raise ValueError, _("\"%s\" POU doesn't exist !!!")%name
+raise ValueError(_("\"%s\" POU doesn't exist !!!") % name)
 setattr(cls, "removepouElement", removepouElement)
 def Search(self, criteria, parent_infos=[]):
 search_result = []
 filter = criteria["filter"]
 for pou in self.pous.getpou():
 search_result.extend(pou.Search(criteria, parent_infos))
 return search_result
 setattr(cls, "Search", Search)
 def _updateBaseTypeElementName(self, old_name, new_name):
 self.baseType.updateElementName(old_name, new_name)
 cls = PLCOpenParser.GetElementClass("dataType", "dataTypes")
 if cls:
 setattr(cls, "updateElementName", _updateBaseTypeElementName)
 def Search(self, criteria, parent_infos=[]):
 search_result = []
 filter = criteria["filter"]
 if filter == "all" or "datatype" in filter:
 parent_infos = parent_infos + ["D::%s" % self.getname()]
 return search_result
 setattr(cls, "Search", Search)
 cls = PLCOpenParser.GetElementClass("dataType")
 if cls:
 def updateElementName(self, old_name, new_name):
 content_name = self.content.getLocalTag()
 if content_name in ["derived", "array", "subrangeSigned", "subrangeUnsigned"]:
 self.content.updateElementName(old_name, new_name)
 elif content_name == "struct":
 if cls:
 def updateElementName(self, old_name, new_name):
 if TextMatched(self.name, old_name):
 self.name = new_name
 setattr(cls, "updateElementName", updateElementName)
 def Search(self, criteria, parent_infos=[]):
 return [(tuple(parent_infos),) + result for result in TestTextElement(self.name, criteria)]
 setattr(cls, "Search", Search)
 cls = PLCOpenParser.GetElementClass("array", "dataType")
 if cls:
 setattr(cls, "updateElementName", _updateBaseTypeElementName)
 def Search(self, criteria, parent_infos=[]):
 search_result = self.baseType.Search(criteria, parent_infos)
 for i, dimension in enumerate(self.getdimension()):
 search_result.extend(_Search([("lower", dimension.getlower()),
 ("upper", dimension.getupper())],
 criteria, parent_infos + ["range", i]))
 return search_result
 setattr(cls, "Search", Search)
 def _SearchInSubrange(self, criteria, parent_infos=[]):
 search_result = self.baseType.Search(criteria, parent_infos)
 search_result.extend(_Search([("lower", self.range.getlower()),
 ("upper", self.range.getupper())],
 criteria, parent_infos))
 return search_result
 cls = PLCOpenParser.GetElementClass("subrangeSigned", "dataType")
 if cls:
 setattr(cls, "updateElementName", _updateBaseTypeElementName)
 setattr(cls, "Search", _SearchInSubrange)
 setattr(cls, "updateElementName", _updateBaseTypeElementName)
 setattr(cls, "Search", _SearchInSubrange)
 cls = PLCOpenParser.GetElementClass("enum", "dataType")
 if cls:
 def updateElementName(self, old_name, new_name):
 pass
 setattr(cls, "updateElementName", updateElementName)
 enumerated_datatype_values_xpath = PLCOpen_XPath("ppx:values/ppx:value")
 def Search(self, criteria, parent_infos=[]):
 search_result = []
 for i, value in enumerate(enumerated_datatype_values_xpath(self)):
 for result in TestTextElement(value.getname(), criteria):
 search_result.append((tuple(parent_infos + ["value", i]),) + result)
 return search_result
 setattr(cls, "Search", Search)
 def _getvariableTypeinfos(variable_type):
 type_content = variable_type.getcontent()
 type_content_type = type_content.getLocalTag()
 if type_content_type == "derived":
 return type_content.getname()
 return type_content_type.upper()
 cls = PLCOpenParser.GetElementClass("pou", "pous")
 if cls:
 block_inputs_xpath = PLCOpen_XPath(
 "ppx:interface/*[self::ppx:inputVars or self::ppx:inOutVars]/ppx:variable")
 block_outputs_xpath = PLCOpen_XPath(
 "ppx:interface/*[self::ppx:outputVars or self::ppx:inOutVars]/ppx:variable")
-def getblockInfos(self):
+def getblockInfos(self):
 block_infos = {
-"name" : self.getname(),
+"name": self.getname(),
-"type" : self.getpouType(),
+"type": self.getpouType(),
-"extensible" : False,
+"extensible": False,
-"inputs" : [],
+"inputs": [],
-"outputs" : [],
+"outputs": [],
-"comment" : self.getdescription()}
+"comment": self.getdescription()}
 if self.interface is not None:
 return_type = self.interface.getreturnType()
 if return_type is not None:
 block_infos["outputs"].append(
 ("OUT", _getvariableTypeinfos(return_type), "none"))
 [(var.getname(), _getvariableTypeinfos(var.type), "none")
 for var in block_inputs_xpath(self)])
 block_infos["outputs"].extend(
 [(var.getname(), _getvariableTypeinfos(var.type), "none")
 for var in block_outputs_xpath(self)])
 block_infos["usage"] = ("\n (%s) => (%s)" %
 (", ".join(["%s:%s" % (input[1], input[0])
 for input in block_infos["inputs"]]),
 ", ".join(["%s:%s" % (output[1], output[0])
 for output in block_infos["outputs"]])))
 return block_infos
 setattr(cls, "getblockInfos", getblockInfos)
 def setdescription(self, description):
 doc = self.getdocumentation()
 if doc is None:
 doc = PLCOpenParser.CreateElement("documentation", "pou")
 self.setdocumentation(doc)
 doc.setanyText(description)
 setattr(cls, "setdescription", setdescription)
 def getdescription(self):
 doc = self.getdocumentation()
 if doc is not None:
 return doc.getanyText()
 return ""
 setattr(cls, "getdescription", getdescription)
 def setbodyType(self, body_type):
 if len(self.body) > 0:
 if body_type in ["IL", "ST", "LD", "FBD", "SFC"]:
 self.body[0].setcontent(PLCOpenParser.CreateElement(body_type, "body"))
 else:
-raise ValueError, "%s isn't a valid body type!"%type
+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().getLocalTag()
 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, instance):
 if len(self.body) > 0:
 self.body[0].appendcontentInstance(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 getinstancesIds(self):
 if len(self.body) > 0:
 return self.body[0].getcontentInstancesIds()
 return []
 setattr(cls, "getinstancesIds", getinstancesIds)
 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)
 reverse_types[value] = name
 for varlist in self.interface.getcontent():
 vars.append((reverse_types[varlist.getLocalTag()], varlist))
 return vars
 setattr(cls, "getvars", getvars)
 def setvars(self, vars):
 if self.interface is None:
 self.interface = PLCOpenParser.CreateElement("interface", "pou")
 self.interface.setcontent(vars)
 setattr(cls, "setvars", setvars)
 def addpouExternalVar(self, var_type, name):
 self.addpouVar(var_type, name, "externalVars")
 setattr(cls, "addpouExternalVar", addpouExternalVar)
 def addpouVar(self, var_type, name, var_class="localVars", location="", description="", initval=""):
 if self.interface is None:
 self.interface = PLCOpenParser.CreateElement("interface", "pou")
 content = self.interface.getcontent()
 if len(content) == 0:
 var.setdocumentation(ft)
 if initval != "":
 el = PLCOpenParser.CreateElement("initialValue", "variable")
 el.setvalue(initval)
 var.setinitialValue(el)
 varlist.appendvariable(var)
 setattr(cls, "addpouVar", addpouVar)
 setattr(cls, "addpouLocalVar", 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.getvariable()
 if vartype_content.getLocalTag() == "derived" and TextMatched(vartype_content.getname(), old_type):
 var.setname(new_name)
 vartype_content.setname(new_type)
 return
 setattr(cls, "changepouVar", changepouVar)
 def removepouVar(self, var_type, name):
 if self.interface is not None:
 content = self.interface.getcontent()
 for varlist in content:
 for var in varlist.getvariable():
 def hasstep(self, name=None):
 if self.getbodyType() in ["SFC"]:
 for instance in self.getinstances():
 if isinstance(instance, PLCOpenParser.GetElementClass("step", "sfcObjects")) and TextMatched(instance.getname(), name):
 return True
 return False
 setattr(cls, "hasstep", hasstep)
 def hasblock(self, name=None, block_type=None):
 if self.getbodyType() in ["FBD", "LD", "SFC"]:
 for instance in self.getinstances():
-if (isinstance(instance, PLCOpenParser.GetElementClass("block", "fbdObjects")) and
+if isinstance(instance, PLCOpenParser.GetElementClass("block", "fbdObjects")) \
-(TextMatched(instance.getinstanceName(), name) or TextMatched(instance.gettypeName(), block_type))):
+and (TextMatched(instance.getinstanceName(), name) or
+TextMatched(instance.gettypeName(), block_type)):
 return True
 if self.transitions:
 for transition in self.transitions.gettransition():
 result = transition.hasblock(name, block_type)
 if result:
 return result
 elif block_type is not None and len(self.body) > 0:
 return self.body[0].hasblock(block_type)
 return False
 setattr(cls, "hasblock", hasblock)
 def addtransition(self, name, body_type):
 if self.transitions is None:
 self.addtransitions()
 self.transitions.settransition([])
 transition = PLCOpenParser.CreateElement("transition", "transitions")
 transition.setname(name)
 transition.setbodyType(body_type)
 if body_type == "ST":
 transition.settext(":= ;")
 setattr(cls, "addtransition", addtransition)
 def gettransition(self, name):
 if self.transitions is not None:
 for transition in self.transitions.gettransition():
 if TextMatched(transition.getname(), name):
 return transition
 return None
 setattr(cls, "gettransition", gettransition)
 def gettransitionList(self):
 if self.transitions is not None:
 return self.transitions.gettransition()
 return []
 setattr(cls, "gettransitionList", gettransitionList)
 def removetransition(self, name):
 if self.transitions is not None:
 removed = False
 for transition in self.transitions.gettransition():
 if TextMatched(transition.getname(), name):
 if transition.getbodyType() in ["FBD", "LD", "SFC"]:
 for instance in transition.getinstances():
 if isinstance(instance, PLCOpenParser.GetElementClass("block", "fbdObjects")):
 self.removepouVar(instance.gettypeName(),
 instance.getinstanceName())
 self.transitions.remove(transition)
 removed = True
 break
 if not removed:
-raise ValueError, _("Transition with name %s doesn't exist!")%name
+raise ValueError(_("Transition with name %s doesn't exist!") % name)
 setattr(cls, "removetransition", removetransition)
 def addaction(self, name, body_type):
 if self.actions is None:
 self.addactions()
 action = PLCOpenParser.CreateElement("action", "actions")
 self.actions.appendaction(action)
 action.setname(name)
 action.setbodyType(body_type)
 setattr(cls, "addaction", addaction)
 def getaction(self, name):
 if self.actions is not None:
 for action in self.actions.getaction():
 if TextMatched(action.getname(), name):
 return action
 return None
 setattr(cls, "getaction", getaction)
 def getactionList(self):
 if self.actions is not None:
 return self.actions.getaction()
 return []
 setattr(cls, "getactionList", getactionList)
 def removeaction(self, name):
 if self.actions is not None:
 removed = False
 for action in self.actions.getaction():
 if TextMatched(action.getname(), name):
 if action.getbodyType() in ["FBD", "LD", "SFC"]:
 for instance in action.getinstances():
 if isinstance(instance, PLCOpenParser.GetElementClass("block", "fbdObjects")):
 self.removepouVar(instance.gettypeName(),
 instance.getinstanceName())
 self.actions.remove(action)
 removed = True
 break
 if not removed:
-raise ValueError, _("Action with name %s doesn't exist!")%name
+raise ValueError(_("Action with name %s doesn't exist!") % name)
 setattr(cls, "removeaction", removeaction)
 def updateElementName(self, old_name, new_name):
 if self.interface is not None:
 for content in self.interface.getcontent():
 if var_address is not None:
 result = address_model.match(var_address)
 if result is not None:
 content.remove(variable)
 setattr(cls, "removeVariableByFilter", removeVariableByFilter)
 def Search(self, criteria, parent_infos=[]):
 search_result = []
 filter = criteria["filter"]
 if filter == "all" or self.getpouType() in filter:
 if 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, body_type):
 if body_type in ["IL", "ST", "LD", "FBD", "SFC"]:
 self.body.setcontent(PLCOpenParser.CreateElement(body_type, "body"))
 else:
-raise ValueError, "%s isn't a valid body type!"%type
+raise ValueError("%s isn't a valid body type!" % type)
 def getbodyType(self):
 return self.body.getcontent().getLocalTag()
 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, instance):
 self.body.appendcontentInstance(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()
 def hasblock(self, name=None, block_type=None):
 if self.getbodyType() in ["FBD", "LD", "SFC"]:
 for instance in self.getinstances():
-if (isinstance(instance, PLCOpenParser.GetElementClass("block", "fbdObjects")) and
+if isinstance(instance, PLCOpenParser.GetElementClass("block", "fbdObjects")) and \
-(TextMatched(instance.getinstanceName(), name) or TextMatched(instance.gettypeName(), block_type))):
+(TextMatched(instance.getinstanceName(), name) or TextMatched(instance.gettypeName(), block_type)):
 return True
 elif block_type is not None:
 return self.body.hasblock(block_type)
 return False
 def updateElementName(self, old_name, new_name):
 self.body.updateElementName(old_name, new_name)
 def updateElementAddress(self, address_model, new_leading):
 self.body.updateElementAddress(address_model, new_leading)
 cls = PLCOpenParser.GetElementClass("transition", "transitions")
 if cls:
 setattr(cls, "setbodyType", setbodyType)
 setattr(cls, "getbodyType", getbodyType)
 setattr(cls, "settext", settext)
 setattr(cls, "gettext", gettext)
 setattr(cls, "hasblock", hasblock)
 setattr(cls, "updateElementName", updateElementName)
 setattr(cls, "updateElementAddress", updateElementAddress)
 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)
 setattr(cls, "settext", settext)
 setattr(cls, "gettext", gettext)
 setattr(cls, "hasblock", hasblock)
 setattr(cls, "updateElementName", updateElementName)
 setattr(cls, "updateElementAddress", updateElementAddress)
 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)
 cls = PLCOpenParser.GetElementClass("body")
 if cls:
 cls.currentExecutionOrderId = 0
 cls.checkedBlocksDict = {}
 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.getLocalTag() == "FBD":
 for element in self.content.getcontent():
 if not isinstance(element, (PLCOpenParser.GetElementClass("comment", "commonObjects"),
 PLCOpenParser.GetElementClass("connector", "commonObjects"),
 PLCOpenParser.GetElementClass("continuation", "commonObjects"))):
 element.setexecutionOrderId(0)
 self.checkedBlocksDict.clear()
 else:
-raise TypeError, _("Can only generate execution order on FBD networks!")
+raise TypeError(_("Can only generate execution order on FBD networks!"))
 setattr(cls, "resetexecutionOrder", resetexecutionOrder)
 def compileexecutionOrder(self):
 if self.content.getLocalTag() == "FBD":
 self.resetexecutionOrder()
 self.resetcurrentExecutionOrderId()
 for element in self.content.getcontent():
 connections = element.connectionPointIn.getconnections()
 if connections and len(connections) == 1:
 self.compileelementExecutionOrder(connections[0])
 element.setexecutionOrderId(self.getnewExecutionOrderId())
 else:
-raise TypeError, _("Can only generate execution order on FBD networks!")
+raise TypeError(_("Can only generate execution order on FBD networks!"))
 setattr(cls, "compileexecutionOrder", compileexecutionOrder)
 def compileelementExecutionOrder(self, link):
 if self.content.getLocalTag() == "FBD":
 localid = link.getrefLocalId()
 instance = self.getcontentInstance(localid)
 self.checkedBlocksDict[localid] = True
 if isinstance(instance, PLCOpenParser.GetElementClass("block", "fbdObjects")) and instance.getexecutionOrderId() == 0:
 for variable in instance.inputVariables.getvariable():
 connections = variable.connectionPointIn.getconnections()
 if connections and len(connections) == 1:
-if (self.checkedBlocksDict.has_key(connections[0].getrefLocalId()) == False):
+if not connections[0].getrefLocalId() in self.checkedBlocksDict:
 self.compileelementExecutionOrder(connections[0])
 if instance.getexecutionOrderId() == 0:
 instance.setexecutionOrderId(self.getnewExecutionOrderId())
 elif isinstance(instance, PLCOpenParser.GetElementClass("continuation", "commonObjects")) and instance.getexecutionOrderId() == 0:
 for tmp_instance in self.getcontentInstances():
-if (isinstance(tmp_instance, PLCOpenParser.GetElementClass("connector", "commonObjects")) and
+if isinstance(tmp_instance, PLCOpenParser.GetElementClass("connector", "commonObjects")) and \
-TextMatched(tmp_instance.getname(), instance.getname()) and tmp_instance.getexecutionOrderId() == 0):
+TextMatched(tmp_instance.getname(), instance.getname()) 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!")
+raise TypeError(_("Can only generate execution order on FBD networks!"))
 setattr(cls, "compileelementExecutionOrder", compileelementExecutionOrder)
 def setelementExecutionOrder(self, instance, new_executionOrder):
 if self.content.getLocalTag() == "FBD":
 old_executionOrder = instance.getexecutionOrderId()
 if old_executionOrder is not None and old_executionOrder != 0 and new_executionOrder != 0:
 for element in self.content.getcontent():
 element.setexecutionOrderId(element_executionOrder - 1)
 if new_executionOrder <= element_executionOrder <= old_executionOrder:
 element.setexecutionOrderId(element_executionOrder + 1)
 instance.setexecutionOrderId(new_executionOrder)
 else:
-raise TypeError, _("Can only generate execution order on FBD networks!")
+raise TypeError(_("Can only generate execution order on FBD networks!"))
 setattr(cls, "setelementExecutionOrder", setelementExecutionOrder)
 def appendcontentInstance(self, instance):
-if self.content.getLocalTag() in ["LD","FBD","SFC"]:
+if self.content.getLocalTag() in ["LD", "FBD", "SFC"]:
 self.content.appendcontent(instance)
 else:
-raise TypeError, _("%s body don't have instances!")%self.content.getLocalTag()
+raise TypeError(_("%s body don't have instances!") % self.content.getLocalTag())
 setattr(cls, "appendcontentInstance", appendcontentInstance)
 def getcontentInstances(self):
-if self.content.getLocalTag() in ["LD","FBD","SFC"]:
+if self.content.getLocalTag() in ["LD", "FBD", "SFC"]:
 return self.content.getcontent()
 else:
-raise TypeError, _("%s body don't have instances!")%self.content.getLocalTag()
+raise TypeError(_("%s body don't have instances!") % self.content.getLocalTag())
 setattr(cls, "getcontentInstances", getcontentInstances)
 instance_by_id_xpath = PLCOpen_XPath("*[@localId=$localId]")
 instance_by_name_xpath = PLCOpen_XPath("ppx:block[@instanceName=$name]")
 def getcontentInstance(self, local_id):
-if self.content.getLocalTag() in ["LD","FBD","SFC"]:
+if self.content.getLocalTag() in ["LD", "FBD", "SFC"]:
 instance = instance_by_id_xpath(self.content, localId=local_id)
 if len(instance) > 0:
 return instance[0]
 return None
 else:
-raise TypeError, _("%s body don't have instances!")%self.content.getLocalTag()
+raise TypeError(_("%s body don't have instances!") % self.content.getLocalTag())
 setattr(cls, "getcontentInstance", getcontentInstance)
 def getcontentInstancesIds(self):
-if self.content.getLocalTag() in ["LD","FBD","SFC"]:
+if self.content.getLocalTag() in ["LD", "FBD", "SFC"]:
 return OrderedDict([(instance.getlocalId(), True)
 for instance in self.content])
 else:
-raise TypeError, _("%s body don't have instances!")%self.content.getLocalTag()
+raise TypeError(_("%s body don't have instances!") % self.content.getLocalTag())
 setattr(cls, "getcontentInstancesIds", getcontentInstancesIds)
 def getcontentInstanceByName(self, name):
-if self.content.getLocalTag() in ["LD","FBD","SFC"]:
+if self.content.getLocalTag() in ["LD", "FBD", "SFC"]:
 instance = instance_by_name_xpath(self.content)
 if len(instance) > 0:
 return instance[0]
 return None
 else:
-raise TypeError, _("%s body don't have instances!")%self.content.getLocalTag()
+raise TypeError(_("%s body don't have instances!") % self.content.getLocalTag())
 setattr(cls, "getcontentInstanceByName", getcontentInstanceByName)
 def removecontentInstance(self, local_id):
-if self.content.getLocalTag() in ["LD","FBD","SFC"]:
+if self.content.getLocalTag() in ["LD", "FBD", "SFC"]:
 instance = instance_by_id_xpath(self.content, localId=local_id)
 if len(instance) > 0:
 self.content.remove(instance[0])
 else:
-raise ValueError, _("Instance with id %d doesn't exist!")%id
+raise ValueError(_("Instance with id %d doesn't exist!") % id)
 else:
-raise TypeError, "%s body don't have instances!"%self.content.getLocalTag()
+raise TypeError(_("%s body don't have instances!") % self.content.getLocalTag())
 setattr(cls, "removecontentInstance", removecontentInstance)
 def settext(self, text):
-if self.content.getLocalTag() in ["IL","ST"]:
+if self.content.getLocalTag() in ["IL", "ST"]:
 self.content.setanyText(text)
 else:
-raise TypeError, _("%s body don't have text!")%self.content.getLocalTag()
+raise TypeError(_("%s body don't have text!") % self.content.getLocalTag())
 setattr(cls, "settext", settext)
 def gettext(self):
-if self.content.getLocalTag() in ["IL","ST"]:
+if self.content.getLocalTag() in ["IL", "ST"]:
 return self.content.getanyText()
 else:
-raise TypeError, _("%s body don't have text!")%self.content.getLocalTag()
+raise TypeError(_("%s body don't have text!") % self.content.getLocalTag())
 setattr(cls, "gettext", gettext)
 def hasblock(self, block_type):
-if self.content.getLocalTag() in ["IL","ST"]:
+if self.content.getLocalTag() in ["IL", "ST"]:
 return self.content.hasblock(block_type)
 else:
-raise TypeError, _("%s body don't have text!")%self.content.getLocalTag()
+raise TypeError(_("%s body don't have text!") % self.content.getLocalTag())
 setattr(cls, "hasblock", hasblock)
 def updateElementName(self, old_name, new_name):
 if self.content.getLocalTag() in ["IL", "ST"]:
 self.content.updateElementName(old_name, new_name)
 else:
 for element in self.content.getcontent():
 for element in self.content.getcontent():
 search_result.extend(element.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:
 for connection in in_connections:
 connected = connection.getrefLocalId()
-if not connections.has_key((localId, connected)) and \
+if not (localId, connected) in connections and \
-not connections.has_key((connected, localId)):
+not (connected, localId) in connections:
 connectionPointIn.remove(connection)
 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):
 local_id = instance.getlocalId()
 return dict([((local_id, end), True) for end in connections_end])
 def _updateConnectionsId(connectionPointIn, translation):
 connections_end = []
 connections = connectionPointIn.getconnections()
 if connections is not None:
 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, parent, connectionPointInType="none"):
 cls = PLCOpenParser.GetElementClass(name, parent)
 if cls:
 setattr(cls, "getx", getx)
 setattr(cls, "filterConnections", _connectionsFunctions["filter"][connectionPointInType])
 setattr(cls, "updateConnectionsId", _connectionsFunctions["update"][connectionPointInType])
 setattr(cls, "Search", _SearchInElement)
 return cls
 cls = _initElementClass("comment", "commonObjects")
 if cls:
 def setcontentText(self, text):
 self.content.setanyText(text)
 setattr(cls, "setcontentText", setcontentText)
 def getcontentText(self):
 return self.content.getanyText()
 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):
 setattr(cls, "Search", Search)
 _initElementClass("leftPowerRail", "ldObjects")
 _initElementClass("rightPowerRail", "ldObjects", "multiple")
 def _UpdateLDElementName(self, old_name, new_name):
 if TextMatched(self.variable, old_name):
 self.variable = new_name
 def _UpdateLDElementAddress(self, address_model, new_leading):
 self.variable = update_address(self.variable, address_model, new_leading)
 def _getSearchInLDElement(ld_element_type):
 def SearchInLDElement(self, criteria, parent_infos=[]):
 return _Search([("reference", self.variable)], criteria, parent_infos + [ld_element_type, self.getlocalId()])
 return SearchInLDElement
 cls = _initElementClass("contact", "ldObjects", "single")
 if cls:
 setattr(cls, "updateElementName", _UpdateLDElementName)
 setattr(cls, "updateElementAddress", _UpdateLDElementAddress)
 condition.setname(value)
 elif condition_type == "inline":
 condition.setcontent(PLCOpenParser.CreateElement("ST", "inline"))
 condition.settext(value)
 setattr(cls, "setconditionContent", setconditionContent)
 def getconditionContent(self):
 if self.condition is not None:
 content = self.condition.getcontent()
-values = {"type" : content.getLocalTag()}
+values = {"type": content.getLocalTag()}
 if values["type"] == "reference":
 values["value"] = content.getname()
 elif values["type"] == "inline":
 values["value"] = content.gettext()
 elif values["type"] == "connectionPointIn":
 condition_connection = self.getconditionConnection()
 if condition_connection is not None:
 bbox.union(_getConnectionsBoundingBox(condition_connection))
 return bbox
 setattr(cls, "getBoundingBox", getBoundingBox)
 def translate(self, dx, dy):
 _translateSingle(self, dx, dy)
 condition_connection = self.getconditionConnection()
 if condition_connection is not None:
 _translateConnections(condition_connection, dx, dy)
 setattr(cls, "translate", translate)
 def filterConnections(self, connections):
 _filterConnectionsSingle(self, connections)
 condition_connection = self.getconditionConnection()
 if condition_connection is not None:
 _filterConnections(condition_connection, self.localId, connections)
 setattr(cls, "filterConnections", filterConnections)
 def updateConnectionsId(self, translation):
 connections_end = []
 if self.connectionPointIn is not None:
 connections_end = _updateConnectionsId(self.connectionPointIn, translation)
 condition_connection = self.getconditionConnection()
 condition_connection = self.getconditionConnection()
 if condition_connection is not None:
 return condition_connection.getconnections()
 return None
 setattr(cls, "getconnections", getconnections)
 def Search(self, criteria, parent_infos=[]):
 parent_infos = parent_infos + ["transition", self.getlocalId()]
 search_result = []
 content = self.condition.getcontent()
 content_name = content.getLocalTag()
 _initElementClass("selectionDivergence", "sfcObjects", "single")
 _initElementClass("selectionConvergence", "sfcObjects", "multiple")
 _initElementClass("simultaneousDivergence", "sfcObjects", "single")
 _initElementClass("simultaneousConvergence", "sfcObjects", "multiple")
 cls = _initElementClass("jumpStep", "sfcObjects", "single")
 if cls:
 def Search(self, criteria, parent_infos):
 return _Search([("target", self.gettargetName())], criteria, parent_infos + ["jump", self.getlocalId()])
 setattr(cls, "Search", Search)
 if cls:
 def setreferenceName(self, name):
 if self.reference is not None:
 self.reference.setname(name)
 setattr(cls, "setreferenceName", setreferenceName)
 def getreferenceName(self):
 if self.reference is not None:
 return self.reference.getname()
 return None
 setattr(cls, "getreferenceName", getreferenceName)
 def setinlineContent(self, content):
 if self.inline is not None:
 self.inline.setcontent(PLCOpenParser.CreateElement("ST", "inline"))
 self.inline.settext(content)
 setattr(cls, "setinlineContent", setinlineContent)
 def getinlineContent(self):
 if self.inline is not None:
 return self.inline.gettext()
 return None
 setattr(cls, "getinlineContent", getinlineContent)
 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)
 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.expression)], criteria, parent_infos + ["io_variable", self.getlocalId()])
 def _UpdateIOElementName(self, old_name, new_name):
 if TextMatched(self.expression, old_name):
 self.expression = new_name
 def _UpdateIOElementAddress(self, address_model, new_leading):
 self.expression = update_address(self.expression, address_model, new_leading)
 cls = _initElementClass("inVariable", "fbdObjects")
 if cls:
 setattr(cls, "updateElementName", _UpdateIOElementName)
 setattr(cls, "updateElementAddress", _UpdateIOElementAddress)
 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")
 if cls:
 setattr(cls, "Search", _SearchInConnector)
 setattr(cls, "setpoints", setpoints)
 def getpoints(self):
 points = []
 for position in self.position:
-points.append((position.getx(),position.gety()))
+points.append((position.getx(), position.gety()))
 return points
 setattr(cls, "getpoints", getpoints)
 cls = PLCOpenParser.GetElementClass("connectionPointIn")
 if cls:
 setattr(cls, "removeconnection", removeconnection)
 def removeconnections(self):
 self.content = None
 setattr(cls, "removeconnections", removeconnections)
 connection_xpath = PLCOpen_XPath("ppx:connection")
 connection_by_position_xpath = PLCOpen_XPath("ppx:connection[position()=$pos]")
 def getconnections(self):
 return connection_xpath(self)
 setattr(cls, "getconnections", getconnections)
 def getconnection(self, idx):
 connection = connection_by_position_xpath(self, pos=idx+1)
 if len(connection) > 0:
 return connection[0]
 return None
 setattr(cls, "getconnection", getconnection)
 def setconnectionId(self, idx, local_id):
 connection = self.getconnection(idx)
 if connection is not None:
 connection.setrefLocalId(local_id)
 setattr(cls, "setconnectionId", setconnectionId)
 def getconnectionId(self, idx):
 connection = self.getconnection(idx)
 if connection is not None:
 return connection.getrefLocalId()
 return None
 setattr(cls, "getconnectionId", getconnectionId)
 def setconnectionPoints(self, idx, points):
 connection = self.getconnection(idx)
 if connection is not None:
 connection.setpoints(points)
 setattr(cls, "setconnectionPoints", setconnectionPoints)
 def setconnectionParameter(self, idx, parameter):
 connection = self.getconnection(idx)
 if connection is not None:
 connection.setformalParameter(parameter)
 setattr(cls, "setconnectionParameter", setconnectionParameter)
 def getconnectionParameter(self, idx):
 connection = self.getconnection(idx)
 if connection is not None:
 return connection.getformalParameter()
 return None
 else:
 content = PLCOpenParser.CreateElement("simpleValue", "value")
 content.setvalue(value)
 self.setcontent(content)
 setattr(cls, "setvalue", setvalue)
 def getvalue(self):
 return self.content.getvalue()
 setattr(cls, "getvalue", getvalue)
 def extractValues(values):
 items = values.split(",")
 i = 1
 while i < len(items):
 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
+raise ValueError(_("\"%s\" is an invalid value!") % value)
 return items
 cls = PLCOpenParser.GetElementClass("arrayValue", "value")
 if cls:
 arrayValue_model = re.compile("([0-9]+)\((.*)\)$")
 def setvalue(self, value):
 elements = []
 for item in extractValues(value[1:-1]):
 item = item.strip()
 element = PLCOpenParser.CreateElement("value", "arrayValue")
 else:
 element.setvalue(item)
 elements.append(element)
 self.value = elements
 setattr(cls, "setvalue", setvalue)
 def getvalue(self):
 values = []
 for element in self.value:
 try:
 repetition = int(element.getrepetitionValue())
-except:
+except Exception:
 repetition = 1
 if repetition > 1:
 value = element.getvalue()
 if value is None:
 value = ""
-values.append("%s(%s)"%(repetition, value))
+values.append("%s(%s)" % (repetition, value))
 else:
 values.append(element.getvalue())
-return "[%s]"%", ".join(values)
+return "[%s]" % ", ".join(values)
 setattr(cls, "getvalue", getvalue)
 cls = PLCOpenParser.GetElementClass("structValue", "value")
 if cls:
 structValue_model = re.compile("(.*):=(.*)")
 def setvalue(self, value):
 elements = []
 for item in extractValues(value[1:-1]):
 result = structValue_model.match(item)
 if result is not None:
 element.setmember(groups[0].strip())
 element.setvalue(groups[1].strip())
 elements.append(element)
 self.value = elements
 setattr(cls, "setvalue", setvalue)
 def getvalue(self):
 values = []
 for element in self.value:
-values.append("%s := %s"%(element.getmember(), element.getvalue()))
+values.append("%s := %s" % (element.getmember(), element.getvalue()))
-return "(%s)"%", ".join(values)
+return "(%s)" % ", ".join(values)
 setattr(cls, "getvalue", getvalue)

changeset 1784	64beb9e9c749
parent 1780	c52d1460cea8
child 1831	56b48961cc68