asyncua.common.statemachine.StateMachineTypeClass.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:11 UTC

StateMachineTypeClass.install() A

↳ Parent: asyncua.common.statemachine

Complexity

Conditions

Size

Total Lines	19
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	12
nop	2
dl	0
loc	19
rs	9.8
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 child 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)

    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 event
        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 event
        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 event
        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 event
        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 event
        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 event
        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 event
        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 event
        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 event
        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


#Devtests
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)

if __name__ == "__main__":
    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 child 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
67			async def write_state(self, state_name, state_node):
68			#FIXME check types/class
69			await self._current_state_node.write_value(state_name)
70			await self._current_state_id_node.write_value(state_node)
71
72			async def write_transition(self, transition_name, transition_node):
73			#FIXME check types/class
74			await self._last_transition_node.write_value(transition_name)
75			await self._last_transition_id_node.write_value(transition_node)
76
77			class FiniteStateMachineTypeClass(StateMachineTypeClass):
78			'''
79			Implementation of an FiniteStateMachineType
80			'''
81			def __init__(self, server=None, parent=None, idx=None, name=None):
82			super().__init__(server, parent, idx, name)
83			if name == None:
84			name = "FiniteStateMachine"
85			self._state_machine_type = ua.NodeId(2771, 0)
86			self._avalible_states = []
87			self._avalible_transitions = []
88
89			async def install(self, optionals=False):
90			'''
91			setup adressspace and initialize
92			'''
93			self._optionals = optionals
94			self._state_machine_node = await self._parent.add_object(
95			self._idx,
96			self._name,
97			objecttype=self._state_machine_type,
98			instantiate_optional=optionals
99			)
100
101			async def set_avalible_states(self, states):
102			self._avalible_states = states
103
104			async def set_avalible_transitions(self, transitions):
105			self._avalible_transitions = transitions
106
107			class ExclusiveLimitStateMachineTypeClass(FiniteStateMachineTypeClass):
108			'''
109			NOT IMPLEMENTED "ExclusiveLimitStateMachineType"
110			'''
111			def __init__(self, server=None, parent=None, idx=None, name=None):
112			super().__init__(server, parent)
113			if name == None:
114			name = "ExclusiveLimitStateMachine"
115			self._state_machine_type = ua.NodeId(9318, 0)
116			raise NotImplementedError
117
118			class FileTransferStateMachineTypeClass(FiniteStateMachineTypeClass):
119			'''
120			NOT IMPLEMENTED "FileTransferStateMachineType"
121			'''
122			def __init__(self, server=None, parent=None, idx=None, name=None):
123			super().__init__(server, parent)
124			if name == None:
125			name = "FileTransferStateMachine"
126			self._state_machine_type = ua.NodeId(15803, 0)
127			raise NotImplementedError
128
129			class ProgramStateMachineTypeClass(FiniteStateMachineTypeClass):
130			'''
131			https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.3/
132			'''
133			def __init__(self, server=None, parent=None, idx=None, name=None):
134			super().__init__(server, parent, idx, name)
135			if name == None:
136			name = "ProgramStateMachine"
137			self._state_machine_type = ua.NodeId(2391, 0)
138			self._ready_state = None #State node
139			self._halted_state = None #State node
140			self._running_state = None #State node
141			self._suspended_state = None #State node
142
143			self._halted_to_ready = None #Transition node
144			self._ready_to_running = None #Transition node
145			self._running_to_halted = None #Transition node
146			self._running_to_ready = None #Transition node
147			self._running_to_suspended = None #Transition node
148			self._suspended_to_running = None #Transition node
149			self._suspended_to_halted = None #Transition node
150			self._suspended_to_ready = None #Transition node
151			self._ready_to_halted = None #Transition node
152
153			self._halt_method_node = None #uamethod node
154			self._reset_method_node = None #uamethod node
155			self._resume_method_node = None #uamethod node
156			self._start_method_node = None #uamethod node
157			self._suspend_method_node = None #uamethod node
158
159			self.lt_ready = ua.LocalizedText("Ready", "en-US")
160			self.lt_running = ua.LocalizedText("Running", "en-US")
161			self.lt_halted = ua.LocalizedText("Halted", "en-US")
162			self.lt_suspended= ua.LocalizedText("Suspended", "en-US")
163			self.lt_halted_to_ready = ua.LocalizedText("HaltedToReady", "en-US")
164			self.lt_ready_to_running = ua.LocalizedText("ReadyToRunning", "en-US")
165			self.lt_running_to_halted = ua.LocalizedText("RunningToHalted", "en-US")
166			self.lt_running_to_ready = ua.LocalizedText("RunningToReady", "en-US")
167			self.lt_running_to_suspended = ua.LocalizedText("RunningToSuspended", "en-US")
168			self.lt_suspended_to_running = ua.LocalizedText("SuspendedToRunning", "en-US")
169			self.lt_suspended_to_halted = ua.LocalizedText("SuspendedToHalted", "en-US")
170			self.lt_suspended_to_ready = ua.LocalizedText("SuspendedToReady", "en-US")
171			self.lt_ready_to_halted = ua.LocalizedText("ReadyToHalted", "en-US")
172
173			async def install(self, optionals=False):
174			'''
175			setup adressspace and initialize
176			'''
177			self._optionals = optionals
178			self._state_machine_node = await self._parent.add_object(
179			self._idx,
180			self._name,
181			objecttype=self._state_machine_type,
182			instantiate_optional=optionals
183			)
184			#FIXME get childnodes:
185			self._ready_state = None #State node
186			self._halted_state = None #State node
187			self._running_state = None #State node
188			self._suspended_state = None #State node
189			self._halted_to_ready = None #Transition node
190			self._ready_to_running = None #Transition node
191			self._running_to_halted = None #Transition node
192			self._running_to_ready = None #Transition node
193			self._running_to_suspended = None #Transition node
194			self._suspended_to_running = None #Transition node
195			self._suspended_to_halted = None #Transition node
196			self._suspended_to_ready = None #Transition node
197			self._ready_to_halted = None #Transition node
198			self._halt_method_node = None #uamethod node
199			self._reset_method_node = None #uamethod node
200			self._resume_method_node = None #uamethod node
201			self._start_method_node = None #uamethod node
202			self._suspend_method_node = None #uamethod node
203
204			#Transition
205			async def HaltedToReady(self):
206			await self._current_state.write_value(
207			self.lt_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.lt_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 trigger event
223			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
224
225			#Transition
226			async def ReadyToRunning(self):
227			await self._current_state.write_value(
228			self.lt_running,
229			varianttype=ua.VariantType.LocalizedText
230			)
231			await self._current_state_id.write_value(
232			self._running_state.nodeid,
233			varianttype=ua.VariantType.NodeId
234			)
235			await self._last_transition.write_value(
236			self.lt_ready_to_running,
237			varianttype=ua.VariantType.LocalizedText
238			)
239			await self._last_transition_id.write_value(
240			self._ready_to_running.nodeid,
241			varianttype=ua.VariantType.NodeId
242			)
243			#FIXME trigger event
244			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
245
246			#Transition
247			async def RunningToHalted(self):
248			await self._current_state.write_value(
249			self.lt_halted,
250			varianttype=ua.VariantType.LocalizedText
251			)
252			await self._current_state_id.write_value(
253			self._halted_state.nodeid,
254			varianttype=ua.VariantType.NodeId
255			)
256			await self._last_transition.write_value(
257			self.lt_running_to_halted,
258			varianttype=ua.VariantType.LocalizedText
259			)
260			await self._last_transition_id.write_value(
261			self._running_to_halted.nodeid,
262			varianttype=ua.VariantType.NodeId
263			)
264			#FIXME trigger event
265			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
266
267			#Transition
268			async def RunningToReady(self):
269			await self._current_state.write_value(
270			self.lt_ready,
271			varianttype=ua.VariantType.LocalizedText
272			)
273			await self._current_state_id.write_value(
274			self._ready_state.nodeid,
275			varianttype=ua.VariantType.NodeId
276			)
277			await self._last_transition.write_value(
278			self.lt_running_to_ready,
279			varianttype=ua.VariantType.LocalizedText
280			)
281			await self._last_transition_id.write_value(
282			self._running_to_ready.nodeid,
283			varianttype=ua.VariantType.NodeId
284			)
285			#FIXME trigger event
286			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
287
288			#Transition
289			async def RunningToSuspended(self):
290			await self._current_state.write_value(
291			self.lt_suspended,
292			varianttype=ua.VariantType.LocalizedText
293			)
294			await self._current_state_id.write_value(
295			self._suspended_state.nodeid,
296			varianttype=ua.VariantType.NodeId
297			)
298			await self._last_transition.write_value(
299			self.lt_running_to_suspended,
300			varianttype=ua.VariantType.LocalizedText
301			)
302			await self._last_transition_id.write_value(
303			self._running_to_suspended.nodeid,
304			varianttype=ua.VariantType.NodeId
305			)
306			#FIXME trigger event
307			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
308
309			#Transition
310			async def SuspendedToRunning(self):
311			await self._current_state.write_value(
312			self.lt_running,
313			varianttype=ua.VariantType.LocalizedText
314			)
315			await self._current_state_id.write_value(
316			self._running_state.nodeid,
317			varianttype=ua.VariantType.NodeId
318			)
319			await self._last_transition.write_value(
320			self.lt_suspended_to_running,
321			varianttype=ua.VariantType.LocalizedText
322			)
323			await self._last_transition_id.write_value(
324			self._suspended_to_running.nodeid,
325			varianttype=ua.VariantType.NodeId
326			)
327			#FIXME trigger event
328			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
329
330			#Transition
331			async def SuspendedToHalted(self):
332			await self._current_state.write_value(
333			self.lt_halted,
334			varianttype=ua.VariantType.LocalizedText
335			)
336			await self._current_state_id.write_value(
337			self._halted_state.nodeid,
338			varianttype=ua.VariantType.NodeId
339			)
340			await self._last_transition.write_value(
341			self.lt_suspended_to_halted,
342			varianttype=ua.VariantType.LocalizedText
343			)
344			await self._last_transition_id.write_value(
345			self._suspended_to_halted.nodeid,
346			varianttype=ua.VariantType.NodeId
347			)
348			#FIXME trigger event
349			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
350
351			#Transition
352			async def SuspendedToReady(self):
353			await self._current_state.write_value(
354			self.lt_ready,
355			varianttype=ua.VariantType.LocalizedText
356			)
357			await self._current_state_id.write_value(
358			self._ready_state.nodeid,
359			varianttype=ua.VariantType.NodeId
360			)
361			await self._last_transition.write_value(
362			self.lt_suspended_to_ready,
363			varianttype=ua.VariantType.LocalizedText
364			)
365			await self._last_transition_id.write_value(
366			self._suspended_to_ready.nodeid,
367			varianttype=ua.VariantType.NodeId
368			)
369			#FIXME trigger event
370			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
371
372			#Transition
373			async def ReadyToHalted(self):
374			await self._current_state.write_value(
375			self.lt_halted,
376			varianttype=ua.VariantType.LocalizedText
377			)
378			await self._current_state_id.write_value(
379			self._halted_state.nodeid,
380			varianttype=ua.VariantType.NodeId
381			)
382			await self._last_transition.write_value(
383			self.lt_ready_to_halted,
384			varianttype=ua.VariantType.LocalizedText
385			)
386			await self._last_transition_id.write_value(
387			self._ready_to_halted.nodeid,
388			varianttype=ua.VariantType.NodeId
389			)
390			#FIXME trigger event
391			return ua.StatusCode(ua.status_codes.StatusCodes.Good)
392
393			#method to be linked to uamethod
394			async def Start(self):
395			if await self._current_state.read_value() == self.lt_ready:
396			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...
397			else:
398			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
399
400			#method to be linked to uamethod
401			async def Suspend(self):
402			if await self._current_state.read_value() == self.lt_running:
403			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...
404			else:
405			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
406
407			#method to be linked to uamethod
408			async def Resume(self):
409			if await self._current_state.read_value() == self.lt_suspended:
410			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...
411			else:
412			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
413
414			#method to be linked to uamethod
415			async def Halt(self):
416			if await self._current_state.read_value() == self.lt_ready:
417			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...
418			elif await self._current_state.read_value() == self.lt_running:
419			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...
420			elif await self._current_state.read_value() == self.lt_suspended:
421			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...
422			else:
423			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
424
425			#method to be linked to uamethod
426			async def Reset(self):
427			if await self._current_state.read_value() == self.lt_halted:
428			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...
429			else:
430			return ua.StatusCode(ua.status_codes.StatusCodes.BadNotExecutable)
431
432			class ShelvedStateMachineTypeClass(FiniteStateMachineTypeClass):
433			'''
434			NOT IMPLEMENTED "ShelvedStateMachineType"
435			'''
436			def __init__(self, server=None, parent=None, idx=None, name=None):
437			super().__init__(server, parent)
438			if name == None:
439			name = "ShelvedStateMachine"
440			self._state_machine_type = ua.NodeId(2929, 0)
441			raise NotImplementedError
442
443
444			#Devtests
445			async def main():
446			logging.basicConfig(level=logging.INFO)
447			_logger = logging.getLogger('asyncua')
448
449			server = Server()
450			await server.init()
451
452			sm = StateMachineTypeClass(server, server.nodes.objects, 0, "StateMachine")
453			await sm.install(True)
454			await sm.change_state(ua.LocalizedText("Test", "en-US"), server.get_node("ns=0;i=2253").nodeid)
455			fsm = FiniteStateMachineTypeClass(server, server.nodes.objects, 0, "FiniteStateMachine")
456			await fsm.install(True)
457			pfsm = ProgramStateMachineTypeClass(server, server.nodes.objects, 0, "ProgramStateMachine")
458			await pfsm.install(True)
459
460			async with server:
461			while 1:
462			await asyncio.sleep(0)
463
464			if __name__ == "__main__":
465			asyncio.run(main())
466

FreeOpcUa / opcua-asyncio

Pull Request — master (#348)

StateMachineTypeClass.install() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like