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

Completed

Pull Request — master (#348)

unknown

created 2020-11-27 10:14 UTC

FiniteStateMachineTypeClass.install() A

↳ Parent: asyncua.common.statemachine

Complexity

Conditions

Size

Total Lines	15
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	9
nop	4
dl	0
loc	15
rs	9.95
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)
    
    async def add_state(self, name, state_type, optionals):
        #FIXME check types/class
        return await self._state_machine_node.add_object(
            self._idx, 
            name, 
            objecttype=state_type, 
            instantiate_optional=optionals
            )

    async def add_transition(self, name, transition_type, optionals):
        #FIXME check types/class
        return await self._state_machine_node.add_object(
            self._idx, 
            name, 
            objecttype=transition_type, 
            instantiate_optional=optionals
            )

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_node = None
        self._avalible_transitions_node = None

    async def install(self, avalible_states, avalible_transitions, 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 children
        # map children
        await self.set_avalible_states(avalible_states)
        await self.set_avalible_transitions(avalible_transitions)

    async def set_avalible_states(self, states):
        #check if its list
        await self._avalible_states_node.write_value(states, varianttype=ua.VariantType.NodeId)

    async def set_avalible_transitions(self, transitions):
        await self._avalible_transitions_node.write_value(transitions, varianttype=ua.VariantType.NodeId)

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			async def add_state(self, name, state_type, optionals):
79			#FIXME check types/class
80			return await self._state_machine_node.add_object(
81			self._idx,
82			name,
83			objecttype=state_type,
84			instantiate_optional=optionals
85			)
86
87			async def add_transition(self, name, transition_type, optionals):
88			#FIXME check types/class
89			return await self._state_machine_node.add_object(
90			self._idx,
91			name,
92			objecttype=transition_type,
93			instantiate_optional=optionals
94			)
95
96			class FiniteStateMachineTypeClass(StateMachineTypeClass):
97			'''
98			Implementation of an FiniteStateMachineType
99			'''
100			def __init__(self, server=None, parent=None, idx=None, name=None):
101			super().__init__(server, parent, idx, name)
102			if name == None:
103			name = "FiniteStateMachine"
104			self._state_machine_type = ua.NodeId(2771, 0)
105			self._avalible_states_node = None
106			self._avalible_transitions_node = None
107
108			async def install(self, avalible_states, avalible_transitions, optionals=False):
109			'''
110			setup adressspace and initialize
111			'''
112			self._optionals = optionals
113			self._state_machine_node = await self._parent.add_object(
114			self._idx,
115			self._name,
116			objecttype=self._state_machine_type,
117			instantiate_optional=optionals
118			)
119			#FIXME get children
120			# map children
121			await self.set_avalible_states(avalible_states)
122			await self.set_avalible_transitions(avalible_transitions)
123
124			async def set_avalible_states(self, states):
125			#check if its list
126			await self._avalible_states_node.write_value(states, varianttype=ua.VariantType.NodeId)
127
128			async def set_avalible_transitions(self, transitions):
129			await self._avalible_transitions_node.write_value(transitions, varianttype=ua.VariantType.NodeId)
130
131			class ExclusiveLimitStateMachineTypeClass(FiniteStateMachineTypeClass):
132			'''
133			NOT IMPLEMENTED "ExclusiveLimitStateMachineType"
134			'''
135			def __init__(self, server=None, parent=None, idx=None, name=None):
136			super().__init__(server, parent, idx, name)
137			if name == None:
138			name = "ExclusiveLimitStateMachine"
139			self._state_machine_type = ua.NodeId(9318, 0)
140			raise NotImplementedError
141
142			class FileTransferStateMachineTypeClass(FiniteStateMachineTypeClass):
143			'''
144			NOT IMPLEMENTED "FileTransferStateMachineType"
145			'''
146			def __init__(self, server=None, parent=None, idx=None, name=None):
147			super().__init__(server, parent, idx, name)
148			if name == None:
149			name = "FileTransferStateMachine"
150			self._state_machine_type = ua.NodeId(15803, 0)
151			raise NotImplementedError
152
153			class ProgramStateMachineTypeClass(FiniteStateMachineTypeClass):
154			'''
155			https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.3/
156			'''
157			def __init__(self, server=None, parent=None, idx=None, name=None):
158			super().__init__(server, parent, idx, name)
159			if name == None:
160			name = "ProgramStateMachine"
161			self._state_machine_type = ua.NodeId(2391, 0)
162			self._ready_state_node = None #State node
163			self._halted_state_node = None #State node
164			self._running_state_node = None #State node
165			self._suspended_state_node = None #State node
166
167			self._halted_to_ready_node = None #Transition node
168			self._ready_to_running_node = None #Transition node
169			self._running_to_halted_node = None #Transition node
170			self._running_to_ready_node = None #Transition node
171			self._running_to_suspended_node = None #Transition node
172			self._suspended_to_running_node = None #Transition node
173			self._suspended_to_halted_node = None #Transition node
174			self._suspended_to_ready_node = None #Transition node
175			self._ready_to_halted_node = None #Transition node
176
177			# 5.2.3.2 ProgramStateMachineType states
178			self._halted_state_id_node = None #State property (StateNumber value 11)
179			self._ready_state_id_node = None #State property (StateNumber value 12)
180			self._running_state_id_node = None #State property (StateNumber value 13)
181			self._suspended_state_id_node = None #State property (StateNumber value 14)
182
183			# 5.2.3.3 ProgramStateMachineType transitions
184			self._halted_to_ready_id_node = None #Transition property (TransitionNumber value 1)
185			self._ready_to_running_id_node = None #Transition property (TransitionNumber value 2)
186			self._running_to_halted_id_node = None #Transition property (TransitionNumber value 3)
187			self._running_to_ready_id_node = None #Transition property (TransitionNumber value 4)
188			self._running_to_suspended_id_node = None #Transition property (TransitionNumber value 5)
189			self._suspended_to_running_id_node = None #Transition property (TransitionNumber value 6)
190			self._suspended_to_halted_id_node = None #Transition property (TransitionNumber value 7)
191			self._suspended_to_ready_id_node = None #Transition property (TransitionNumber value 8)
192			self._ready_to_halted_id_node = None #Transition property (TransitionNumber value 9)
193
194			self._halt_method_node = None #uamethod node
195			self._reset_method_node = None #uamethod node
196			self._resume_method_node = None #uamethod node
197			self._start_method_node = None #uamethod node
198			self._suspend_method_node = None #uamethod node
199
200			self.localizedtext_ready = ua.LocalizedText("Ready", "en-US")
201			self.localizedtext_running = ua.LocalizedText("Running", "en-US")
202			self.localizedtext_halted = ua.LocalizedText("Halted", "en-US")
203			self.localizedtext_suspended= ua.LocalizedText("Suspended", "en-US")
204			self.localizedtext_halted_to_ready = ua.LocalizedText("HaltedToReady", "en-US")
205			self.localizedtext_ready_to_running = ua.LocalizedText("ReadyToRunning", "en-US")
206			self.localizedtext_running_to_halted = ua.LocalizedText("RunningToHalted", "en-US")
207			self.localizedtext_running_to_ready = ua.LocalizedText("RunningToReady", "en-US")
208			self.localizedtext_running_to_suspended = ua.LocalizedText("RunningToSuspended", "en-US")
209			self.localizedtext_suspended_to_running = ua.LocalizedText("SuspendedToRunning", "en-US")
210			self.localizedtext_suspended_to_halted = ua.LocalizedText("SuspendedToHalted", "en-US")
211			self.localizedtext_suspended_to_ready = ua.LocalizedText("SuspendedToReady", "en-US")
212			self.localizedtext_ready_to_halted = ua.LocalizedText("ReadyToHalted", "en-US")
213
214			async def install(self, optionals=False):
215			'''
216			setup adressspace and initialize
217			'''
218			self._optionals = optionals
219			self._state_machine_node = await self._parent.add_object(
220			self._idx,
221			self._name,
222			objecttype=self._state_machine_type,
223			instantiate_optional=optionals
224			)
225			#FIXME get childnodes:
226
227			#Transition
228			async def HaltedToReady(self):
229			await self._current_state.write_value(
230			self.localizedtext_ready,
231			varianttype=ua.VariantType.LocalizedText
232			)
233			await self._current_state_id.write_value(
234			self._ready_state.nodeid,
235			varianttype=ua.VariantType.NodeId
236			)
237			await self._last_transition.write_value(
238			self.localizedtext_halted_to_ready,
239			varianttype=ua.VariantType.LocalizedText
240			)
241			await self._last_transition_id.write_value(
242			self._halted_to_ready.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 ReadyToRunning(self):
252			await self._current_state.write_value(
253			self.localizedtext_running,
254			varianttype=ua.VariantType.LocalizedText
255			)
256			await self._current_state_id.write_value(
257			self._running_state.nodeid,
258			varianttype=ua.VariantType.NodeId
259			)
260			await self._last_transition.write_value(
261			self.localizedtext_ready_to_running,
262			varianttype=ua.VariantType.LocalizedText
263			)
264			await self._last_transition_id.write_value(
265			self._ready_to_running.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 RunningToHalted(self):
275			await self._current_state.write_value(
276			self.localizedtext_halted,
277			varianttype=ua.VariantType.LocalizedText
278			)
279			await self._current_state_id.write_value(
280			self._halted_state.nodeid,
281			varianttype=ua.VariantType.NodeId
282			)
283			await self._last_transition.write_value(
284			self.localizedtext_running_to_halted,
285			varianttype=ua.VariantType.LocalizedText
286			)
287			await self._last_transition_id.write_value(
288			self._running_to_halted.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 RunningToReady(self):
298			await self._current_state.write_value(
299			self.localizedtext_ready,
300			varianttype=ua.VariantType.LocalizedText
301			)
302			await self._current_state_id.write_value(
303			self._ready_state.nodeid,
304			varianttype=ua.VariantType.NodeId
305			)
306			await self._last_transition.write_value(
307			self.localizedtext_running_to_ready,
308			varianttype=ua.VariantType.LocalizedText
309			)
310			await self._last_transition_id.write_value(
311			self._running_to_ready.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 RunningToSuspended(self):
321			await self._current_state.write_value(
322			self.localizedtext_suspended,
323			varianttype=ua.VariantType.LocalizedText
324			)
325			await self._current_state_id.write_value(
326			self._suspended_state.nodeid,
327			varianttype=ua.VariantType.NodeId
328			)
329			await self._last_transition.write_value(
330			self.localizedtext_running_to_suspended,
331			varianttype=ua.VariantType.LocalizedText
332			)
333			await self._last_transition_id.write_value(
334			self._running_to_suspended.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 SuspendedToRunning(self):
344			await self._current_state.write_value(
345			self.localizedtext_running,
346			varianttype=ua.VariantType.LocalizedText
347			)
348			await self._current_state_id.write_value(
349			self._running_state.nodeid,
350			varianttype=ua.VariantType.NodeId
351			)
352			await self._last_transition.write_value(
353			self.localizedtext_suspended_to_running,
354			varianttype=ua.VariantType.LocalizedText
355			)
356			await self._last_transition_id.write_value(
357			self._suspended_to_running.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 SuspendedToHalted(self):
367			await self._current_state.write_value(
368			self.localizedtext_halted,
369			varianttype=ua.VariantType.LocalizedText
370			)
371			await self._current_state_id.write_value(
372			self._halted_state.nodeid,
373			varianttype=ua.VariantType.NodeId
374			)
375			await self._last_transition.write_value(
376			self.localizedtext_suspended_to_halted,
377			varianttype=ua.VariantType.LocalizedText
378			)
379			await self._last_transition_id.write_value(
380			self._suspended_to_halted.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 SuspendedToReady(self):
390			await self._current_state.write_value(
391			self.localizedtext_ready,
392			varianttype=ua.VariantType.LocalizedText
393			)
394			await self._current_state_id.write_value(
395			self._ready_state.nodeid,
396			varianttype=ua.VariantType.NodeId
397			)
398			await self._last_transition.write_value(
399			self.localizedtext_suspended_to_ready,
400			varianttype=ua.VariantType.LocalizedText
401			)
402			await self._last_transition_id.write_value(
403			self._suspended_to_ready.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			#Transition
412			async def ReadyToHalted(self):
413			await self._current_state.write_value(
414			self.localizedtext_halted,
415			varianttype=ua.VariantType.LocalizedText
416			)
417			await self._current_state_id.write_value(
418			self._halted_state.nodeid,
419			varianttype=ua.VariantType.NodeId
420			)
421			await self._last_transition.write_value(
422			self.localizedtext_ready_to_halted,
423			varianttype=ua.VariantType.LocalizedText
424			)
425			await self._last_transition_id.write_value(
426			self._ready_to_halted.nodeid,
427			varianttype=ua.VariantType.NodeId
428			)
429			#FIXME
430			# trigger ProgramTransitionEventType and
431			# AuditUpdateMethodEvents/AuditProgramTransitionEventType (https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.2/)
432			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
433
434			#method to be linked to uamethod
435			async def Start(self):
436			if await self._current_state.read_value() == self.localizedtext_ready:
437			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...
438			else:
439			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
440
441			#method to be linked to uamethod
442			async def Suspend(self):
443			if await self._current_state.read_value() == self.localizedtext_running:
444			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...
445			else:
446			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
447
448			#method to be linked to uamethod
449			async def Resume(self):
450			if await self._current_state.read_value() == self.localizedtext_suspended:
451			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...
452			else:
453			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
454
455			#method to be linked to uamethod
456			async def Halt(self):
457			if await self._current_state.read_value() == self.localizedtext_ready:
458			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...
459			elif await self._current_state.read_value() == self.localizedtext_running:
460			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...
461			elif await self._current_state.read_value() == self.localizedtext_suspended:
462			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...
463			else:
464			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
465
466			#method to be linked to uamethod
467			async def Reset(self):
468			if await self._current_state.read_value() == self.localizedtext_halted:
469			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...
470			else:
471			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
472
473			class ShelvedStateMachineTypeClass(FiniteStateMachineTypeClass):
474			'''
475			NOT IMPLEMENTED "ShelvedStateMachineType"
476			'''
477			def __init__(self, server=None, parent=None, idx=None, name=None):
478			super().__init__(server, parent, idx, name)
479			if name == None:
480			name = "ShelvedStateMachine"
481			self._state_machine_type = ua.NodeId(2929, 0)
482			raise NotImplementedError
483
484
485			#Devtest / Workbench
486			if __name__ == "__main__":
487			async def main():
488			logging.basicConfig(level=logging.INFO)
489			_logger = logging.getLogger('asyncua')
490
491			server = Server()
492			await server.init()
493
494			sm = StateMachineTypeClass(server, server.nodes.objects, 0, "StateMachine")
495			await sm.install(True)
496			await sm.change_state(ua.LocalizedText("Test", "en-US"), server.get_node("ns=0;i=2253").nodeid)
497			fsm = FiniteStateMachineTypeClass(server, server.nodes.objects, 0, "FiniteStateMachine")
498			await fsm.install(True)
499			pfsm = ProgramStateMachineTypeClass(server, server.nodes.objects, 0, "ProgramStateMachine")
500			await pfsm.install(True)
501
502			async with server:
503			while 1:
504			await asyncio.sleep(0)
505
506			asyncio.run(main())
507

FreeOpcUa / opcua-asyncio

Pull Request — master (#348)

FiniteStateMachineTypeClass.install() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like