asyncua.common.statemachine.ProgramStateMachineTypeClass.ReadyToRunning() - 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 09:40 UTC

ProgramStateMachineTypeClass.ReadyToRunning() A

↳ Parent: asyncua.common.statemachine

Complexity

Conditions

Size

Total Lines	21
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	14
nop	1
dl	0
loc	21
rs	9.7
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, idx, name)
        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, idx, name)
        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_node = None #State node
        self._halted_state_node = None #State node
        self._running_state_node = None #State node
        self._suspended_state_node = None #State node

        self._halted_to_ready_node = None #Transition node
        self._ready_to_running_node = None #Transition node
        self._running_to_halted_node = None #Transition node
        self._running_to_ready_node = None #Transition node
        self._running_to_suspended_node = None #Transition node
        self._suspended_to_running_node = None #Transition node
        self._suspended_to_halted_node = None #Transition node
        self._suspended_to_ready_node = None #Transition node
        self._ready_to_halted_node = None #Transition node

        # 5.2.3.2 ProgramStateMachineType states
        self._halted_state_id_node = None #State property (StateNumber value 11)
        self._ready_state_id_node = None #State property (StateNumber value 12)
        self._running_state_id_node = None #State property (StateNumber value 13)
        self._suspended_state_id_node = None #State property (StateNumber value 14)

        # 5.2.3.3 ProgramStateMachineType transitions
        self._halted_to_ready_id_node = None #Transition property (TransitionNumber value 1)
        self._ready_to_running_id_node = None #Transition property (TransitionNumber value 2)
        self._running_to_halted_id_node = None #Transition property (TransitionNumber value 3)
        self._running_to_ready_id_node = None #Transition property (TransitionNumber value 4)
        self._running_to_suspended_id_node = None #Transition property (TransitionNumber value 5)
        self._suspended_to_running_id_node = None #Transition property (TransitionNumber value 6)
        self._suspended_to_halted_id_node = None #Transition property (TransitionNumber value 7)
        self._suspended_to_ready_id_node = None #Transition property (TransitionNumber value 8)
        self._ready_to_halted_id_node = None #Transition property (TransitionNumber value 9)

        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.localizedtext_ready = ua.LocalizedText("Ready", "en-US")
        self.localizedtext_running = ua.LocalizedText("Running", "en-US")
        self.localizedtext_halted = ua.LocalizedText("Halted", "en-US")
        self.localizedtext_suspended= ua.LocalizedText("Suspended", "en-US")
        self.localizedtext_halted_to_ready = ua.LocalizedText("HaltedToReady", "en-US")
        self.localizedtext_ready_to_running = ua.LocalizedText("ReadyToRunning", "en-US")
        self.localizedtext_running_to_halted = ua.LocalizedText("RunningToHalted", "en-US")
        self.localizedtext_running_to_ready = ua.LocalizedText("RunningToReady", "en-US")
        self.localizedtext_running_to_suspended = ua.LocalizedText("RunningToSuspended", "en-US")
        self.localizedtext_suspended_to_running = ua.LocalizedText("SuspendedToRunning", "en-US")
        self.localizedtext_suspended_to_halted = ua.LocalizedText("SuspendedToHalted", "en-US")
        self.localizedtext_suspended_to_ready = ua.LocalizedText("SuspendedToReady", "en-US")
        self.localizedtext_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:

    #Transition
    async def HaltedToReady(self):
        await self._current_state.write_value(
            self.localizedtext_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.localizedtext_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/AuditProgramTransitionEventType (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.localizedtext_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.localizedtext_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/AuditProgramTransitionEventType (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.localizedtext_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.localizedtext_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/AuditProgramTransitionEventType (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.localizedtext_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.localizedtext_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/AuditProgramTransitionEventType (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.localizedtext_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.localizedtext_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/AuditProgramTransitionEventType (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.localizedtext_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.localizedtext_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/AuditProgramTransitionEventType (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.localizedtext_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.localizedtext_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/AuditProgramTransitionEventType (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.localizedtext_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.localizedtext_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/AuditProgramTransitionEventType (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.localizedtext_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.localizedtext_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/AuditProgramTransitionEventType (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.localizedtext_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.localizedtext_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.localizedtext_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.localizedtext_ready:
            return await ReadyToHalted()

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

        elif await self._current_state.read_value() == self.localizedtext_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.localizedtext_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, idx, name)
        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, idx, name)
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, idx, name)
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_node = None #State node
140			self._halted_state_node = None #State node
141			self._running_state_node = None #State node
142			self._suspended_state_node = None #State node
143
144			self._halted_to_ready_node = None #Transition node
145			self._ready_to_running_node = None #Transition node
146			self._running_to_halted_node = None #Transition node
147			self._running_to_ready_node = None #Transition node
148			self._running_to_suspended_node = None #Transition node
149			self._suspended_to_running_node = None #Transition node
150			self._suspended_to_halted_node = None #Transition node
151			self._suspended_to_ready_node = None #Transition node
152			self._ready_to_halted_node = None #Transition node
153
154			# 5.2.3.2 ProgramStateMachineType states
155			self._halted_state_id_node = None #State property (StateNumber value 11)
156			self._ready_state_id_node = None #State property (StateNumber value 12)
157			self._running_state_id_node = None #State property (StateNumber value 13)
158			self._suspended_state_id_node = None #State property (StateNumber value 14)
159
160			# 5.2.3.3 ProgramStateMachineType transitions
161			self._halted_to_ready_id_node = None #Transition property (TransitionNumber value 1)
162			self._ready_to_running_id_node = None #Transition property (TransitionNumber value 2)
163			self._running_to_halted_id_node = None #Transition property (TransitionNumber value 3)
164			self._running_to_ready_id_node = None #Transition property (TransitionNumber value 4)
165			self._running_to_suspended_id_node = None #Transition property (TransitionNumber value 5)
166			self._suspended_to_running_id_node = None #Transition property (TransitionNumber value 6)
167			self._suspended_to_halted_id_node = None #Transition property (TransitionNumber value 7)
168			self._suspended_to_ready_id_node = None #Transition property (TransitionNumber value 8)
169			self._ready_to_halted_id_node = None #Transition property (TransitionNumber value 9)
170
171			self._halt_method_node = None #uamethod node
172			self._reset_method_node = None #uamethod node
173			self._resume_method_node = None #uamethod node
174			self._start_method_node = None #uamethod node
175			self._suspend_method_node = None #uamethod node
176
177			self.localizedtext_ready = ua.LocalizedText("Ready", "en-US")
178			self.localizedtext_running = ua.LocalizedText("Running", "en-US")
179			self.localizedtext_halted = ua.LocalizedText("Halted", "en-US")
180			self.localizedtext_suspended= ua.LocalizedText("Suspended", "en-US")
181			self.localizedtext_halted_to_ready = ua.LocalizedText("HaltedToReady", "en-US")
182			self.localizedtext_ready_to_running = ua.LocalizedText("ReadyToRunning", "en-US")
183			self.localizedtext_running_to_halted = ua.LocalizedText("RunningToHalted", "en-US")
184			self.localizedtext_running_to_ready = ua.LocalizedText("RunningToReady", "en-US")
185			self.localizedtext_running_to_suspended = ua.LocalizedText("RunningToSuspended", "en-US")
186			self.localizedtext_suspended_to_running = ua.LocalizedText("SuspendedToRunning", "en-US")
187			self.localizedtext_suspended_to_halted = ua.LocalizedText("SuspendedToHalted", "en-US")
188			self.localizedtext_suspended_to_ready = ua.LocalizedText("SuspendedToReady", "en-US")
189			self.localizedtext_ready_to_halted = ua.LocalizedText("ReadyToHalted", "en-US")
190
191			async def install(self, optionals=False):
192			'''
193			setup adressspace and initialize
194			'''
195			self._optionals = optionals
196			self._state_machine_node = await self._parent.add_object(
197			self._idx,
198			self._name,
199			objecttype=self._state_machine_type,
200			instantiate_optional=optionals
201			)
202			#FIXME get childnodes:
203
204			#Transition
205			async def HaltedToReady(self):
206			await self._current_state.write_value(
207			self.localizedtext_ready,
208			varianttype=ua.VariantType.LocalizedText
209			)
210			await self._current_state_id.write_value(
211			self._ready_state.nodeid,
212			varianttype=ua.VariantType.NodeId
213			)
214			await self._last_transition.write_value(
215			self.localizedtext_halted_to_ready,
216			varianttype=ua.VariantType.LocalizedText
217			)
218			await self._last_transition_id.write_value(
219			self._halted_to_ready.nodeid,
220			varianttype=ua.VariantType.NodeId
221			)
222			#FIXME
223			# trigger ProgramTransitionEventType and
224			# AuditUpdateMethodEvents/AuditProgramTransitionEventType (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
225			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
226
227			#Transition
228			async def ReadyToRunning(self):
229			await self._current_state.write_value(
230			self.localizedtext_running,
231			varianttype=ua.VariantType.LocalizedText
232			)
233			await self._current_state_id.write_value(
234			self._running_state.nodeid,
235			varianttype=ua.VariantType.NodeId
236			)
237			await self._last_transition.write_value(
238			self.localizedtext_ready_to_running,
239			varianttype=ua.VariantType.LocalizedText
240			)
241			await self._last_transition_id.write_value(
242			self._ready_to_running.nodeid,
243			varianttype=ua.VariantType.NodeId
244			)
245			#FIXME
246			# trigger ProgramTransitionEventType and
247			# AuditUpdateMethodEvents/AuditProgramTransitionEventType (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
248			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
249
250			#Transition
251			async def RunningToHalted(self):
252			await self._current_state.write_value(
253			self.localizedtext_halted,
254			varianttype=ua.VariantType.LocalizedText
255			)
256			await self._current_state_id.write_value(
257			self._halted_state.nodeid,
258			varianttype=ua.VariantType.NodeId
259			)
260			await self._last_transition.write_value(
261			self.localizedtext_running_to_halted,
262			varianttype=ua.VariantType.LocalizedText
263			)
264			await self._last_transition_id.write_value(
265			self._running_to_halted.nodeid,
266			varianttype=ua.VariantType.NodeId
267			)
268			#FIXME
269			# trigger ProgramTransitionEventType and
270			# AuditUpdateMethodEvents/AuditProgramTransitionEventType (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
271			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
272
273			#Transition
274			async def RunningToReady(self):
275			await self._current_state.write_value(
276			self.localizedtext_ready,
277			varianttype=ua.VariantType.LocalizedText
278			)
279			await self._current_state_id.write_value(
280			self._ready_state.nodeid,
281			varianttype=ua.VariantType.NodeId
282			)
283			await self._last_transition.write_value(
284			self.localizedtext_running_to_ready,
285			varianttype=ua.VariantType.LocalizedText
286			)
287			await self._last_transition_id.write_value(
288			self._running_to_ready.nodeid,
289			varianttype=ua.VariantType.NodeId
290			)
291			#FIXME
292			# trigger ProgramTransitionEventType and
293			# AuditUpdateMethodEvents/AuditProgramTransitionEventType (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
294			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
295
296			#Transition
297			async def RunningToSuspended(self):
298			await self._current_state.write_value(
299			self.localizedtext_suspended,
300			varianttype=ua.VariantType.LocalizedText
301			)
302			await self._current_state_id.write_value(
303			self._suspended_state.nodeid,
304			varianttype=ua.VariantType.NodeId
305			)
306			await self._last_transition.write_value(
307			self.localizedtext_running_to_suspended,
308			varianttype=ua.VariantType.LocalizedText
309			)
310			await self._last_transition_id.write_value(
311			self._running_to_suspended.nodeid,
312			varianttype=ua.VariantType.NodeId
313			)
314			#FIXME
315			# trigger ProgramTransitionEventType and
316			# AuditUpdateMethodEvents/AuditProgramTransitionEventType (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
317			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
318
319			#Transition
320			async def SuspendedToRunning(self):
321			await self._current_state.write_value(
322			self.localizedtext_running,
323			varianttype=ua.VariantType.LocalizedText
324			)
325			await self._current_state_id.write_value(
326			self._running_state.nodeid,
327			varianttype=ua.VariantType.NodeId
328			)
329			await self._last_transition.write_value(
330			self.localizedtext_suspended_to_running,
331			varianttype=ua.VariantType.LocalizedText
332			)
333			await self._last_transition_id.write_value(
334			self._suspended_to_running.nodeid,
335			varianttype=ua.VariantType.NodeId
336			)
337			#FIXME
338			# trigger ProgramTransitionEventType and
339			# AuditUpdateMethodEvents/AuditProgramTransitionEventType (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
340			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
341
342			#Transition
343			async def SuspendedToHalted(self):
344			await self._current_state.write_value(
345			self.localizedtext_halted,
346			varianttype=ua.VariantType.LocalizedText
347			)
348			await self._current_state_id.write_value(
349			self._halted_state.nodeid,
350			varianttype=ua.VariantType.NodeId
351			)
352			await self._last_transition.write_value(
353			self.localizedtext_suspended_to_halted,
354			varianttype=ua.VariantType.LocalizedText
355			)
356			await self._last_transition_id.write_value(
357			self._suspended_to_halted.nodeid,
358			varianttype=ua.VariantType.NodeId
359			)
360			#FIXME
361			# trigger ProgramTransitionEventType and
362			# AuditUpdateMethodEvents/AuditProgramTransitionEventType (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
363			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
364
365			#Transition
366			async def SuspendedToReady(self):
367			await self._current_state.write_value(
368			self.localizedtext_ready,
369			varianttype=ua.VariantType.LocalizedText
370			)
371			await self._current_state_id.write_value(
372			self._ready_state.nodeid,
373			varianttype=ua.VariantType.NodeId
374			)
375			await self._last_transition.write_value(
376			self.localizedtext_suspended_to_ready,
377			varianttype=ua.VariantType.LocalizedText
378			)
379			await self._last_transition_id.write_value(
380			self._suspended_to_ready.nodeid,
381			varianttype=ua.VariantType.NodeId
382			)
383			#FIXME
384			# trigger ProgramTransitionEventType and
385			# AuditUpdateMethodEvents/AuditProgramTransitionEventType (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
386			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
387
388			#Transition
389			async def ReadyToHalted(self):
390			await self._current_state.write_value(
391			self.localizedtext_halted,
392			varianttype=ua.VariantType.LocalizedText
393			)
394			await self._current_state_id.write_value(
395			self._halted_state.nodeid,
396			varianttype=ua.VariantType.NodeId
397			)
398			await self._last_transition.write_value(
399			self.localizedtext_ready_to_halted,
400			varianttype=ua.VariantType.LocalizedText
401			)
402			await self._last_transition_id.write_value(
403			self._ready_to_halted.nodeid,
404			varianttype=ua.VariantType.NodeId
405			)
406			#FIXME
407			# trigger ProgramTransitionEventType and
408			# AuditUpdateMethodEvents/AuditProgramTransitionEventType (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
409			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
410
411			#method to be linked to uamethod
412			async def Start(self):
413			if await self._current_state.read_value() == self.localizedtext_ready:
414			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...
415			else:
416			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
417
418			#method to be linked to uamethod
419			async def Suspend(self):
420			if await self._current_state.read_value() == self.localizedtext_running:
421			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...
422			else:
423			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
424
425			#method to be linked to uamethod
426			async def Resume(self):
427			if await self._current_state.read_value() == self.localizedtext_suspended:
428			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...
429			else:
430			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
431
432			#method to be linked to uamethod
433			async def Halt(self):
434			if await self._current_state.read_value() == self.localizedtext_ready:
435			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...
436			elif await self._current_state.read_value() == self.localizedtext_running:
437			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...
438			elif await self._current_state.read_value() == self.localizedtext_suspended:
439			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...
440			else:
441			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
442
443			#method to be linked to uamethod
444			async def Reset(self):
445			if await self._current_state.read_value() == self.localizedtext_halted:
446			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...
447			else:
448			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
449
450			class ShelvedStateMachineTypeClass(FiniteStateMachineTypeClass):
451			'''
452			NOT IMPLEMENTED "ShelvedStateMachineType"
453			'''
454			def __init__(self, server=None, parent=None, idx=None, name=None):
455			super().__init__(server, parent, idx, name)
456			if name == None:
457			name = "ShelvedStateMachine"
458			self._state_machine_type = ua.NodeId(2929, 0)
459			raise NotImplementedError
460
461
462			#Devtest / Workbench
463			if __name__ == "__main__":
464			async def main():
465			logging.basicConfig(level=logging.INFO)
466			_logger = logging.getLogger('asyncua')
467
468			server = Server()
469			await server.init()
470
471			sm = StateMachineTypeClass(server, server.nodes.objects, 0, "StateMachine")
472			await sm.install(True)
473			await sm.change_state(ua.LocalizedText("Test", "en-US"), server.get_node("ns=0;i=2253").nodeid)
474			fsm = FiniteStateMachineTypeClass(server, server.nodes.objects, 0, "FiniteStateMachine")
475			await fsm.install(True)
476			pfsm = ProgramStateMachineTypeClass(server, server.nodes.objects, 0, "ProgramStateMachine")
477			await pfsm.install(True)
478
479			async with server:
480			while 1:
481			await asyncio.sleep(0)
482
483			asyncio.run(main())
484

FreeOpcUa / opcua-asyncio

Pull Request — master (#348)

ProgramStateMachineTypeClass.ReadyToRunning() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like