Continue fixing deadlock of previous commit, this time occuring when waiting for 'cleanup' python runtime call to finish. Now 'init' and 'cleanup' python runtime calls are called directly from main thread, and aren't anymore invoked in the context of wxMainloop and/or twisted reactor.
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This file is part of Beremiz, a Integrated Development Environment for
# programming IEC 61131-3 automates supporting plcopen standard and CanFestival.
#
# Copyright (C) 2007: Edouard TISSERANT and Laurent BESSARD
# Copyright (C) 2017: Andrey Skvortsov
#
# See COPYING file for copyrights details.
#
# This program 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
# of the License, or (at your option) any later version.
#
# This program 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 program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
from __future__ import absolute_import
from __future__ import division
import re
from collections import OrderedDict
from six.moves import xrange
from lxml import etree
from xmlclass import *
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",
"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 = OrderedDict([
("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 = r"(%%[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
class rect(object):
"""
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.span()[1]
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.end():
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"))
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>
<sfc><scaling x="0" y="0"/></sfc>
</coordinateInfo>
</contentHeader>
<types>
<dataTypes/>
<pous>%s</pous>
</types>
<instances>
<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(r"(?<!<xhtml:p>)(?:<!\[CDATA\[)"), "<xhtml:p><![CDATA["),
(re.compile(r"(?:]]>)(?!</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:
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" % (time_values[3] / 1000)
else:
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 as e:
return None, str(e)
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()
# ----------------------------------------------------------------------
def _updateFormattedTextClass(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:
groups = result.groups()
new_address = groups[0] + new_leading + groups[2]
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("formattedText")
if cls:
_updateFormattedTextClass(cls)
# ----------------------------------------------------------------------
def _updateProjectClass(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 [
("companyName", fileheader_obj.getcompanyName()),
("companyURL", fileheader_obj.getcompanyURL()),
("productName", fileheader_obj.getproductName()),
("productVersion", fileheader_obj.getproductVersion()),
("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 [
("projectName", contentheader_obj.getname()),
("projectVersion", contentheader_obj.getversion()),
("modificationDateTime", contentheader_obj.getmodificationDateTime()),
("organization", contentheader_obj.getorganization()),
("authorName", contentheader_obj.getauthor()),
("language", contentheader_obj.getlanguage())]
}
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,
"authorName": contentheader_obj.setauthor,
"pageSize": lambda v: contentheader_obj.setpageSize(*v),
"scaling": contentheader_obj.setscaling}.get(attr)
if func is not None and value 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)
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=None):
filter = [] if filter is None else 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)
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)
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:
raise ValueError(
_("\"{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)
setattr(cls, "addconfigurationResource", addconfigurationResource)
def removeconfigurationResource(self, config_name, name):
configuration = self.getconfiguration(config_name)
found = False
if configuration is not None:
resource = self.getconfigurationResource(config_name, name)
if resource is not None:
configuration.remove(resource)
found = True
if not found:
raise ValueError(
_("\"{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)
for pou in self.getpous():
pou.updateElementName(old_name, new_name)
for configuration in self.getconfigurations():
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.getpous():
pou.updateElementAddress(address_model, new_leading)
for configuration in self.getconfigurations():
configuration.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def removeVariableByAddress(self, address):
for pou in self.getpous():
pou.removeVariableByAddress(address)
for configuration in self.getconfigurations():
configuration.removeVariableByAddress(address)
setattr(cls, "removeVariableByAddress", removeVariableByAddress)
def removeVariableByFilter(self, leading):
address_model = re.compile(FILTER_ADDRESS_MODEL % leading)
for pou in self.getpous():
pou.removeVariableByFilter(address_model)
for configuration in self.getconfigurations():
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=None):
parent_infos = [] if parent_infos is None else 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 = PLCOpenParser.GetElementClass("project")
if cls:
_updateProjectClass(cls)
# ----------------------------------------------------------------------
def _updateContentHeaderProjectClass(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")
return scaling
setattr(cls, "getscaling", getscaling)
cls = PLCOpenParser.GetElementClass("contentHeader", "project")
if cls:
_updateContentHeaderProjectClass(cls)
# ----------------------------------------------------------------------
def _updateCoordinateInfoContentHeaderClass(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)
cls = PLCOpenParser.GetElementClass("coordinateInfo", "contentHeader")
if cls:
_updateCoordinateInfoContentHeaderClass(cls)
# ----------------------------------------------------------------------
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 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=None):
parent_infos = [] if parent_infos is None else parent_infos
search_result = _Search([("name", self.getname())], criteria, parent_infos)
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
# ----------------------------------------------------------------------
def _updateConfigurationConfigurationsClass(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")
var.setname(name)
var.settype(var_type)
if location != "":
var.setaddress(location)
if description != "":
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)
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=None):
parent_infos = [] if parent_infos is None else 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 = PLCOpenParser.GetElementClass("configuration", "configurations")
if cls:
_updateConfigurationConfigurationsClass(cls)
# ----------------------------------------------------------------------
def _updateResourceConfigurationClass(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=None):
# FIXME : two next lines are incompatible [][-1] raises exception !
parent_infos = [] if parent_infos is None else 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 = PLCOpenParser.GetElementClass("resource", "configuration")
if cls:
_updateResourceConfigurationClass(cls)
# ----------------------------------------------------------------------
def _updateTaskResourceClass(cls):
def updateElementName(self, old_name, new_name):
if TextMatched(self.single, old_name):
self.single = new_name
if TextMatched(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=None):
parent_infos = [] if parent_infos is None else parent_infos
return _Search([("single", self.getsingle()),
("interval", self.getinterval()),
("priority", str(self.getpriority()))],
criteria, parent_infos)
setattr(cls, "Search", Search)
cls = PLCOpenParser.GetElementClass("task", "resource")
if cls:
_updateTaskResourceClass(cls)
# ----------------------------------------------------------------------
def _updatePouInstanceClass(cls):
def updateElementName(self, old_name, new_name):
if TextMatched(self.typeName, old_name):
self.typeName = new_name
setattr(cls, "updateElementName", updateElementName)
def Search(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else parent_infos
return _Search([("name", self.getname()),
("type", self.gettypeName())],
criteria, parent_infos)
setattr(cls, "Search", Search)
cls = PLCOpenParser.GetElementClass("pouInstance")
if cls:
_updatePouInstanceClass(cls)
# ----------------------------------------------------------------------
def _updateVariableVarListPlain(cls):
def gettypeAsText(self):
vartype_content = self.gettype().getcontent()
vartype_content_name = vartype_content.getLocalTag()
# Variable type is a user data type
if vartype_content_name == "derived":
return vartype_content.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.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)
# Variable type is an elementary type
return vartype_content_name
setattr(cls, "gettypeAsText", gettypeAsText)
def Search(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else 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:
search_result.extend(_Search([("initial value", initial.getvalue())], criteria, parent_infos))
doc = self.getdocumentation()
if doc is not None:
search_result.extend(doc.Search(criteria, parent_infos + ["documentation"]))
return search_result
setattr(cls, "Search", Search)
cls = PLCOpenParser.GetElementClass("variable", "varListPlain")
if cls:
_updateVariableVarListPlain(cls)
# ----------------------------------------------------------------------
def _updateTypesProjectClass(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
return None
setattr(cls, "getdataTypeElement", getdataTypeElement)
def appenddataTypeElement(self, name):
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)
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
return None
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)
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)
setattr(cls, "removepouElement", removepouElement)
def Search(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else parent_infos
search_result = []
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 = PLCOpenParser.GetElementClass("types", "project")
if cls:
_updateTypesProjectClass(cls)
# ----------------------------------------------------------------------
def _updateBaseTypeElementName(self, old_name, new_name):
self.baseType.updateElementName(old_name, new_name)
def _updateDataTypeDataTypesClass(cls):
setattr(cls, "updateElementName", _updateBaseTypeElementName)
def Search(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else parent_infos
search_result = []
filter = criteria["filter"]
if filter == "all" or "datatype" in filter:
parent_infos = parent_infos + ["D::%s" % self.getname()]
search_result.extend(_Search([("name", self.getname())], criteria, parent_infos))
search_result.extend(self.baseType.Search(criteria, parent_infos))
if self.initialValue is not None:
search_result.extend(_Search([("initial", self.initialValue.getvalue())], criteria, parent_infos))
return search_result
setattr(cls, "Search", Search)
cls = PLCOpenParser.GetElementClass("dataType", "dataTypes")
if cls:
_updateDataTypeDataTypesClass(cls)
# ----------------------------------------------------------------------
def _updateDataTypeClass(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":
for element in self.content.getvariable():
element.type.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def Search(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else parent_infos
search_result = []
content_name = self.content.getLocalTag()
if content_name in ["derived", "array", "enum", "subrangeSigned", "subrangeUnsigned"]:
search_result.extend(self.content.Search(criteria, parent_infos + ["base"]))
elif content_name == "struct":
for i, element in enumerate(self.content.getvariable()):
search_result.extend(element.Search(criteria, parent_infos + ["struct", i]))
else:
if content_name in ["string", "wstring"]:
content_name = content_name.upper()
search_result.extend(_Search([("base", content_name)], criteria, parent_infos))
return search_result
setattr(cls, "Search", Search)
cls = PLCOpenParser.GetElementClass("dataType")
if cls:
_updateDataTypeClass(cls)
# ----------------------------------------------------------------------
def _updateDerivedDataTypeClass(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=None):
parent_infos = [] if parent_infos is None else parent_infos
return [(tuple(parent_infos),) + result for result in TestTextElement(self.name, criteria)]
setattr(cls, "Search", Search)
cls = PLCOpenParser.GetElementClass("derived", "dataType")
if cls:
_updateDerivedDataTypeClass(cls)
# ----------------------------------------------------------------------
def _updateArrayDataTypeClass(cls):
setattr(cls, "updateElementName", _updateBaseTypeElementName)
def Search(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else 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)
cls = PLCOpenParser.GetElementClass("array", "dataType")
if cls:
_updateArrayDataTypeClass(cls)
# ----------------------------------------------------------------------
def _SearchInSubrange(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else 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)
# ----------------------------------------------------------------------
cls = PLCOpenParser.GetElementClass("subrangeUnsigned", "dataType")
if cls:
setattr(cls, "updateElementName", _updateBaseTypeElementName)
setattr(cls, "Search", _SearchInSubrange)
# ----------------------------------------------------------------------
def _updateEnumDataTypeClass(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=None):
parent_infos = [] if parent_infos is None else 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)
cls = PLCOpenParser.GetElementClass("enum", "dataType")
if cls:
_updateEnumDataTypeClass(cls)
# ----------------------------------------------------------------------
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()
def _updatePouPousClass(cls):
def getblockInfos(self):
block_infos = {
"name": self.getname(),
"type": self.getpouType(),
"extensible": False,
"inputs": [],
"outputs": [],
"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"))
block_infos["inputs"].extend(
[(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)
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)
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.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:
varlist = PLCOpenParser.CreateElement(var_class, "interface")
self.interface.setcontent([varlist])
elif content[-1].getLocalTag() != var_class:
varlist = PLCOpenParser.CreateElement(var_class, "interface")
content[-1].addnext(varlist)
else:
varlist = content[-1]
variables = varlist.getvariable()
if varlist.getconstant() or varlist.getretain() or len(variables) > 0 and variables[0].getaddress():
varlist = PLCOpenParser.CreateElement(var_class, "interface")
content[-1].addnext(varlist)
var = PLCOpenParser.CreateElement("variable", "varListPlain")
var.setname(name)
var.settype(var_type)
if location != "":
var.setaddress(location)
if description != "":
ft = PLCOpenParser.CreateElement("documentation", "variable")
ft.setanyText(description)
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()
for var in variables:
if TextMatched(var.getname(), old_name):
vartype_content = var.gettype().getcontent()
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():
if TextMatched(var.getname(), name):
vartype_content = var.gettype().getcontent()
if vartype_content.getLocalTag() == "derived" and TextMatched(vartype_content.getname(), var_type):
varlist.remove(var)
if len(varlist.getvariable()) == 0:
self.interface.remove(varlist)
break
setattr(cls, "removepouVar", removepouVar)
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 (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
if self.actions:
for action in self.actions.getaction():
result = action.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")
self.transitions.appendtransition(transition)
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)
setattr(cls, "removetransition", removetransition)
def addaction(self, name, body_type):
if self.actions is None:
self.addactions()
self.actions.setaction([])
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)
setattr(cls, "removeaction", removeaction)
def updateElementName(self, old_name, new_name):
if self.interface is not None:
for content in self.interface.getcontent():
for var in content.getvariable():
var_address = var.getaddress()
if var_address is not None:
if TextMatched(var_address, old_name):
var.setaddress(new_name)
if TextMatched(var.getname(), old_name):
var.setname(new_name)
var_type_content = var.gettype().getcontent()
if var_type_content.getLocalTag() == "derived":
if TextMatched(var_type_content.getname(), old_name):
var_type_content.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 is not None:
for content in self.interface.getcontent():
for var in content.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 is not None:
for content in self.interface.getcontent():
for variable in content.getvariable():
if TextMatched(variable.getaddress(), address):
content.remove(variable)
setattr(cls, "removeVariableByAddress", removeVariableByAddress)
def removeVariableByFilter(self, address_model):
if self.interface is not None:
for content in self.interface.getcontent():
for variable in content.getvariable():
var_address = variable.getaddress()
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=None):
parent_infos = [] if parent_infos is None else parent_infos
search_result = []
filter = criteria["filter"]
if filter == "all" or self.getpouType() in filter:
if parent_infos == []:
parent_infos = parent_infos + ["P::%s" % self.getname()]
search_result.extend(_Search([("name", self.getname())], criteria, parent_infos))
if self.interface is not None:
var_number = 0
for content in self.interface.getcontent():
variable_type = searchResultVarTypes.get(content, "var_local")
variables = content.getvariable()
for modifier, has_modifier in [("constant", content.getconstant()),
("retain", content.getretain()),
("non_retain", content.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)
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")
_updatePouPousClass(cls)
# ----------------------------------------------------------------------
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)
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 \
(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)
# ----------------------------------------------------------------------
def _updateTransitionTransitionsClass(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)
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)
search_result.extend(self.body.Search(criteria, parent_infos))
return search_result
setattr(cls, "Search", Search)
cls = PLCOpenParser.GetElementClass("transition", "transitions")
if cls:
_updateTransitionTransitionsClass(cls)
# ----------------------------------------------------------------------
def _updateActionActionsClass(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)
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)
search_result.extend(self.body.Search(criteria, parent_infos))
return search_result
setattr(cls, "Search", Search)
cls = PLCOpenParser.GetElementClass("action", "actions")
if cls:
_updateActionActionsClass(cls)
# ----------------------------------------------------------------------
def _updateBodyClass(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!"))
setattr(cls, "resetexecutionOrder", resetexecutionOrder)
def compileexecutionOrder(self):
if self.content.getLocalTag() == "FBD":
self.resetexecutionOrder()
self.resetcurrentExecutionOrderId()
for element in self.content.getcontent():
if isinstance(element, PLCOpenParser.GetElementClass("outVariable", "fbdObjects")) and element.getexecutionOrderId() == 0:
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!"))
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 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 \
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!"))
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():
if element != instance and not isinstance(element, PLCOpenParser.GetElementClass("comment", "commonObjects")):
element_executionOrder = element.getexecutionOrderId()
if old_executionOrder <= element_executionOrder <= new_executionOrder:
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!"))
setattr(cls, "setelementExecutionOrder", setelementExecutionOrder)
def appendcontentInstance(self, instance):
if self.content.getLocalTag() in ["LD", "FBD", "SFC"]:
self.content.appendcontent(instance)
else:
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"]:
return self.content.getcontent()
else:
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"]:
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())
setattr(cls, "getcontentInstance", getcontentInstance)
def getcontentInstancesIds(self):
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())
setattr(cls, "getcontentInstancesIds", getcontentInstancesIds)
def getcontentInstanceByName(self, name):
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())
setattr(cls, "getcontentInstanceByName", getcontentInstanceByName)
def removecontentInstance(self, local_id):
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)
else:
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"]:
self.content.setanyText(text)
else:
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"]:
return self.content.getanyText()
else:
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"]:
return self.content.hasblock(block_type)
else:
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():
element.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
if self.content.getLocalTag() in ["IL", "ST"]:
self.content.updateElementAddress(address_model, new_leading)
else:
for element in self.content.getcontent():
element.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def Search(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else parent_infos
if self.content.getLocalTag() in ["IL", "ST"]:
search_result = self.content.Search(criteria, parent_infos + ["body", 0])
else:
search_result = []
for element in self.content.getcontent():
search_result.extend(element.Search(criteria, parent_infos))
return search_result
setattr(cls, "Search", Search)
cls = PLCOpenParser.GetElementClass("body")
if cls:
_updateBodyClass(cls)
# ----------------------------------------------------------------------
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 (localId, connected) in connections and \
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:
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=None):
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, "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 _updateCommentCommonObjectsClass(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):
self.content.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def Search(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else parent_infos
return self.content.Search(criteria, parent_infos + ["comment", self.getlocalId(), "content"])
setattr(cls, "Search", Search)
cls = _initElementClass("comment", "commonObjects")
if cls:
_updateCommentCommonObjectsClass(cls)
# ----------------------------------------------------------------------
def _updateBlockFbdObjectsClass(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 updateElementName(self, old_name, new_name):
if TextMatched(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=None):
parent_infos = [] if parent_infos is None else 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("block", "fbdObjects")
if cls:
_updateBlockFbdObjectsClass(cls)
# ----------------------------------------------------------------------
_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=None):
parent_infos = [] if parent_infos is None else 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)
setattr(cls, "Search", _getSearchInLDElement("contact"))
# ----------------------------------------------------------------------
cls = _initElementClass("coil", "ldObjects", "single")
if cls:
setattr(cls, "updateElementName", _UpdateLDElementName)
setattr(cls, "updateElementAddress", _UpdateLDElementAddress)
setattr(cls, "Search", _getSearchInLDElement("coil"))
# ----------------------------------------------------------------------
def _updateStepSfcObjectSingleClass(cls):
def Search(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else parent_infos
return _Search([("name", self.getname())], criteria, parent_infos + ["step", self.getlocalId()])
setattr(cls, "Search", Search)
cls = _initElementClass("step", "sfcObjects", "single")
if cls:
_updateStepSfcObjectSingleClass(cls)
# ----------------------------------------------------------------------
def _updateTransitionSfcObjectsClass(cls):
def setconditionContent(self, condition_type, value):
if self.condition is None:
self.addcondition()
if condition_type == "connection":
condition = PLCOpenParser.CreateElement("connectionPointIn", "condition")
else:
condition = PLCOpenParser.CreateElement(condition_type, "condition")
self.condition.setcontent(condition)
if condition_type == "reference":
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()}
if values["type"] == "reference":
values["value"] = content.getname()
elif values["type"] == "inline":
values["value"] = content.gettext()
elif values["type"] == "connectionPointIn":
values["type"] = "connection"
values["value"] = content
return values
return ""
setattr(cls, "getconditionContent", getconditionContent)
def getconditionConnection(self):
if self.condition is not None:
content = self.condition.getcontent()
if content.getLocalTag() == "connectionPointIn":
return content
return None
setattr(cls, "getconditionConnection", getconditionConnection)
def getBoundingBox(self):
bbox = _getBoundingBoxSingle(self)
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()
if condition_connection is not None:
connections_end.extend(_updateConnectionsId(condition_connection, translation))
return _getconnectionsdefinition(self, connections_end)
setattr(cls, "updateConnectionsId", updateConnectionsId)
def updateElementName(self, old_name, new_name):
if self.condition is not None:
content = self.condition.getcontent()
content_name = content.getLocalTag()
if content_name == "reference":
if TextMatched(content.getname(), old_name):
content.setname(new_name)
elif content_name == "inline":
content.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
if self.condition is not None:
content = self.condition.getcontent()
content_name = content.getLocalTag()
if content_name == "reference":
content.setname(update_address(content.getname(), address_model, new_leading))
elif content_name == "inline":
content.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def getconnections(self):
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=None):
parent_infos = [] if parent_infos is None else parent_infos
parent_infos = parent_infos + ["transition", self.getlocalId()]
search_result = []
content = self.condition.getcontent()
content_name = content.getLocalTag()
if content_name == "reference":
search_result.extend(_Search([("reference", content.getname())], criteria, parent_infos))
elif content_name == "inline":
search_result.extend(content.Search(criteria, parent_infos + ["inline"]))
return search_result
setattr(cls, "Search", Search)
cls = _initElementClass("transition", "sfcObjects")
if cls:
_updateTransitionSfcObjectsClass(cls)
# ----------------------------------------------------------------------
_initElementClass("selectionDivergence", "sfcObjects", "single")
_initElementClass("selectionConvergence", "sfcObjects", "multiple")
_initElementClass("simultaneousDivergence", "sfcObjects", "single")
_initElementClass("simultaneousConvergence", "sfcObjects", "multiple")
def _updateJumpStepSfcObjectSingleClass(cls):
def Search(self, criteria, parent_infos):
return _Search([("target", self.gettargetName())], criteria, parent_infos + ["jump", self.getlocalId()])
setattr(cls, "Search", Search)
cls = _initElementClass("jumpStep", "sfcObjects", "single")
if cls:
_updateJumpStepSfcObjectSingleClass(cls)
# ----------------------------------------------------------------------
def _updateActionActionBlockClass(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 updateElementName(self, old_name, new_name):
if self.reference is not None and TextMatched(self.reference.getname(), old_name):
self.reference.setname(new_name)
if self.inline is not None:
self.inline.updateElementName(old_name, new_name)
setattr(cls, "updateElementName", updateElementName)
def updateElementAddress(self, address_model, new_leading):
if self.reference is not None:
self.reference.setname(update_address(self.reference.getname(), address_model, new_leading))
if self.inline is not None:
self.inline.updateElementAddress(address_model, new_leading)
setattr(cls, "updateElementAddress", updateElementAddress)
def Search(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else 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 = PLCOpenParser.GetElementClass("action", "actionBlock")
if cls:
_updateActionActionBlockClass(cls)
# ----------------------------------------------------------------------
def _updateActionBlockCommonObjectsSingleClass(cls):
def setactions(self, actions):
self.action = []
for params in actions:
action = PLCOpenParser.CreateElement("action", "actionBlock")
self.appendaction(action)
action.setqualifier(params.qualifier)
if params.type == "reference":
action.addreference()
action.setreferenceName(params.value)
else:
action.addinline()
action.setinlineContent(params.value)
if params.duration != "":
action.setduration(params.duration)
if params.indicator != "":
action.setindicator(params.indicator)
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() is not None:
params["type"] = "reference"
params["value"] = action.getreferenceName()
elif action.getinline() is not None:
params["type"] = "inline"
params["value"] = action.getinlineContent()
duration = action.getduration()
if duration:
params["duration"] = duration
indicator = action.getindicator()
if indicator is not None:
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=None):
parent_infos = [] if parent_infos is None else 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)
cls = _initElementClass("actionBlock", "commonObjects", "single")
if cls:
_updateActionBlockCommonObjectsSingleClass(cls)
# ----------------------------------------------------------------------
def _SearchInIOVariable(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else 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)
cls = _initElementClass("outVariable", "fbdObjects", "single")
if cls:
setattr(cls, "updateElementName", _UpdateIOElementName)
setattr(cls, "updateElementAddress", _UpdateIOElementAddress)
setattr(cls, "Search", _SearchInIOVariable)
cls = _initElementClass("inOutVariable", "fbdObjects", "single")
if cls:
setattr(cls, "updateElementName", _UpdateIOElementName)
setattr(cls, "updateElementAddress", _UpdateIOElementAddress)
setattr(cls, "Search", _SearchInIOVariable)
def _SearchInConnector(self, criteria, parent_infos=None):
parent_infos = [] if parent_infos is None else parent_infos
return _Search([("name", self.getname())], criteria, parent_infos + ["connector", self.getlocalId()])
# ----------------------------------------------------------------------
def _updateContinuationCommonObjectsClass(cls):
setattr(cls, "Search", _SearchInConnector)
def updateElementName(self, old_name, new_name):
if TextMatched(self.name, old_name):
self.name = new_name
setattr(cls, "updateElementName", updateElementName)
cls = _initElementClass("continuation", "commonObjects")
if cls:
_updateContinuationCommonObjectsClass(cls)
# ----------------------------------------------------------------------
def _updateConnectorCommonObjectsSingleClass(cls):
setattr(cls, "Search", _SearchInConnector)
def updateElementName(self, old_name, new_name):
if TextMatched(self.name, old_name):
self.name = new_name
setattr(cls, "updateElementName", updateElementName)
cls = _initElementClass("connector", "commonObjects", "single")
if cls:
_updateConnectorCommonObjectsSingleClass(cls)
# ----------------------------------------------------------------------
def _updateConnectionClass(cls):
def setpoints(self, points):
positions = []
for point in points:
position = PLCOpenParser.CreateElement("position", "connection")
position.setx(point.x)
position.sety(point.y)
positions.append(position)
self.position = positions
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 = PLCOpenParser.GetElementClass("connection")
if cls:
_updateConnectionClass(cls)
# ----------------------------------------------------------------------
def _updateConnectionPointInClass(cls):
def setrelPositionXY(self, x, y):
self.relPosition = PLCOpenParser.CreateElement("relPosition", "connectionPointIn")
self.relPosition.setx(x)
self.relPosition.sety(y)
setattr(cls, "setrelPositionXY", setrelPositionXY)
def getrelPositionXY(self):
if self.relPosition is not None:
return self.relPosition.getx(), self.relPosition.gety()
return self.relPosition
setattr(cls, "getrelPositionXY", getrelPositionXY)
def addconnection(self):
self.append(PLCOpenParser.CreateElement("connection", "connectionPointIn"))
setattr(cls, "addconnection", addconnection)
def removeconnection(self, idx):
if len(self.content) > idx:
self.remove(self.content[idx])
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 getconnectionPoints(self, idx):
connection = self.getconnection(idx)
if connection is not None:
return connection.getpoints()
return []
setattr(cls, "getconnectionPoints", getconnectionPoints)
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
setattr(cls, "getconnectionParameter", getconnectionParameter)
cls = PLCOpenParser.GetElementClass("connectionPointOut")
if cls:
def setrelPositionXY(self, x, y):
self.relPosition = PLCOpenParser.CreateElement("relPosition", "connectionPointOut")
self.relPosition.setx(x)
self.relPosition.sety(y)
setattr(cls, "setrelPositionXY", setrelPositionXY)
def getrelPositionXY(self):
if self.relPosition is not None:
return self.relPosition.getx(), self.relPosition.gety()
return self.relPosition
setattr(cls, "getrelPositionXY", getrelPositionXY)
cls = PLCOpenParser.GetElementClass("connectionPointIn")
if cls:
_updateConnectionPointInClass(cls)
# ----------------------------------------------------------------------
def _updateValueClass(cls):
def setvalue(self, value):
value = value.strip()
if value.startswith("[") and value.endswith("]"):
content = PLCOpenParser.CreateElement("arrayValue", "value")
elif value.startswith("(") and value.endswith(")"):
content = PLCOpenParser.CreateElement("structValue", "value")
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)
cls = PLCOpenParser.GetElementClass("value")
if cls:
_updateValueClass(cls)
# ----------------------------------------------------------------------
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!") % values)
return items
def _updateArrayValueValueClass(cls):
def setvalue(self, value):
elements = []
for item in extractValues(value[1:-1]):
item = item.strip()
element = PLCOpenParser.CreateElement("value", "arrayValue")
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)
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 Exception:
repetition = 1
if 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 = PLCOpenParser.GetElementClass("arrayValue", "value")
if cls:
arrayValue_model = re.compile(r"([0-9]+)\((.*)\)$")
_updateArrayValueValueClass(cls)
# ----------------------------------------------------------------------
def _updateStructValueValueClass(cls):
def setvalue(self, value):
elements = []
for item in extractValues(value[1:-1]):
result = structValue_model.match(item)
if result is not None:
groups = result.groups()
element = PLCOpenParser.CreateElement("value", "structValue")
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()))
return "(%s)" % ", ".join(values)
setattr(cls, "getvalue", getvalue)
cls = PLCOpenParser.GetElementClass("structValue", "value")
if cls:
structValue_model = re.compile("(.*):=(.*)")
_updateStructValueValueClass(cls)