asyncua.common.statemachine.FiniteStateMachineTypeClass.install() - Code Metrics - Inspection of "Statemachines" - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#348)

unknown

created 2020-11-27 08:35 UTC

FiniteStateMachineTypeClass.install() A

↳ Parent: asyncua.common.statemachine

Complexity

Conditions

Size

Total Lines	10
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	7
nop	2
dl	0
loc	10
rs	10
c	0
b	0
f	0

import asyncio, logging

from asyncua import Server, ua, Node
from asyncua.common.instantiate_util import instantiate

class StateMachineTypeClass(object):
    '''
    Implementation of an StateMachineType
    '''
    def __init__(self, server=None, parent=None, idx=None, name=None):
        if not server: raise ValueError #FIXME change to check instance type
        if not parent: raise ValueError #FIXME change to check instance type
        if idx == None:
            idx = parent.nodeid.NamespaceIndex
        if name == None:
            name = "StateMachine"
        self._server = server
        self._parent = parent
        self._state_machine_node = None
        self._state_machine_type = ua.NodeId(2299, 0)
        self._name = name
        self._idx = idx
        self._optionals = False
        self._current_state_node = None
        self._current_state_id_node = None
        self._last_transition_node = None
        self._last_transition_node_id = None

    async def install(self, optionals=False):
        '''
        setup adressspace and initialize 
        '''
        self._optionals = optionals
        self._state_machine_node = await self._parent.add_object(
            self._idx, 
            self._name, 
            objecttype=self._state_machine_type, 
            instantiate_optional=optionals
            )
        self._current_state_node = await self._state_machine_node.get_child(["CurrentState"])
        self._current_state_id_node = await self._state_machine_node.get_child(["CurrentState","Id"])
        if self._optionals:
            self._current_transition_node = await self._state_machine_node.get_child(["LastTransition"])
            self._current_transition_id_node = await self._state_machine_node.get_child(["LastTransition","Id"])

        #FIXME initialise values
        '''
        maybe its smart to check parents children for a initial state instance (InitialStateType) 
        and initialize it with its id but if no state instance is provided ... i will log a 
        warning
        '''

    async def change_state(self, state_name, state_node, transition_name=None, transition_node=None):
        '''
        method to change the state of the statemachine
        state_name: ua.LocalizedText()
        state: ua.NodeId() <- StateType node
        transition_name: ua.LocalizedText()
        transition: ua.NodeId() <- TransitionType node
        '''
        #FIXME check StateType exist
        #FIXME check TransitionTypeType exist
        await self.write_state(state_name, state_node)
        if self._optionals and transition_name and transition_node:
            await self.write_transition(transition_name, transition_node)
        #FIXME trigger TransitionEventType

    async def write_state(self, state_name, state_node):
        #FIXME check types/class
        await self._current_state_node.write_value(state_name)
        await self._current_state_id_node.write_value(state_node)

    async def write_transition(self, transition_name, transition_node):
        #FIXME check types/class
        await self._last_transition_node.write_value(transition_name)
        await self._last_transition_id_node.write_value(transition_node)

class FiniteStateMachineTypeClass(StateMachineTypeClass):
    '''
    Implementation of an FiniteStateMachineType
    '''
    def __init__(self, server=None, parent=None, idx=None, name=None):
        super().__init__(server, parent, idx, name)
        if name == None:
            name = "FiniteStateMachine"
        self._state_machine_type = ua.NodeId(2771, 0)
        self._avalible_states = []
        self._avalible_transitions = []

    async def install(self, optionals=False):
        '''
        setup adressspace and initialize 
        '''
        self._optionals = optionals
        self._state_machine_node = await self._parent.add_object(
            self._idx, 
            self._name, 
            objecttype=self._state_machine_type, 
            instantiate_optional=optionals
            )

    async def set_avalible_states(self, states):
        self._avalible_states = states

    async def set_avalible_transitions(self, transitions):
        self._avalible_transitions = transitions      

class ExclusiveLimitStateMachineTypeClass(FiniteStateMachineTypeClass):
    '''
    NOT IMPLEMENTED "ExclusiveLimitStateMachineType"
    '''
    def __init__(self, server=None, parent=None, idx=None, name=None):
        super().__init__(server, parent)
        if name == None:
            name = "ExclusiveLimitStateMachine"
        self._state_machine_type = ua.NodeId(9318, 0)
        raise NotImplementedError

class FileTransferStateMachineTypeClass(FiniteStateMachineTypeClass):
    '''
    NOT IMPLEMENTED "FileTransferStateMachineType"
    '''
    def __init__(self, server=None, parent=None, idx=None, name=None):
        super().__init__(server, parent)
        if name == None:
            name = "FileTransferStateMachine"
        self._state_machine_type = ua.NodeId(15803, 0)
        raise NotImplementedError

class ProgramStateMachineTypeClass(FiniteStateMachineTypeClass):
    '''
    https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.3/
    '''
    def __init__(self, server=None, parent=None, idx=None, name=None):
        super().__init__(server, parent, idx, name)
        if name == None:
            name = "ProgramStateMachine"
        self._state_machine_type = ua.NodeId(2391, 0)
        self._ready_state = None #State node
        self._halted_state = None #State node
        self._running_state = None #State node
        self._suspended_state = None #State node

        self._halted_to_ready = None #Transition node
        self._ready_to_running = None #Transition node
        self._running_to_halted = None #Transition node
        self._running_to_ready = None #Transition node
        self._running_to_suspended = None #Transition node
        self._suspended_to_running = None #Transition node
        self._suspended_to_halted = None #Transition node
        self._suspended_to_ready = None #Transition node
        self._ready_to_halted = None #Transition node

        self._halt_method_node = None #uamethod node
        self._reset_method_node = None #uamethod node
        self._resume_method_node = None #uamethod node
        self._start_method_node = None #uamethod node
        self._suspend_method_node = None #uamethod node

        self.lt_ready = ua.LocalizedText("Ready", "en-US")
        self.lt_running = ua.LocalizedText("Running", "en-US")
        self.lt_halted = ua.LocalizedText("Halted", "en-US")
        self.lt_suspended= ua.LocalizedText("Suspended", "en-US")
        self.lt_halted_to_ready = ua.LocalizedText("HaltedToReady", "en-US")
        self.lt_ready_to_running = ua.LocalizedText("ReadyToRunning", "en-US")
        self.lt_running_to_halted = ua.LocalizedText("RunningToHalted", "en-US")
        self.lt_running_to_ready = ua.LocalizedText("RunningToReady", "en-US")
        self.lt_running_to_suspended = ua.LocalizedText("RunningToSuspended", "en-US")
        self.lt_suspended_to_running = ua.LocalizedText("SuspendedToRunning", "en-US")
        self.lt_suspended_to_halted = ua.LocalizedText("SuspendedToHalted", "en-US")
        self.lt_suspended_to_ready = ua.LocalizedText("SuspendedToReady", "en-US")
        self.lt_ready_to_halted = ua.LocalizedText("ReadyToHalted", "en-US")

    async def install(self, optionals=False):
        '''
        setup adressspace and initialize 
        '''
        self._optionals = optionals
        self._state_machine_node = await self._parent.add_object(
            self._idx, 
            self._name, 
            objecttype=self._state_machine_type, 
            instantiate_optional=optionals
            )
        #FIXME get childnodes:
        self._ready_state = None #State node
        self._halted_state = None #State node
        self._running_state = None #State node
        self._suspended_state = None #State node
        self._halted_to_ready = None #Transition node
        self._ready_to_running = None #Transition node
        self._running_to_halted = None #Transition node
        self._running_to_ready = None #Transition node
        self._running_to_suspended = None #Transition node
        self._suspended_to_running = None #Transition node
        self._suspended_to_halted = None #Transition node
        self._suspended_to_ready = None #Transition node
        self._ready_to_halted = None #Transition node
        self._halt_method_node = None #uamethod node
        self._reset_method_node = None #uamethod node
        self._resume_method_node = None #uamethod node
        self._start_method_node = None #uamethod node
        self._suspend_method_node = None #uamethod node

    #Transition
    async def HaltedToReady(self):
        await self._current_state.write_value(
            self.lt_ready,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._current_state_id.write_value(
            self._ready_state.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        await self._last_transition.write_value(
            self.lt_halted_to_ready,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._last_transition_id.write_value(
            self._halted_to_ready.nodeid,
            varianttype=ua.VariantType.NodeId
            )
        #FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
        return ua.StatusCode(ua.status_codes.StatusCodes.Good)

    #Transition
    async def ReadyToRunning(self):
        await self._current_state.write_value(
            self.lt_running,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._current_state_id.write_value(
            self._running_state.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        await self._last_transition.write_value(
            self.lt_ready_to_running,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._last_transition_id.write_value(
            self._ready_to_running.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        #FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
        return ua.StatusCode(ua.status_codes.StatusCodes.Good)

    #Transition
    async def RunningToHalted(self):
        await self._current_state.write_value(
            self.lt_halted,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._current_state_id.write_value(
            self._halted_state.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        await self._last_transition.write_value(
            self.lt_running_to_halted,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._last_transition_id.write_value(
            self._running_to_halted.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        #FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
        return ua.StatusCode(ua.status_codes.StatusCodes.Good)

    #Transition
    async def RunningToReady(self):
        await self._current_state.write_value(
            self.lt_ready,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._current_state_id.write_value(
            self._ready_state.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        await self._last_transition.write_value(
            self.lt_running_to_ready,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._last_transition_id.write_value(
            self._running_to_ready.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        #FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
        return ua.StatusCode(ua.status_codes.StatusCodes.Good)

    #Transition
    async def RunningToSuspended(self):
        await self._current_state.write_value(
            self.lt_suspended,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._current_state_id.write_value(
            self._suspended_state.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        await self._last_transition.write_value(
            self.lt_running_to_suspended,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._last_transition_id.write_value(
            self._running_to_suspended.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        #FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
        return ua.StatusCode(ua.status_codes.StatusCodes.Good)

    #Transition 
    async def SuspendedToRunning(self):
        await self._current_state.write_value(
            self.lt_running,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._current_state_id.write_value(
            self._running_state.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        await self._last_transition.write_value(
            self.lt_suspended_to_running,
            varianttype=ua.VariantType.LocalizedText
            )
        await self._last_transition_id.write_value(
            self._suspended_to_running.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        #FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
        return ua.StatusCode(ua.status_codes.StatusCodes.Good)

    #Transition
    async def SuspendedToHalted(self):
        await self._current_state.write_value(
            self.lt_halted,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._current_state_id.write_value(
            self._halted_state.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        await self._last_transition.write_value(
            self.lt_suspended_to_halted,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._last_transition_id.write_value(
            self._suspended_to_halted.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        #FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
        return ua.StatusCode(ua.status_codes.StatusCodes.Good)

    #Transition
    async def SuspendedToReady(self):
        await self._current_state.write_value(
            self.lt_ready,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._current_state_id.write_value(
            self._ready_state.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        await self._last_transition.write_value(
            self.lt_suspended_to_ready,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._last_transition_id.write_value(
            self._suspended_to_ready.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        #FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
        return ua.StatusCode(ua.status_codes.StatusCodes.Good)

    #Transition 
    async def ReadyToHalted(self):
        await self._current_state.write_value(
            self.lt_halted,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._current_state_id.write_value(
            self._halted_state.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        await self._last_transition.write_value(
            self.lt_ready_to_halted,
            varianttype=ua.VariantType.LocalizedText
            ) 
        await self._last_transition_id.write_value(
            self._ready_to_halted.nodeid, 
            varianttype=ua.VariantType.NodeId
            )
        #FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
        return ua.StatusCode(ua.status_codes.StatusCodes.Good)

    #method to be linked to uamethod
    async def Start(self):
        if await self._current_state.read_value() == self.lt_ready:
            return await ReadyToRunning()

        else:
            return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)

    #method to be linked to uamethod
    async def Suspend(self):
        if await self._current_state.read_value() == self.lt_running:
            return await RunningToSuspended()

        else:
            return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)

    #method to be linked to uamethod
    async def Resume(self):
        if await self._current_state.read_value() == self.lt_suspended:
            return await SuspendedToRunning()

        else:
            return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)

    #method to be linked to uamethod
    async def Halt(self):
        if await self._current_state.read_value() == self.lt_ready:
            return await ReadyToHalted()

        elif await self._current_state.read_value() == self.lt_running:
            return await RunningToHalted()

        elif await self._current_state.read_value() == self.lt_suspended:
            return await SuspendedToHalted()

        else:
            return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)

    #method to be linked to uamethod
    async def Reset(self):
        if await self._current_state.read_value() == self.lt_halted:
            return await HaltedToReady()

        else:
            return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)

class ShelvedStateMachineTypeClass(FiniteStateMachineTypeClass):
    '''
    NOT IMPLEMENTED "ShelvedStateMachineType"
    '''
    def __init__(self, server=None, parent=None, idx=None, name=None):
        super().__init__(server, parent)
        if name == None:
            name = "ShelvedStateMachine"
        self._state_machine_type = ua.NodeId(2929, 0)
        raise NotImplementedError


#Devtest / Workbench
if __name__ == "__main__":
    async def main():
        logging.basicConfig(level=logging.INFO)
        _logger = logging.getLogger('asyncua')

        server = Server()
        await server.init()

        sm = StateMachineTypeClass(server, server.nodes.objects, 0, "StateMachine")
        await sm.install(True)
        await sm.change_state(ua.LocalizedText("Test", "en-US"), server.get_node("ns=0;i=2253").nodeid)
        fsm = FiniteStateMachineTypeClass(server, server.nodes.objects, 0, "FiniteStateMachine")
        await fsm.install(True)
        pfsm = ProgramStateMachineTypeClass(server, server.nodes.objects, 0, "ProgramStateMachine")
        await pfsm.install(True)

        async with server:
            while 1:
                await asyncio.sleep(0)

    asyncio.run(main())


1			import asyncio, logging
2
3			from asyncua import Server, ua, Node
4			from asyncua.common.instantiate_util import instantiate
5
6			class StateMachineTypeClass(object):
7			'''
8			Implementation of an StateMachineType
9			'''
10			def __init__(self, server=None, parent=None, idx=None, name=None):
11			if not server: raise ValueError #FIXME change to check instance type
12			if not parent: raise ValueError #FIXME change to check instance type
13			if idx == None:
14			idx = parent.nodeid.NamespaceIndex
15			if name == None:
16			name = "StateMachine"
17			self._server = server
18			self._parent = parent
19			self._state_machine_node = None
20			self._state_machine_type = ua.NodeId(2299, 0)
21			self._name = name
22			self._idx = idx
23			self._optionals = False
24			self._current_state_node = None
25			self._current_state_id_node = None
26			self._last_transition_node = None
27			self._last_transition_node_id = None
28
29			async def install(self, optionals=False):
30			'''
31			setup adressspace and initialize
32			'''
33			self._optionals = optionals
34			self._state_machine_node = await self._parent.add_object(
35			self._idx,
36			self._name,
37			objecttype=self._state_machine_type,
38			instantiate_optional=optionals
39			)
40			self._current_state_node = await self._state_machine_node.get_child(["CurrentState"])
41			self._current_state_id_node = await self._state_machine_node.get_child(["CurrentState","Id"])
42			if self._optionals:
43			self._current_transition_node = await self._state_machine_node.get_child(["LastTransition"])
44			self._current_transition_id_node = await self._state_machine_node.get_child(["LastTransition","Id"])
45
46			#FIXME initialise values
47			'''
48			maybe its smart to check parents children for a initial state instance (InitialStateType)
49			and initialize it with its id but if no state instance is provided ... i will log a
50			warning
51			'''
52
53			async def change_state(self, state_name, state_node, transition_name=None, transition_node=None):
54			'''
55			method to change the state of the statemachine
56			state_name: ua.LocalizedText()
57			state: ua.NodeId() <- StateType node
58			transition_name: ua.LocalizedText()
59			transition: ua.NodeId() <- TransitionType node
60			'''
61			#FIXME check StateType exist
62			#FIXME check TransitionTypeType exist
63			await self.write_state(state_name, state_node)
64			if self._optionals and transition_name and transition_node:
65			await self.write_transition(transition_name, transition_node)
66			#FIXME trigger TransitionEventType
67
68			async def write_state(self, state_name, state_node):
69			#FIXME check types/class
70			await self._current_state_node.write_value(state_name)
71			await self._current_state_id_node.write_value(state_node)
72
73			async def write_transition(self, transition_name, transition_node):
74			#FIXME check types/class
75			await self._last_transition_node.write_value(transition_name)
76			await self._last_transition_id_node.write_value(transition_node)
77
78			class FiniteStateMachineTypeClass(StateMachineTypeClass):
79			'''
80			Implementation of an FiniteStateMachineType
81			'''
82			def __init__(self, server=None, parent=None, idx=None, name=None):
83			super().__init__(server, parent, idx, name)
84			if name == None:
85			name = "FiniteStateMachine"
86			self._state_machine_type = ua.NodeId(2771, 0)
87			self._avalible_states = []
88			self._avalible_transitions = []
89
90			async def install(self, optionals=False):
91			'''
92			setup adressspace and initialize
93			'''
94			self._optionals = optionals
95			self._state_machine_node = await self._parent.add_object(
96			self._idx,
97			self._name,
98			objecttype=self._state_machine_type,
99			instantiate_optional=optionals
100			)
101
102			async def set_avalible_states(self, states):
103			self._avalible_states = states
104
105			async def set_avalible_transitions(self, transitions):
106			self._avalible_transitions = transitions
107
108			class ExclusiveLimitStateMachineTypeClass(FiniteStateMachineTypeClass):
109			'''
110			NOT IMPLEMENTED "ExclusiveLimitStateMachineType"
111			'''
112			def __init__(self, server=None, parent=None, idx=None, name=None):
113			super().__init__(server, parent)
114			if name == None:
115			name = "ExclusiveLimitStateMachine"
116			self._state_machine_type = ua.NodeId(9318, 0)
117			raise NotImplementedError
118
119			class FileTransferStateMachineTypeClass(FiniteStateMachineTypeClass):
120			'''
121			NOT IMPLEMENTED "FileTransferStateMachineType"
122			'''
123			def __init__(self, server=None, parent=None, idx=None, name=None):
124			super().__init__(server, parent)
125			if name == None:
126			name = "FileTransferStateMachine"
127			self._state_machine_type = ua.NodeId(15803, 0)
128			raise NotImplementedError
129
130			class ProgramStateMachineTypeClass(FiniteStateMachineTypeClass):
131			'''
132			https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.3/
133			'''
134			def __init__(self, server=None, parent=None, idx=None, name=None):
135			super().__init__(server, parent, idx, name)
136			if name == None:
137			name = "ProgramStateMachine"
138			self._state_machine_type = ua.NodeId(2391, 0)
139			self._ready_state = None #State node
140			self._halted_state = None #State node
141			self._running_state = None #State node
142			self._suspended_state = None #State node
143
144			self._halted_to_ready = None #Transition node
145			self._ready_to_running = None #Transition node
146			self._running_to_halted = None #Transition node
147			self._running_to_ready = None #Transition node
148			self._running_to_suspended = None #Transition node
149			self._suspended_to_running = None #Transition node
150			self._suspended_to_halted = None #Transition node
151			self._suspended_to_ready = None #Transition node
152			self._ready_to_halted = None #Transition node
153
154			self._halt_method_node = None #uamethod node
155			self._reset_method_node = None #uamethod node
156			self._resume_method_node = None #uamethod node
157			self._start_method_node = None #uamethod node
158			self._suspend_method_node = None #uamethod node
159
160			self.lt_ready = ua.LocalizedText("Ready", "en-US")
161			self.lt_running = ua.LocalizedText("Running", "en-US")
162			self.lt_halted = ua.LocalizedText("Halted", "en-US")
163			self.lt_suspended= ua.LocalizedText("Suspended", "en-US")
164			self.lt_halted_to_ready = ua.LocalizedText("HaltedToReady", "en-US")
165			self.lt_ready_to_running = ua.LocalizedText("ReadyToRunning", "en-US")
166			self.lt_running_to_halted = ua.LocalizedText("RunningToHalted", "en-US")
167			self.lt_running_to_ready = ua.LocalizedText("RunningToReady", "en-US")
168			self.lt_running_to_suspended = ua.LocalizedText("RunningToSuspended", "en-US")
169			self.lt_suspended_to_running = ua.LocalizedText("SuspendedToRunning", "en-US")
170			self.lt_suspended_to_halted = ua.LocalizedText("SuspendedToHalted", "en-US")
171			self.lt_suspended_to_ready = ua.LocalizedText("SuspendedToReady", "en-US")
172			self.lt_ready_to_halted = ua.LocalizedText("ReadyToHalted", "en-US")
173
174			async def install(self, optionals=False):
175			'''
176			setup adressspace and initialize
177			'''
178			self._optionals = optionals
179			self._state_machine_node = await self._parent.add_object(
180			self._idx,
181			self._name,
182			objecttype=self._state_machine_type,
183			instantiate_optional=optionals
184			)
185			#FIXME get childnodes:
186			self._ready_state = None #State node
187			self._halted_state = None #State node
188			self._running_state = None #State node
189			self._suspended_state = None #State node
190			self._halted_to_ready = None #Transition node
191			self._ready_to_running = None #Transition node
192			self._running_to_halted = None #Transition node
193			self._running_to_ready = None #Transition node
194			self._running_to_suspended = None #Transition node
195			self._suspended_to_running = None #Transition node
196			self._suspended_to_halted = None #Transition node
197			self._suspended_to_ready = None #Transition node
198			self._ready_to_halted = None #Transition node
199			self._halt_method_node = None #uamethod node
200			self._reset_method_node = None #uamethod node
201			self._resume_method_node = None #uamethod node
202			self._start_method_node = None #uamethod node
203			self._suspend_method_node = None #uamethod node
204
205			#Transition
206			async def HaltedToReady(self):
207			await self._current_state.write_value(
208			self.lt_ready,
209			varianttype=ua.VariantType.LocalizedText
210			)
211			await self._current_state_id.write_value(
212			self._ready_state.nodeid,
213			varianttype=ua.VariantType.NodeId
214			)
215			await self._last_transition.write_value(
216			self.lt_halted_to_ready,
217			varianttype=ua.VariantType.LocalizedText
218			)
219			await self._last_transition_id.write_value(
220			self._halted_to_ready.nodeid,
221			varianttype=ua.VariantType.NodeId
222			)
223			#FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
224			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
225
226			#Transition
227			async def ReadyToRunning(self):
228			await self._current_state.write_value(
229			self.lt_running,
230			varianttype=ua.VariantType.LocalizedText
231			)
232			await self._current_state_id.write_value(
233			self._running_state.nodeid,
234			varianttype=ua.VariantType.NodeId
235			)
236			await self._last_transition.write_value(
237			self.lt_ready_to_running,
238			varianttype=ua.VariantType.LocalizedText
239			)
240			await self._last_transition_id.write_value(
241			self._ready_to_running.nodeid,
242			varianttype=ua.VariantType.NodeId
243			)
244			#FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
245			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
246
247			#Transition
248			async def RunningToHalted(self):
249			await self._current_state.write_value(
250			self.lt_halted,
251			varianttype=ua.VariantType.LocalizedText
252			)
253			await self._current_state_id.write_value(
254			self._halted_state.nodeid,
255			varianttype=ua.VariantType.NodeId
256			)
257			await self._last_transition.write_value(
258			self.lt_running_to_halted,
259			varianttype=ua.VariantType.LocalizedText
260			)
261			await self._last_transition_id.write_value(
262			self._running_to_halted.nodeid,
263			varianttype=ua.VariantType.NodeId
264			)
265			#FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
266			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
267
268			#Transition
269			async def RunningToReady(self):
270			await self._current_state.write_value(
271			self.lt_ready,
272			varianttype=ua.VariantType.LocalizedText
273			)
274			await self._current_state_id.write_value(
275			self._ready_state.nodeid,
276			varianttype=ua.VariantType.NodeId
277			)
278			await self._last_transition.write_value(
279			self.lt_running_to_ready,
280			varianttype=ua.VariantType.LocalizedText
281			)
282			await self._last_transition_id.write_value(
283			self._running_to_ready.nodeid,
284			varianttype=ua.VariantType.NodeId
285			)
286			#FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
287			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
288
289			#Transition
290			async def RunningToSuspended(self):
291			await self._current_state.write_value(
292			self.lt_suspended,
293			varianttype=ua.VariantType.LocalizedText
294			)
295			await self._current_state_id.write_value(
296			self._suspended_state.nodeid,
297			varianttype=ua.VariantType.NodeId
298			)
299			await self._last_transition.write_value(
300			self.lt_running_to_suspended,
301			varianttype=ua.VariantType.LocalizedText
302			)
303			await self._last_transition_id.write_value(
304			self._running_to_suspended.nodeid,
305			varianttype=ua.VariantType.NodeId
306			)
307			#FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
308			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
309
310			#Transition
311			async def SuspendedToRunning(self):
312			await self._current_state.write_value(
313			self.lt_running,
314			varianttype=ua.VariantType.LocalizedText
315			)
316			await self._current_state_id.write_value(
317			self._running_state.nodeid,
318			varianttype=ua.VariantType.NodeId
319			)
320			await self._last_transition.write_value(
321			self.lt_suspended_to_running,
322			varianttype=ua.VariantType.LocalizedText
323			)
324			await self._last_transition_id.write_value(
325			self._suspended_to_running.nodeid,
326			varianttype=ua.VariantType.NodeId
327			)
328			#FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
329			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
330
331			#Transition
332			async def SuspendedToHalted(self):
333			await self._current_state.write_value(
334			self.lt_halted,
335			varianttype=ua.VariantType.LocalizedText
336			)
337			await self._current_state_id.write_value(
338			self._halted_state.nodeid,
339			varianttype=ua.VariantType.NodeId
340			)
341			await self._last_transition.write_value(
342			self.lt_suspended_to_halted,
343			varianttype=ua.VariantType.LocalizedText
344			)
345			await self._last_transition_id.write_value(
346			self._suspended_to_halted.nodeid,
347			varianttype=ua.VariantType.NodeId
348			)
349			#FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
350			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
351
352			#Transition
353			async def SuspendedToReady(self):
354			await self._current_state.write_value(
355			self.lt_ready,
356			varianttype=ua.VariantType.LocalizedText
357			)
358			await self._current_state_id.write_value(
359			self._ready_state.nodeid,
360			varianttype=ua.VariantType.NodeId
361			)
362			await self._last_transition.write_value(
363			self.lt_suspended_to_ready,
364			varianttype=ua.VariantType.LocalizedText
365			)
366			await self._last_transition_id.write_value(
367			self._suspended_to_ready.nodeid,
368			varianttype=ua.VariantType.NodeId
369			)
370			#FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
371			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
372
373			#Transition
374			async def ReadyToHalted(self):
375			await self._current_state.write_value(
376			self.lt_halted,
377			varianttype=ua.VariantType.LocalizedText
378			)
379			await self._current_state_id.write_value(
380			self._halted_state.nodeid,
381			varianttype=ua.VariantType.NodeId
382			)
383			await self._last_transition.write_value(
384			self.lt_ready_to_halted,
385			varianttype=ua.VariantType.LocalizedText
386			)
387			await self._last_transition_id.write_value(
388			self._ready_to_halted.nodeid,
389			varianttype=ua.VariantType.NodeId
390			)
391			#FIXME trigger ProgramTransitionEventType and AuditUpdateMethodEvents (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
392			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
393
394			#method to be linked to uamethod
395			async def Start(self):
396			if await self._current_state.read_value() == self.lt_ready:
397			return await ReadyToRunning()
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-11-27 08:14 UTC by Report Bug Copy Issue Report The variable `ReadyToRunning` does not seem to be defined. Loading history...
398			else:
399			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
400
401			#method to be linked to uamethod
402			async def Suspend(self):
403			if await self._current_state.read_value() == self.lt_running:
404			return await RunningToSuspended()
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-11-26 14:04 UTC by Report Bug Copy Issue Report The variable `RunningToSuspended` does not seem to be defined. Loading history...
405			else:
406			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
407
408			#method to be linked to uamethod
409			async def Resume(self):
410			if await self._current_state.read_value() == self.lt_suspended:
411			return await SuspendedToRunning()
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-11-26 14:04 UTC by Report Bug Copy Issue Report The variable `SuspendedToRunning` does not seem to be defined. Loading history...
412			else:
413			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
414
415			#method to be linked to uamethod
416			async def Halt(self):
417			if await self._current_state.read_value() == self.lt_ready:
418			return await ReadyToHalted()
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-11-27 08:14 UTC by Report Bug Copy Issue Report The variable `ReadyToHalted` does not seem to be defined. Loading history...
419			elif await self._current_state.read_value() == self.lt_running:
420			return await RunningToHalted()
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-11-27 08:14 UTC by Report Bug Copy Issue Report The variable `RunningToHalted` does not seem to be defined. Loading history...
421			elif await self._current_state.read_value() == self.lt_suspended:
422			return await SuspendedToHalted()
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-11-26 14:04 UTC by Report Bug Copy Issue Report The variable `SuspendedToHalted` does not seem to be defined. Loading history...
423			else:
424			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
425
426			#method to be linked to uamethod
427			async def Reset(self):
428			if await self._current_state.read_value() == self.lt_halted:
429			return await HaltedToReady()
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-11-27 08:14 UTC by Report Bug Copy Issue Report The variable `HaltedToReady` does not seem to be defined. Loading history...
430			else:
431			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
432
433			class ShelvedStateMachineTypeClass(FiniteStateMachineTypeClass):
434			'''
435			NOT IMPLEMENTED "ShelvedStateMachineType"
436			'''
437			def __init__(self, server=None, parent=None, idx=None, name=None):
438			super().__init__(server, parent)
439			if name == None:
440			name = "ShelvedStateMachine"
441			self._state_machine_type = ua.NodeId(2929, 0)
442			raise NotImplementedError
443
444
445			#Devtest / Workbench
446			if __name__ == "__main__":
447			async def main():
448			logging.basicConfig(level=logging.INFO)
449			_logger = logging.getLogger('asyncua')
450
451			server = Server()
452			await server.init()
453
454			sm = StateMachineTypeClass(server, server.nodes.objects, 0, "StateMachine")
455			await sm.install(True)
456			await sm.change_state(ua.LocalizedText("Test", "en-US"), server.get_node("ns=0;i=2253").nodeid)
457			fsm = FiniteStateMachineTypeClass(server, server.nodes.objects, 0, "FiniteStateMachine")
458			await fsm.install(True)
459			pfsm = ProgramStateMachineTypeClass(server, server.nodes.objects, 0, "ProgramStateMachine")
460			await pfsm.install(True)
461
462			async with server:
463			while 1:
464			await asyncio.sleep(0)
465
466			asyncio.run(main())
467

FreeOpcUa / opcua-asyncio

Pull Request — master (#348)

FiniteStateMachineTypeClass.install() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like