# This file is part of Beremiz.
# See COPYING file for copyrights details.
from __future__ import absolute_import
from collections import OrderedDict, namedtuple
from plcopen.XSLTModelQuery import XSLTModelQuery, _StringValue, _BoolValue, _translate_args
# -------------------------------------------------------------------------------
# Helpers object for generating pou block instances list
# -------------------------------------------------------------------------------
_Point = namedtuple("Point", ["x", "y"])
_BlockInstanceInfos = namedtuple(
["type", "id", "x", "y", "width", "height", "specific_values", "inputs", "outputs"])
namedtuple("BlockSpecificValues",
["name", "execution_order"]),
_VariableSpecificValues = (
namedtuple("VariableSpecificValues",
["name", "value_type", "execution_order"]),
[_StringValue, _StringValue, int])
_ConnectionSpecificValues = (
namedtuple("ConnectionSpecificValues", ["name"]),
_PowerRailSpecificValues = (
namedtuple("PowerRailSpecificValues", ["connectors"]),
_LDElementSpecificValues = (
namedtuple("LDElementSpecificValues",
["name", "negated", "edge", "storage", "execution_order"]),
[_StringValue, _BoolValue, _StringValue, _StringValue, int])
_DivergenceSpecificValues = (
namedtuple("DivergenceSpecificValues", ["connectors"]),
_SpecificValuesTuples = {
namedtuple("CommentSpecificValues", ["content"]),
"input": _VariableSpecificValues,
"output": _VariableSpecificValues,
"inout": _VariableSpecificValues,
"connector": _ConnectionSpecificValues,
"continuation": _ConnectionSpecificValues,
"leftPowerRail": _PowerRailSpecificValues,
"rightPowerRail": _PowerRailSpecificValues,
"contact": _LDElementSpecificValues,
"coil": _LDElementSpecificValues,
namedtuple("StepSpecificValues", ["name", "initial", "action"]),
[_StringValue, _BoolValue, lambda x: x]),
namedtuple("TransitionSpecificValues",
["priority", "condition_type", "condition", "connection"]),
[int, _StringValue, _StringValue, lambda x: x]),
"selectionDivergence": _DivergenceSpecificValues,
"selectionConvergence": _DivergenceSpecificValues,
"simultaneousDivergence": _DivergenceSpecificValues,
"simultaneousConvergence": _DivergenceSpecificValues,
namedtuple("JumpSpecificValues", ["target"]),
namedtuple("ActionBlockSpecificValues", ["actions"]),
_InstanceConnectionInfos = namedtuple(
"InstanceConnectionInfos",
["name", "negated", "edge", "position", "links"])
_ConnectionLinkInfos = namedtuple(
["refLocalId", "formalParameter", "points"])
class _ActionInfos(object):
__slots__ = ["qualifier", "type", "value", "duration", "indicator"]
def __init__(self, *args):
for attr, value in zip(self.__slots__, args):
setattr(self, attr, value if value is not None else "")
return _ActionInfos(*[getattr(self, attr) for attr in self.__slots__])
class BlockInstanceFactory(object):
def __init__(self, block_instances):
self.BlockInstances = block_instances
self.CurrentInstance = None
self.SpecificValues = None
self.CurrentConnection = None
def SetSpecificValues(self, context, *args):
self.SpecificValues = list(args)
self.CurrentInstance = None
self.CurrentConnection = None
def AddBlockInstance(self, context, *args):
specific_values_tuple, specific_values_translation = \
_SpecificValuesTuples.get(args[0][0], _BlockSpecificValues)
if args[0][0] == "step" and len(self.SpecificValues) < 3 or \
args[0][0] == "transition" and len(self.SpecificValues) < 4:
self.SpecificValues.append([None])
elif args[0][0] == "actionBlock" and len(self.SpecificValues) < 1:
self.SpecificValues.append([[]])
specific_values = specific_values_tuple(*_translate_args(
specific_values_translation, self.SpecificValues))
self.SpecificValues = None
self.CurrentInstance = _BlockInstanceInfos(
*(_translate_args([_StringValue, int] + [float] * 4, args) +
[specific_values, [], []]))
self.BlockInstances[self.CurrentInstance.id] = self.CurrentInstance
def AddInstanceConnection(self, context, *args):
connection_args = _translate_args(
[_StringValue] * 2 + [_BoolValue, _StringValue] + [float] * 2, args)
self.CurrentConnection = _InstanceConnectionInfos(
*(connection_args[1:4] + [
_Point(*connection_args[4:6]), []]))
if self.CurrentInstance is not None:
if connection_args[0] == "input":
self.CurrentInstance.inputs.append(self.CurrentConnection)
self.CurrentInstance.outputs.append(self.CurrentConnection)
self.SpecificValues.append([self.CurrentConnection])
def AddConnectionLink(self, context, *args):
self.CurrentLink = _ConnectionLinkInfos(
*(_translate_args([int, _StringValue], args) + [[]]))
self.CurrentConnection.links.append(self.CurrentLink)
def AddLinkPoint(self, context, *args):
self.CurrentLink.points.append(_Point(
*_translate_args([float] * 2, args)))
def AddAction(self, context, *args):
if len(self.SpecificValues) == 0:
self.SpecificValues.append([[]])
translated_args = _translate_args([_StringValue] * 5, args)
self.SpecificValues[0][0].append(_ActionInfos(*translated_args))
class BlockInstanceCollector(XSLTModelQuery):
""" object for collecting instances path list"""
def __init__(self, controller):
XSLTModelQuery.__init__(self,
"pou_block_instances.xslt",
[(name, self.FactoryCaller(name))
for name in ["AddBlockInstance",
def FactoryCaller(self, funcname):
return getattr(self.factory, funcname)(*args)
def Collect(self, root, debug):
element_instances = OrderedDict()
self.factory = BlockInstanceFactory(element_instances)
self._process_xslt(root, debug)
