plcopen/BlockInstanceCollector.py
author Edouard Tisserant
Thu, 04 Jun 2020 11:14:21 +0200
branchsvghmi
changeset 2980 2a21d6060d64
parent 1973 cc7a46953471
child 3750 f62625418bff
permissions -rw-r--r--
SVGHMI: add "unsubscribable" property to widgets in order to generalize what already happens for jump buttons.
In most cases jump buttons do not really subscribe to pointed HMI variable, path is given as a relative page jump path. When widget.unsubscribable is set to true, no subscription is made on page switch, but still offset is updated.
This fixes bug happening on relative jump buttons without "disabled" element where offset did not change on relative page switch.
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# 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(
    "BlockInstanceInfos",
    ["type", "id", "x", "y", "width", "height", "specific_values", "inputs", "outputs"])

_BlockSpecificValues = (
    namedtuple("BlockSpecificValues",
               ["name", "execution_order"]),
    [_StringValue, int])
_VariableSpecificValues = (
    namedtuple("VariableSpecificValues",
               ["name", "value_type", "execution_order"]),
    [_StringValue, _StringValue, int])
_ConnectionSpecificValues = (
    namedtuple("ConnectionSpecificValues", ["name"]),
    [_StringValue])

_PowerRailSpecificValues = (
    namedtuple("PowerRailSpecificValues", ["connectors"]),
    [int])

_LDElementSpecificValues = (
    namedtuple("LDElementSpecificValues",
               ["name", "negated", "edge", "storage", "execution_order"]),
    [_StringValue, _BoolValue, _StringValue, _StringValue, int])

_DivergenceSpecificValues = (
    namedtuple("DivergenceSpecificValues", ["connectors"]),
    [int])

_SpecificValuesTuples = {
    "comment": (
        namedtuple("CommentSpecificValues", ["content"]),
        [_StringValue]),
    "input": _VariableSpecificValues,
    "output": _VariableSpecificValues,
    "inout": _VariableSpecificValues,
    "connector": _ConnectionSpecificValues,
    "continuation": _ConnectionSpecificValues,
    "leftPowerRail": _PowerRailSpecificValues,
    "rightPowerRail": _PowerRailSpecificValues,
    "contact": _LDElementSpecificValues,
    "coil": _LDElementSpecificValues,
    "step": (
        namedtuple("StepSpecificValues", ["name", "initial", "action"]),
        [_StringValue, _BoolValue, lambda x: x]),
    "transition": (
        namedtuple("TransitionSpecificValues",
                   ["priority", "condition_type", "condition", "connection"]),
        [int, _StringValue, _StringValue, lambda x: x]),
    "selectionDivergence": _DivergenceSpecificValues,
    "selectionConvergence": _DivergenceSpecificValues,
    "simultaneousDivergence": _DivergenceSpecificValues,
    "simultaneousConvergence": _DivergenceSpecificValues,
    "jump": (
        namedtuple("JumpSpecificValues", ["target"]),
        [_StringValue]),
    "actionBlock": (
        namedtuple("ActionBlockSpecificValues", ["actions"]),
        [lambda x: x]),
}

_InstanceConnectionInfos = namedtuple(
    "InstanceConnectionInfos",
    ["name", "negated", "edge", "position", "links"])

_ConnectionLinkInfos = namedtuple(
    "ConnectionLinkInfos",
    ["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 "")

    def copy(self):
        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
        self.CurrentLink = None

    def SetSpecificValues(self, context, *args):
        self.SpecificValues = list(args)
        self.CurrentInstance = None
        self.CurrentConnection = None
        self.CurrentLink = 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)
            else:
                self.CurrentInstance.outputs.append(self.CurrentConnection)
        else:
            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,
                                controller,
                                "pou_block_instances.xslt",
                                [(name, self.FactoryCaller(name))
                                 for name in ["AddBlockInstance",
                                              "SetSpecificValues",
                                              "AddInstanceConnection",
                                              "AddConnectionLink",
                                              "AddLinkPoint",
                                              "AddAction"]])

    def FactoryCaller(self, funcname):
        def CallFactory(*args):
            return getattr(self.factory, funcname)(*args)
        return CallFactory

    def Collect(self, root, debug):
        element_instances = OrderedDict()
        self.factory = BlockInstanceFactory(element_instances)
        self._process_xslt(root, debug)
        self.factory = None
        return element_instances