plcopen/BlockInstanceCollector.py
author Edouard Tisserant <edouard.tisserant@gmail.com>
Sun, 07 Apr 2019 21:08:07 +0200
changeset 2579 8fb5c6eddc72
parent 1973 cc7a46953471
child 3750 f62625418bff
permissions -rw-r--r--
Conform to pep8 and pylint :

pep8 version: 2.3.1
./ConfigTreeNode.py:130:49: E231 missing whitespace after ','
./editors/Viewer.py:643:24: E128 continuation line under-indented for visual indent
./editors/Viewer.py:670:12: E221 multiple spaces before operator
./editors/Viewer.py:671:13: E221 multiple spaces before operator
./editors/Viewer.py:2138:52: E203 whitespace before ':'
./editors/Viewer.py:2139:66: W291 trailing whitespace
./controls/VariablePanel.py:154:25: E231 missing whitespace after ','
./controls/LocationCellEditor.py:88:1: W293 blank line contains whitespace
./controls/LocationCellEditor.py:191:25: E221 multiple spaces before operator
./controls/LocationCellEditor.py:200:17: E128 continuation line under-indented for visual indent

pylint 1.8.3,
************* Module controls.LocationCellEditor
controls/LocationCellEditor.py:200: [C0330(bad-continuation), ] Wrong continued indentation (add 9 spaces).
_("Selected location is identical to previous one"))
^ |
************* Module controls.VariablePanel
controls/VariablePanel.py:154: [E1601(print-statement), VariableTable.SetValue] print statement used
************* Module editors.Viewer
editors/Viewer.py:643: [C0330(bad-continuation), ] Wrong continued indentation (add 1 space).
self.GetAddMenuCallBack(self.AddNewComment))
^|
editors/Viewer.py:598: [W0612(unused-variable), Viewer.AddDivergenceMenuItems] Unused variable 'add_branch'
editors/Viewer.py:1655: [E0602(undefined-variable), Viewer.PopupConnectionMenu] Undefined variable 'variable_type'
editors/Viewer.py:1649: [W0612(unused-variable), Viewer.PopupConnectionMenu] Unused variable 'connection_type'
************* Module connectors.PYRO.PSK_Adapter
#!/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