Skip to content
Snippets Groups Projects
Commit d6363294 authored by Drew Devereux's avatar Drew Devereux
Browse files

Merge branch 'pytransitions' into 'master' 

Pytransitions refactor

Closes MCCS-59

See merge request ska-telescope/lmc-base-classes!26
parents c101a6ce f5051ec8
No related branches found
No related tags found
No related merge requests found
Expand all Show whitespace changes
Inline Side-by-side
Showing
with 1421 additions and 632 deletions
.release
+ 2
− 2
View file @ d6363294
release=0.6.3
tag=lmcbaseclasses-0.6.3
release=0.6.4
tag=lmcbaseclasses-0.6.4
README.md
+ 7
− 0
View file @ d6363294
......@@ -25,6 +25,13 @@ The lmc-base-classe repository contains set of eight classes as mentioned in SKA
## Version History
#### 0.6.4
- Refactor state machine to use pytransitions library.
- Minor behavioural change: Off() command is accepted in every obsState, rather
than only EMPTY obsState.
- support `straight_to_state` shortcuts to simplify test setups
- Refactor of state machine testing to make it more portable
#### 0.6.3
- Fix omission of fatal_error transition from base device state machine.
......
docs/source/SKABaseDevice.rst
+ 0
− 8
View file @ d6363294
......@@ -6,14 +6,6 @@
SKA BaseDevice
============================================
The SKABaseDevice implements the basic device state machine, as illustrated
below, but without, at present, a Standby state.
.. image:: images/device_state_diagram.png
:width: 400
:alt: Diagram of the device state machine showing states and transitions
.. toctree::
:maxdepth: 2
......
docs/source/SKASubarray.rst
+ 0
− 7
View file @ d6363294
......@@ -6,13 +6,6 @@
SKA Subarray
============================================
The SKA Subarray device implements the observation state machine as defined
in ADR-8:
.. image:: images/ADR-8.png
:width: 400
:alt: Diagram of the observation state machine showing states and transitions
.. toctree::
:maxdepth: 2
......
docs/source/State_Machine.rst 0 → 100644
+ 69
− 0
View file @ d6363294
State Machine
=============
The state machine modules implements SKA's two fundamental state machines: the
base device state machine, and the observation state machine.
Base device state machine
-------------------------
The base device state machine provides basic state needed for all devices,
covering initialisation, off and on states, and a fault state. This state
machine is implemented by all SKA Tango devices that inherit from these LMC
base classes, though some devices with standby power modes may need to
implement further states.
.. figure:: images/device_state_diagram.png
:width: 80%
:alt: Diagram of the device state machine, taken from SKA design
documentation, showing the state machine as designed
Diagram of the device state machine, taken from SKA design
documentation, showing the state machine as designed
.. figure:: images/BaseDeviceStateMachine.png
:width: 80%
:alt: Diagram of the device state machine, automatically generated
from the state machine as specified in code.
Diagram of the device state machine, automatically generated from the
state machine as specified in code. The equivalence of this diagram to
the diagram previous demonstrates that the machine has been
implemented as designed.
Observation state machine
-------------------------
The observation state machine is implemented by devices that manage
observations (currently only subarray devices).
.. figure:: images/ADR-8.png
:width: 80%
:alt: Diagram of the observation state machine, as decided and published in ADR-8.
Diagram of the observation state machine, as decided and published in ADR-8.
.. figure:: images/ObservationStateMachine.png
:width: 80%
:alt: Diagram of the observation state machine, automatically generated from
the state machine as specified in code.
Diagram of the observation state machine, automatically generated from
the state machine as specified in code. The equivalance of this
diagram to the diagram previous demonstrates that the machine has been
implemented in conformance with ADR-8.
API
---
.. toctree::
:maxdepth: 2
.. automodule:: ska.base.state_machine
:members:
:undoc-members:
docs/source/images/BaseDeviceStateMachine.png 0 → 100644
+ 0
− 0
View file @ d6363294
docs/source/images/BaseDeviceStateMachine.png

131 KiB

docs/source/images/ObservationStateMachine.png 0 → 100644
+ 0
− 0
View file @ d6363294
docs/source/images/ObservationStateMachine.png

241 KiB

docs/source/index.rst
+ 1
− 1
View file @ d6363294
......@@ -22,6 +22,7 @@ Welcome to LMC Base Classes documentation!
SKA Control Model<Control_Model>
SKA Commands<Commands>
SKA State Machine<State_Machine>
Indices and tables
==================
......@@ -29,4 +30,3 @@ Indices and tables
* :ref:`genindex`
* :ref:`modindex`
* :ref:`search`
setup.py
+ 1
− 1
View file @ d6363294
......@@ -35,7 +35,7 @@ setuptools.setup(
],
platforms=["OS Independent"],
setup_requires=[] + pytest_runner,
install_requires=["future", "ska_logging >= 0.3.0"],
install_requires=["future", "transitions", "ska_logging >= 0.3.0"],
tests_require=["pytest", "coverage", "pytest-json-report", "pytest-forked"],
entry_points={
"console_scripts": [
......
src/ska/base/__init__.py
+ 3
− 2
View file @ d6363294
__all__ = (
"commands",
"control_model",
"state_machine",
"SKAAlarmHandler",
"SKABaseDevice", "SKABaseDeviceStateModel",
"SKACapability",
"SKALogger",
"SKAMaster",
"SKAObsDevice", "SKAObsDeviceStateModel",
"SKAObsDevice",
"SKASubarray", "SKASubarrayStateModel", "SKASubarrayResourceManager",
"SKATelState",
)
......@@ -21,7 +22,7 @@ from .master_device import SKAMaster
from .tel_state_device import SKATelState
# SKAObsDevice, and then classes that inherit from it
from .obs_device import SKAObsDevice, SKAObsDeviceStateModel
from .obs_device import SKAObsDevice
from .capability_device import SKACapability
from .subarray_device import (
SKASubarray, SKASubarrayStateModel, SKASubarrayResourceManager
......
src/ska/base/base_device.py
+ 171
− 218
View file @ d6363294
......@@ -19,7 +19,7 @@ import socket
import sys
import threading
import warnings
from transitions import MachineError
from urllib.parse import urlparse
from urllib.request import url2pathname
......@@ -35,10 +35,12 @@ from ska.base.commands import (
)
from ska.base.control_model import (
AdminMode, ControlMode, SimulationMode, TestMode, HealthState,
LoggingLevel, DeviceStateModel
LoggingLevel
)
from ska.base.faults import StateModelError
from ska.base.state_machine import BaseDeviceStateMachine
from ska.base.utils import get_groups_from_json
from ska.base.utils import get_groups_from_json, for_testing_only
from ska.base.faults import (GroupDefinitionsError,
LoggingTargetError,
LoggingLevelError)
......@@ -307,211 +309,40 @@ class LoggingUtils:
# PROTECTED REGION END # // SKABaseDevice.additionnal_import
__all__ = ["SKABaseDevice", "SKABaseDeviceStateModel", "main"]
__all__ = ["SKABaseDevice", "main"]
class SKABaseDeviceStateModel(DeviceStateModel):
class SKABaseDeviceStateModel:
"""
Implements the state model for the SKABaseDevice
"""
__transitions = {
('UNINITIALISED', 'init_started'): (
"INIT (ENABLED)",
lambda self: (
self._set_admin_mode(AdminMode.MAINTENANCE),
self._set_dev_state(DevState.INIT),
)
),
('INIT (ENABLED)', 'init_succeeded'): (
'OFF',
lambda self: self._set_dev_state(DevState.OFF)
),
('INIT (ENABLED)', 'init_failed'): (
'FAULT (ENABLED)',
lambda self: self._set_dev_state(DevState.FAULT)
),
('INIT (ENABLED)', 'fatal_error'): (
"FAULT (ENABLED)",
lambda self: self._set_dev_state(DevState.FAULT)
),
('INIT (ENABLED)', 'to_notfitted'): (
"INIT (DISABLED)",
lambda self: self._set_admin_mode(AdminMode.NOT_FITTED)
),
('INIT (ENABLED)', 'to_offline'): (
"INIT (DISABLED)",
lambda self: self._set_admin_mode(AdminMode.OFFLINE)
),
('INIT (ENABLED)', 'to_maintenance'): (
"INIT (ENABLED)",
lambda self: self._set_admin_mode(AdminMode.MAINTENANCE)
),
('INIT (ENABLED)', 'to_online'): (
"INIT (ENABLED)",
lambda self: self._set_admin_mode(AdminMode.ONLINE)
),
('INIT (DISABLED)', 'init_succeeded'): (
'DISABLED',
lambda self: self._set_dev_state(DevState.DISABLE)
),
('INIT (DISABLED)', 'init_failed'): (
'FAULT (DISABLED)',
lambda self: self._set_dev_state(DevState.FAULT)
),
('INIT (DISABLED)', 'fatal_error'): (
"FAULT (DISABLED)",
lambda self: self._set_dev_state(DevState.FAULT)
),
('INIT (DISABLED)', 'to_notfitted'): (
"INIT (DISABLED)",
lambda self: self._set_admin_mode(AdminMode.NOT_FITTED)
),
('INIT (DISABLED)', 'to_offline'): (
"INIT (DISABLED)",
lambda self: self._set_admin_mode(AdminMode.OFFLINE)
),
('INIT (DISABLED)', 'to_maintenance'): (
"INIT (ENABLED)",
lambda self: self._set_admin_mode(AdminMode.MAINTENANCE)
),
('INIT (DISABLED)', 'to_online'): (
"INIT (ENABLED)",
lambda self: self._set_admin_mode(AdminMode.ONLINE)
),
('FAULT (DISABLED)', 'reset_succeeded'): (
"DISABLED",
lambda self: self._set_dev_state(DevState.DISABLE)
),
('FAULT (DISABLED)', 'reset_failed'): ("FAULT (DISABLED)", None),
('FAULT (DISABLED)', 'fatal_error'): ("FAULT (DISABLED)", None),
('FAULT (DISABLED)', 'to_notfitted'): (
"FAULT (DISABLED)",
lambda self: self._set_admin_mode(AdminMode.NOT_FITTED)
),
('FAULT (DISABLED)', 'to_offline'): (
"FAULT (DISABLED)",
lambda self: self._set_admin_mode(AdminMode.OFFLINE)
),
('FAULT (DISABLED)', 'to_maintenance'): (
"FAULT (ENABLED)",
lambda self: self._set_admin_mode(AdminMode.MAINTENANCE)
),
('FAULT (DISABLED)', 'to_online'): (
"FAULT (ENABLED)",
lambda self: self._set_admin_mode(AdminMode.ONLINE)
),
('FAULT (ENABLED)', 'reset_succeeded'): (
"OFF",
lambda self: self._set_dev_state(DevState.OFF)
),
('FAULT (ENABLED)', 'reset_failed'): ("FAULT (ENABLED)", None),
('FAULT (ENABLED)', 'fatal_error'): ("FAULT (ENABLED)", None),
('FAULT (ENABLED)', 'to_notfitted'): (
"FAULT (DISABLED)",
lambda self: self._set_admin_mode(AdminMode.NOT_FITTED)),
('FAULT (ENABLED)', 'to_offline'): (
"FAULT (DISABLED)",
lambda self: self._set_admin_mode(AdminMode.OFFLINE)),
('FAULT (ENABLED)', 'to_maintenance'): (
"FAULT (ENABLED)",
lambda self: self._set_admin_mode(AdminMode.MAINTENANCE)
),
('FAULT (ENABLED)', 'to_online'): (
"FAULT (ENABLED)",
lambda self: self._set_admin_mode(AdminMode.ONLINE)
),
('DISABLED', 'to_offline'): (
"DISABLED",
lambda self: self._set_admin_mode(AdminMode.OFFLINE)
),
('DISABLED', 'to_online'): (
"OFF",
lambda self: (
self._set_admin_mode(AdminMode.ONLINE),
self._set_dev_state(DevState.OFF)
)
),
('DISABLED', 'to_maintenance'): (
"OFF",
lambda self: (
self._set_admin_mode(AdminMode.MAINTENANCE),
self._set_dev_state(DevState.OFF)
)
),
('DISABLED', 'to_notfitted'): (
"DISABLED",
lambda self: self._set_admin_mode(AdminMode.NOT_FITTED)
),
('DISABLED', 'fatal_error'): (
"FAULT (DISABLED)",
lambda self: self._set_dev_state(DevState.FAULT)
),
('OFF', 'to_notfitted'): (
"DISABLED",
lambda self: (
self._set_admin_mode(AdminMode.NOT_FITTED),
self._set_dev_state(DevState.DISABLE)
)
),
('OFF', 'to_offline'): (
"DISABLED", lambda self: (
self._set_admin_mode(AdminMode.OFFLINE),
self._set_dev_state(DevState.DISABLE)
)
),
('OFF', 'to_online'): (
"OFF",
lambda self: self._set_admin_mode(AdminMode.ONLINE)
),
('OFF', 'to_maintenance'): (
"OFF",
lambda self: self._set_admin_mode(AdminMode.MAINTENANCE)
),
('OFF', 'fatal_error'): (
"FAULT (ENABLED)",
lambda self: self._set_dev_state(DevState.FAULT)
),
('OFF', 'on_succeeded'): (
"ON",
lambda self: self._set_dev_state(DevState.ON)
),
('OFF', 'on_failed'): (
"FAULT (ENABLED)",
lambda self: self._set_dev_state(DevState.FAULT)
),
('ON', 'off_succeeded'): (
"OFF",
lambda self: self._set_dev_state(DevState.OFF)
),
('ON', 'off_failed'): (
"FAULT (ENABLED)",
lambda self: self._set_dev_state(DevState.FAULT)
),
('ON', 'fatal_error'): (
"FAULT (ENABLED)",
lambda self: self._set_dev_state(DevState.FAULT)
),
}
def __init__(self, dev_state_callback=None, admin_mode_callback=None):
def __init__(self, logger, op_state_callback=None, admin_mode_callback=None):
"""
Initialises the state model.
:param dev_state_callback: A callback to be called when a
transition implies a change to device state
:type dev_state_callback: callable
:param logger: the logger to be used by this state model.
:type logger: a logger that implements the standard library
logger interface
:param op_state_callback: A callback to be called when a
transition implies a change to op state
:type op_state_callback: callable
:param admin_mode_callback: A callback to be called when a
transition causes a change to device admin_mode
:type admin_mode_callback: callable
"""
super().__init__(self.__transitions, "UNINITIALISED")
self.logger = logger
self._op_state = None
self._admin_mode = None
self._op_state_callback = op_state_callback
self._admin_mode_callback = admin_mode_callback
self._dev_state = None
self._dev_state_callback = dev_state_callback
self._state_machine = BaseDeviceStateMachine(
op_state_callback=self._update_op_state,
admin_mode_callback=self._update_admin_mode
)
@property
def admin_mode(self):
......@@ -523,40 +354,124 @@ class SKABaseDeviceStateModel(DeviceStateModel):
"""
return self._admin_mode
def _set_admin_mode(self, admin_mode):
def _update_admin_mode(self, admin_mode):
"""
Helper method: calls the admin_mode callback if one exists
Helper method that updates admin_mode, ensuring that the callback is
called if one exists.
:param admin_mode: the new admin_mode value
:param admin_mode: the new adminMode attribute value
:type admin_mode: AdminMode
"""
if self._admin_mode != admin_mode:
self._admin_mode = admin_mode
if self._admin_mode_callback is not None:
self._admin_mode_callback(self._admin_mode)
self._admin_mode_callback(admin_mode)
@property
def dev_state(self):
def op_state(self):
"""
Returns the dev_state
Returns the op_state
:returns: dev_state of this state model
:returns: op_state of this state model
:rtype: tango.DevState
"""
return self._dev_state
return self._op_state
def _update_op_state(self, op_state):
"""
Helper method that updates op_state, ensuring that the callback is
called if one exists.
:param op_state: the new opState attribute value
:type op_state: tango.DevState
"""
if self._op_state != op_state:
self._op_state = op_state
if self._op_state_callback is not None:
self._op_state_callback(op_state)
def is_action_allowed(self, action):
"""
Whether a given action is allowed in the current state.
def _set_dev_state(self, dev_state):
:param action: an action, as given in the transitions table
:type action: ANY
"""
Helper method: sets this state models dev_state, and calls the
dev_state callback if one exists
return action in self._state_machine.get_triggers(self._state_machine.state)
:param dev_state: the new state value
:type admin_mode: DevState
def try_action(self, action):
"""
if self._dev_state != dev_state:
self._dev_state = dev_state
if self._dev_state_callback is not None:
self._dev_state_callback(self._dev_state)
Checks whether a given action is allowed in the current state,
and raises a StateModelError if it is not.
:param action: an action, as given in the transitions table
:type action: ANY
:raises StateModelError: if the action is not allowed in the
current state
:returns: True if the action is allowed
:rtype: boolean
"""
if not self.is_action_allowed(action):
raise StateModelError(
f"Action '{action}' not allowed in current state ({self._state_machine.state})."
)
return True
def perform_action(self, action):
"""
Performs an action on the state model
:param action: an action, as given in the transitions table
:type action: ANY
:raises StateModelError: if the action is not allowed in the
current state
"""
try:
self._state_machine.trigger(action)
except MachineError as error:
raise StateModelError(error)
@for_testing_only
def _straight_to_state(self, state):
"""
Takes the DeviceStateModel straight to the specified state. This method
exists to simplify testing; for example, if testing that a command may
be run in a given state, one can push the state model straight to that
state, rather than having to drive it to that state through a sequence
of actions. It is not intended that this method would be called outside
of test setups. A warning will be raised if it is.
Note that states are non-deterministic with respect to adminMode. For
example, in state "FAULT-DISABLED", the adminMode could be OFFLINE or
NOT_FITTED. When you drive the state machine through its transitions,
the adminMode will be set accordingly. When using this method, the
adminMode will simply be set to something sensible.
:param state: the target state
:type state: string
"""
if state == "UNINITIALISED":
pass
elif "DISABLED" in state:
if self._admin_mode not in [AdminMode.OFFLINE, AdminMode.NOT_FITTED]:
self._state_machine._update_admin_mode(AdminMode.OFFLINE)
else:
if self._admin_mode not in [AdminMode.ONLINE, AdminMode.MAINTENANCE]:
self._state_machine._update_admin_mode(AdminMode.ONLINE)
getattr(self._state_machine, f"to_{state}")()
@property
def _state(self):
"""
Returns the state of the underlying state machine. This would normally
be a hidden implementation detail, but is exposed here for testing
purposes.
"""
return self._state_machine.state
class SKABaseDevice(Device):
......@@ -578,8 +493,7 @@ class SKABaseDevice(Device):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
:type state_model: SKABaseClassStateModel or a subclass of
same
:type state_model: SKABaseDeviceStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
......@@ -830,11 +744,25 @@ class SKABaseDevice(Device):
self.set_status(f"The device is in {state} state.")
def set_state(self, state):
"""
Helper method for setting device state, ensuring that change
events are pushed.
:param state: the new state
:type state: tango.DevState
"""
super().set_state(state)
self.push_change_event('state')
self.push_archive_event('state')
def set_status(self, status):
"""
Helper method for setting device status, ensuring that change
events are pushed.
:param status: the new status
:type status: str
"""
super().set_status(status)
self.push_change_event('status')
self.push_archive_event('status')
......@@ -874,17 +802,43 @@ class SKABaseDevice(Device):
Creates the state model for the device
"""
self.state_model = SKABaseDeviceStateModel(
dev_state_callback=self._update_state,
logger=self.logger,
op_state_callback=self._update_state,
admin_mode_callback=self._update_admin_mode
)
def register_command_object(self, command_name, command_object):
"""
Registers a command object as the object to handle invocations
of a given command
:param command_name: name of the command for which the object is
being registered
:type command_name: str
:param command_object: the object that will handle invocations
of the given command
:type command_object: Command instance
"""
self._command_objects[command_name] = command_object
def get_command_object(self, command_name):
"""
Returns the command object (handler) for a given command.
:param command_name: name of the command for which a command
object (handler) is sought
:type command_name: str
:return: the registered command object (handler) for the command
:rtype: Command instance
"""
return self._command_objects[command_name]
def init_command_objects(self):
"""
Creates and registers command objects (handlers) for the
commands supported by this device.
"""
device_args = (self, self.state_model, self.logger)
self.register_command_object("On", self.OnCommand(*device_args))
......@@ -1164,8 +1118,7 @@ class SKABaseDevice(Device):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
:type state_model: SKABaseClassStateModel or a subclass of
same
:type state_model: SKABaseDeviceStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
......@@ -1237,8 +1190,7 @@ class SKABaseDevice(Device):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
:type state_model: SKABaseClassStateModel or a subclass of
same
:type state_model: SKABaseDeviceStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
......@@ -1264,6 +1216,7 @@ class SKABaseDevice(Device):
Check if command `On` is allowed in the current device state.
:raises ``tango.DevFailed``: if the command is not allowed
:return: ``True`` if the command is allowed
:rtype: boolean
"""
......@@ -1301,8 +1254,7 @@ class SKABaseDevice(Device):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
:type state_model: SKABaseClassStateModel or a subclass of
same
:type state_model: SKABaseDeviceStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
......@@ -1328,6 +1280,7 @@ class SKABaseDevice(Device):
Check if command `Off` is allowed in the current device state.
:raises ``tango.DevFailed``: if the command is not allowed
:return: ``True`` if the command is allowed
:rtype: boolean
"""
......
src/ska/base/commands.py
+ 8
− 1
View file @ d6363294
......@@ -153,7 +153,8 @@ class BaseCommand:
return self.state_model.try_action(action)
except StateModelError as exc:
raise CommandError(
f"Error executing command {self.name}") from exc
f"Error executing command {self.name}"
) from exc
def _perform_action(self, action):
"""
......@@ -167,6 +168,12 @@ class BaseCommand:
class ResponseCommand(BaseCommand):
"""
Abstract base class for a tango command handler, for commands that
execute a procedure/operation and return a (ResultCode, message)
tuple.
"""
def __call__(self, argin=None):
"""
What to do when the command is called. This base class simply
......
src/ska/base/control_model.py
+ 1
− 94
View file @ d6363294
......@@ -14,13 +14,11 @@ other useful enumerations.
"""
import enum
from ska.base.faults import StateModelError
# ---------------------------------
# Core SKA Control Model attributes
# ---------------------------------
class HealthState(enum.IntEnum):
"""Python enumerated type for ``healthState`` attribute."""
......@@ -328,94 +326,3 @@ class LoggingLevel(enum.IntEnum):
WARNING = 3
INFO = 4
DEBUG = 5
class DeviceStateModel:
"""
Base class for the state model used by SKA devices.
"""
def __init__(self, transitions, initial_state):
"""
Create a new device state model.
:param transitions: a dictionary for which each key is a (state,
event) tuple, and each value is a (state, side-effect)
tuple. When the device is in state `IN-STATE`, and action
`ACTION` is attempted, the transitions table will be checked
for an entry under key `(IN-STATE, EVENT)`. If no such key
exists, the action will be denied and a model will raise a
`StateModelError`. If the key does exist, then its value
`(OUT-STATE, SIDE-EFFECT)` will result in the model
transitioning to state `OUT-STATE`, and executing
`SIDE-EFFECT`, which must be a function or lambda that
takes a single parameter - a model instance.
:type transitions: dict
:param initial_state: the starting state of the model
:type initial_state: a state with an entry in the transitions
table
"""
# instances may update transitions dict, so need a copy
self._transitions = transitions.copy()
self._state = initial_state
@property
def state(self):
"""Return current state as a string."""
return self._state
def update_transitions(self, transitions):
"""
Update the transitions table with new transitions.
:param transitions: new transitions to be included in the
transitions table. Transitions with pre-existing keys will
replace the transitions for that key. Transitions with novel
keys will be added. There is no provision for removing
transitions
:type transitions: dict
"""
self._transitions.update(transitions)
def is_action_allowed(self, action):
"""
Whether a given action is allowed in the current state.
:param action: an action, as given in the transitions table
:type action: ANY
"""
return (self._state, action) in self._transitions
def try_action(self, action):
"""
Checks whether a given action is allowed in the current state,
and raises a StateModelError if it is not.
:param action: an action, as given in the transitions table
:type action: ANY
:raises StateModelError: if the action is not allowed in the
current state
:returns: True if the action is allowed
:rtype: boolean
"""
if not self.is_action_allowed(action):
raise StateModelError(
f"Action '{action}' not allowed in current state ({self._state})."
)
return True
def perform_action(self, action):
"""
Performs an action on the state model
:param action: an action, as given in the transitions table
:type action: ANY
:raises StateModelError: if the action is not allowed in the
current state
"""
self.try_action(action)
(self._state, side_effect) = self._transitions[(self._state, action)]
if side_effect is not None:
side_effect(self)
src/ska/base/obs_device.py
+ 16
− 77
View file @ d6363294
......@@ -14,78 +14,15 @@ instead of just SKABaseDevice.
# Additional import
# PROTECTED REGION ID(SKAObsDevice.additionnal_import) ENABLED START #
# Tango imports
from tango import DevState
from tango.server import run, attribute
# SKA specific imports
from ska.base import SKABaseDevice, SKABaseDeviceStateModel
from ska.base import SKABaseDevice
from ska.base.commands import ResultCode
from ska.base.control_model import AdminMode, ObsMode, ObsState
from ska.base.control_model import ObsMode, ObsState
# PROTECTED REGION END # // SKAObsDevice.additionnal_imports
__all__ = ["SKAObsDevice", "SKAObsDeviceStateModel", "main"]
class SKAObsDeviceStateModel(SKABaseDeviceStateModel):
"""
Implements the state model for the SKABaseDevice
"""
def __init__(
self,
dev_state_callback=None,
admin_mode_callback=None,
obs_state_callback=None
):
"""
Initialises the model. Note that this does not imply moving to
INIT state. The INIT state is managed by the model itself.
:param dev_state_callback: A callback to be called when a
transition implies a change to device state
:type dev_state_callback: callable
:param admin_mode_callback: A callback to be called when a
transition causes a change to device admin_mode
:type admin_mode_callback: callable
:param obs_state_callback: A callback to be called when a
transition causes a change to device obs_state
:type obs_state_callback: callable
"""
super().__init__(
dev_state_callback=dev_state_callback,
admin_mode_callback=admin_mode_callback
)
self._obs_state_callback = obs_state_callback
self.update_transitions(
{
('UNINITIALISED', 'init_started'): (
"INIT (ENABLED)",
lambda self: (
self._set_admin_mode(AdminMode.MAINTENANCE),
self._set_dev_state(DevState.INIT),
self._set_obs_state(ObsState.EMPTY)
)
)
}
)
self._obs_state = None
def _set_obs_state(self, obs_state):
"""
Helper method: set the value of obs_state value, and calls the
obs_state_callback if one exists.
:param obs_state: the new obs_state value
:type obs_state: ObsState
"""
if self._obs_state != obs_state:
self._obs_state = obs_state
if self._obs_state_callback is not None:
self._obs_state_callback(self._obs_state)
@property
def obs_state(self):
return self._obs_state
__all__ = ["SKAObsDevice", "main"]
class SKAObsDevice(SKABaseDevice):
......@@ -111,6 +48,7 @@ class SKAObsDevice(SKABaseDevice):
device.set_change_event("obsState", True, True)
device.set_archive_event("obsState", True, True)
device._obs_state = ObsState.EMPTY
device._obs_mode = ObsMode.IDLE
device._config_progress = 0
device._config_delay_expected = 0
......@@ -166,26 +104,27 @@ class SKAObsDevice(SKABaseDevice):
:param obs_state: the new obs_state value
:type obs_state: ObsState
"""
self._obs_state = obs_state
self.push_change_event("obsState", obs_state)
self.push_archive_event("obsState", obs_state)
def _init_state_model(self):
"""
Sets up the state model for the device
"""
self.state_model = SKAObsDeviceStateModel(
dev_state_callback=self._update_state,
admin_mode_callback=self._update_admin_mode,
obs_state_callback=self._update_obs_state
)
def always_executed_hook(self):
# PROTECTED REGION ID(SKAObsDevice.always_executed_hook) ENABLED START #
"""
Method that is always executed before any device command gets executed.
:return: None
"""
pass
# PROTECTED REGION END # // SKAObsDevice.always_executed_hook
def delete_device(self):
# PROTECTED REGION ID(SKAObsDevice.delete_device) ENABLED START #
"""
Method to cleanup when device is stopped.
:return: None
"""
pass
# PROTECTED REGION END # // SKAObsDevice.delete_device
......@@ -196,7 +135,7 @@ class SKAObsDevice(SKABaseDevice):
def read_obsState(self):
# PROTECTED REGION ID(SKAObsDevice.obsState_read) ENABLED START #
"""Reads Observation State of the device"""
return self.state_model.obs_state
return self._obs_state
# PROTECTED REGION END # // SKAObsDevice.obsState_read
def read_obsMode(self):
......
src/ska/base/release.py
+ 1
− 1
View file @ d6363294
......@@ -7,7 +7,7 @@
"""Release information for lmc-base-classes Python Package"""
name = """lmcbaseclasses"""
version = "0.6.3"
version = "0.6.4"
version_info = version.split(".")
description = """A set of generic base devices for SKA Telescope."""
author = "SKA India and SARAO and CSIRO"
......
src/ska/base/state_machine.py 0 → 100644
+ 469
− 0
View file @ d6363294
"""
This module contains specifications of SKA state machines.
"""
from transitions import Machine, State
from tango import DevState
from ska.base.control_model import AdminMode, ObsState
class BaseDeviceStateMachine(Machine):
"""
State machine for an SKA base device. Supports ON and OFF states,
states, plus initialisation and fault states, and
also the basic admin modes.
"""
def __init__(self, op_state_callback=None, admin_mode_callback=None):
"""
Initialises the state model.
:param op_state_callback: A callback to be called when a
transition implies a change to op state
:type op_state_callback: callable
:param admin_mode_callback: A callback to be called when a
transition causes a change to device admin_mode
:type admin_mode_callback: callable
"""
self._admin_mode = None
self._admin_mode_callback = admin_mode_callback
self._op_state = None
self._op_state_callback = op_state_callback
states = [
State("UNINITIALISED"),
State("INIT_ENABLED", on_enter="_init_entered"),
State("INIT_DISABLED", on_enter="_init_entered"),
State("FAULT_ENABLED", on_enter="_fault_entered"),
State("FAULT_DISABLED", on_enter="_fault_entered"),
State("DISABLED", on_enter="_disabled_entered"),
State("OFF", on_enter="_off_entered"),
State("ON", on_enter="_on_entered"),
]
transitions = [
{
"source": "UNINITIALISED",
"trigger": "init_started",
"dest": "INIT_ENABLED",
"after": self._maintenance_callback,
},
{
"source": ["INIT_ENABLED", "OFF", "FAULT_ENABLED", "ON"],
"trigger": "fatal_error",
"dest": "FAULT_ENABLED",
},
{
"source": ["INIT_DISABLED", "DISABLED", "FAULT_DISABLED"],
"trigger": "fatal_error",
"dest": "FAULT_DISABLED",
},
{
"source": "INIT_ENABLED",
"trigger": "init_succeeded",
"dest": "OFF",
},
{
"source": "INIT_DISABLED",
"trigger": "init_succeeded",
"dest": "DISABLED",
},
{
"source": "INIT_ENABLED",
"trigger": "init_failed",
"dest": "FAULT_ENABLED",
},
{
"source": "INIT_DISABLED",
"trigger": "init_failed",
"dest": "FAULT_DISABLED",
},
{
"source": ["INIT_ENABLED", "INIT_DISABLED"],
"trigger": "to_notfitted",
"dest": "INIT_DISABLED",
"after": self._not_fitted_callback
},
{
"source": ["INIT_ENABLED", "INIT_DISABLED"],
"trigger": "to_offline",
"dest": "INIT_DISABLED",
"after": self._offline_callback
},
{
"source": ["INIT_ENABLED", "INIT_DISABLED"],
"trigger": "to_maintenance",
"dest": "INIT_ENABLED",
"after": self._maintenance_callback
},
{
"source": ["INIT_ENABLED", "INIT_DISABLED"],
"trigger": "to_online",
"dest": "INIT_ENABLED",
"after": self._online_callback
},
{
"source": "FAULT_DISABLED",
"trigger": "reset_succeeded",
"dest": "DISABLED",
},
{
"source": "FAULT_ENABLED",
"trigger": "reset_succeeded",
"dest": "OFF",
},
{
"source": ["FAULT_DISABLED", "FAULT_ENABLED"],
"trigger": "reset_failed",
"dest": None,
},
{
"source": ["FAULT_DISABLED", "FAULT_ENABLED"],
"trigger": "to_notfitted",
"dest": "FAULT_DISABLED",
"after": self._not_fitted_callback
},
{
"source": ["FAULT_DISABLED", "FAULT_ENABLED"],
"trigger": "to_offline",
"dest": "FAULT_DISABLED",
"after": self._offline_callback
},
{
"source": ["FAULT_DISABLED", "FAULT_ENABLED"],
"trigger": "to_maintenance",
"dest": "FAULT_ENABLED",
"after": self._maintenance_callback
},
{
"source": ["FAULT_DISABLED", "FAULT_ENABLED"],
"trigger": "to_online",
"dest": "FAULT_ENABLED",
"after": self._online_callback
},
{
"source": "DISABLED",
"trigger": "to_notfitted",
"dest": "DISABLED",
"after": self._not_fitted_callback
},
{
"source": "DISABLED",
"trigger": "to_offline",
"dest": "DISABLED",
"after": self._offline_callback
},
{
"source": "DISABLED",
"trigger": "to_maintenance",
"dest": "OFF",
"after": self._maintenance_callback
},
{
"source": "DISABLED",
"trigger": "to_online",
"dest": "OFF",
"after": self._online_callback
},
{
"source": "OFF",
"trigger": "to_notfitted",
"dest": "DISABLED",
"after": self._not_fitted_callback
},
{
"source": "OFF",
"trigger": "to_offline",
"dest": "DISABLED",
"after": self._offline_callback
},
{
"source": "OFF",
"trigger": "to_maintenance",
"dest": "OFF",
"after": self._maintenance_callback
},
{
"source": "OFF",
"trigger": "to_online",
"dest": "OFF",
"after": self._online_callback
},
{
"source": "OFF",
"trigger": "on_succeeded",
"dest": "ON",
},
{
"source": "OFF",
"trigger": "on_failed",
"dest": "FAULT_ENABLED",
},
{
"source": "ON",
"trigger": "off_succeeded",
"dest": "OFF",
},
{
"source": "ON",
"trigger": "off_failed",
"dest": "FAULT_ENABLED",
},
]
super().__init__(
states=states,
initial="UNINITIALISED",
transitions=transitions,
)
def _init_entered(self):
"""
called when the state machine enters the "" state.
"""
self._update_op_state(DevState.INIT)
def _fault_entered(self):
"""
called when the state machine enters the "" state.
"""
self._update_op_state(DevState.FAULT)
def _disabled_entered(self):
"""
called when the state machine enters the "" state.
"""
self._update_op_state(DevState.DISABLE)
def _off_entered(self):
"""
called when the state machine enters the "" state.
"""
self._update_op_state(DevState.OFF)
def _on_entered(self):
"""
called when the state machine enters the "" state.
"""
self._update_op_state(DevState.ON)
def _not_fitted_callback(self):
"""
callback called when the state machine is set to admin mode
NOT FITTED
"""
self._update_admin_mode(AdminMode.NOT_FITTED)
def _offline_callback(self):
"""
callback called when the state machine is set to admin mode
OFFLINE
"""
self._update_admin_mode(AdminMode.OFFLINE)
def _maintenance_callback(self):
"""
callback called when the state machine is set to admin mode
MAINTENANCE
"""
self._update_admin_mode(AdminMode.MAINTENANCE)
def _online_callback(self):
"""
callback called when the state machine is set to admin mode
online
"""
self._update_admin_mode(AdminMode.ONLINE)
def _update_admin_mode(self, admin_mode):
"""
Helper method: calls the admin_mode callback if one exists
:param admin_mode: the new admin_mode value
:type admin_mode: AdminMode
"""
if self._admin_mode != admin_mode:
self._admin_mode = admin_mode
if self._admin_mode_callback is not None:
self._admin_mode_callback(self._admin_mode)
def _update_op_state(self, op_state):
"""
Helper method: sets this state models op_state, and calls the
op_state callback if one exists
:param op_state: the new op state value
:type op_state: DevState
"""
if self._op_state != op_state:
self._op_state = op_state
if self._op_state_callback is not None:
self._op_state_callback(self._op_state)
class ObservationStateMachine(Machine):
"""
The observation state machine used by an observing subarray, per
ADR-8.
"""
def __init__(self, obs_state_callback=None):
"""
Initialises the model.
:param obs_state_callback: A callback to be called when a
transition causes a change to device obs_state
:type obs_state_callback: callable
"""
self._obs_state = ObsState.EMPTY
self._obs_state_callback = obs_state_callback
states = [obs_state.name for obs_state in ObsState]
transitions = [
{
"source": "*",
"trigger": "fatal_error",
"dest": ObsState.FAULT.name,
},
{
"source": [ObsState.EMPTY.name, ObsState.IDLE.name],
"trigger": "assign_started",
"dest": ObsState.RESOURCING.name,
},
{
"source": ObsState.IDLE.name,
"trigger": "release_started",
"dest": ObsState.RESOURCING.name,
},
{
"source": ObsState.RESOURCING.name,
"trigger": "resourcing_succeeded_some_resources",
"dest": ObsState.IDLE.name,
},
{
"source": ObsState.RESOURCING.name,
"trigger": "resourcing_succeeded_no_resources",
"dest": ObsState.EMPTY.name,
},
{
"source": ObsState.RESOURCING.name,
"trigger": "resourcing_failed",
"dest": ObsState.FAULT.name,
},
{
"source": [ObsState.IDLE.name, ObsState.READY.name],
"trigger": "configure_started",
"dest": ObsState.CONFIGURING.name,
},
{
"source": ObsState.CONFIGURING.name,
"trigger": "configure_succeeded",
"dest": ObsState.READY.name,
},
{
"source": ObsState.CONFIGURING.name,
"trigger": "configure_failed",
"dest": ObsState.FAULT.name,
},
{
"source": ObsState.READY.name,
"trigger": "end_succeeded",
"dest": ObsState.IDLE.name,
},
{
"source": ObsState.READY.name,
"trigger": "end_failed",
"dest": ObsState.FAULT.name,
},
{
"source": ObsState.READY.name,
"trigger": "scan_started",
"dest": ObsState.SCANNING.name,
},
{
"source": ObsState.SCANNING.name,
"trigger": "scan_succeeded",
"dest": ObsState.READY.name,
},
{
"source": ObsState.SCANNING.name,
"trigger": "scan_failed",
"dest": ObsState.FAULT.name,
},
{
"source": ObsState.SCANNING.name,
"trigger": "end_scan_succeeded",
"dest": ObsState.READY.name,
},
{
"source": ObsState.SCANNING.name,
"trigger": "end_scan_failed",
"dest": ObsState.FAULT.name,
},
{
"source": [
ObsState.IDLE.name, ObsState.CONFIGURING.name,
ObsState.READY.name, ObsState.SCANNING.name,
],
"trigger": "abort_started",
"dest": ObsState.ABORTING.name,
},
{
"source": ObsState.ABORTING.name,
"trigger": "abort_succeeded",
"dest": ObsState.ABORTED.name,
},
{
"source": ObsState.ABORTING.name,
"trigger": "abort_failed",
"dest": ObsState.FAULT.name,
},
{
"source": [ObsState.ABORTED.name, ObsState.FAULT.name],
"trigger": "obs_reset_started",
"dest": ObsState.RESETTING.name,
},
{
"source": ObsState.RESETTING.name,
"trigger": "obs_reset_succeeded",
"dest": ObsState.IDLE.name,
},
{
"source": ObsState.RESETTING.name,
"trigger": "obs_reset_failed",
"dest": ObsState.FAULT.name,
},
{
"source": [ObsState.ABORTED.name, ObsState.FAULT.name],
"trigger": "restart_started",
"dest": ObsState.RESTARTING.name,
},
{
"source": ObsState.RESTARTING.name,
"trigger": "restart_succeeded",
"dest": ObsState.EMPTY.name,
},
{
"source": ObsState.RESTARTING.name,
"trigger": "restart_failed",
"dest": ObsState.FAULT.name,
},
]
super().__init__(
states=states,
initial=ObsState.EMPTY.name,
transitions=transitions,
after_state_change=self._obs_state_changed
)
def _obs_state_changed(self):
"""
State machine callback that is called every time the obs_state
changes. Responsible for ensuring that callbacks are called.
"""
obs_state = ObsState[self.state]
if self._obs_state != obs_state:
self._obs_state = obs_state
if self._obs_state_callback is not None:
self._obs_state_callback(self._obs_state)
src/ska/base/subarray_device.py
+ 195
− 220
View file @ d6363294
This diff is collapsed.
src/ska/base/utils.py
+ 25
− 0
View file @ d6363294
"""General utilities that may be useful to SKA devices and clients."""
from builtins import str
import ast
import functools
import inspect
import json
import pydoc
import traceback
import sys
import warnings
from datetime import datetime
......@@ -462,3 +464,26 @@ def convert_dict_to_list(dictionary):
the_list.append("{}:{}".format(key, value))
return sorted(the_list)
def for_testing_only(func, _testing_check=lambda: 'pytest' in sys.modules):
"""
A decorator that marks a function as available for testing purposes only.
If the decorated function is called outside of testing, a warning is raised.
Testing this decorator leads to a Godelian paradox: how to test that a
warning is raised when we are not testing. Monkeypatching sys.modules would
break everything, so instead, the condition that we evaluate to decide
whether we are testing or not is exposed through a `_testing_check`
argument, allowing for it to be replaced in testing. (The `_testing_check`
argument is inaccessible via the @-syntax, which is a nice bonus.)
"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
"""
Function wrapper for `testing_only` decorator.
"""
if not _testing_check():
warnings.warn(f"{func.__name__} should only be used for testing purposes.")
return func(*args, **kwargs)
return wrapper
tests/conftest.py
+ 230
− 0
View file @ d6363294
......@@ -2,13 +2,238 @@
A module defining a list of fixtures that are shared across all ska.base tests.
"""
import importlib
import itertools
import json
import pytest
from queue import Empty, Queue
from transitions import MachineError
from tango import EventType
from tango.test_context import DeviceTestContext
def pytest_configure(config):
"""
pytest hook, used here to register custom "state_machine_tester" marks
"""
config.addinivalue_line(
"markers",
"state_machine_tester: indicate that this class is state machine "
"tester class, and tests should be parameterised by the states and "
"actions in the specification provided in its argument."
)
def pytest_generate_tests(metafunc):
"""
pytest hook that generates tests; this hook ensures that any test
class that is marked with the `state_machine_tester` custom marker
will have its tests parameterised by the states and actions in the
specification provided by that mark
"""
# called once per each test function
mark = metafunc.definition.get_closest_marker("state_machine_tester")
if mark:
spec = mark.args[0]
states = set()
triggers = set()
expected = {}
for (from_state, trigger, to_state) in spec:
states.add(from_state)
states.add(to_state)
triggers.add(trigger)
expected[(from_state, trigger)] = to_state
states = sorted(states)
triggers = sorted(triggers)
metafunc.parametrize(
"state_under_test, action_under_test, expected_state",
[
(
state,
trigger,
expected[(state, trigger)] if (state, trigger) in expected else None
) for (state, trigger) in itertools.product(states, triggers)
]
)
class StateMachineTester:
"""
Abstract base class for a class for testing state machines
"""
def test_state_machine(
self, machine, state_under_test, action_under_test, expected_state,
):
"""
Implements the unit test for a state machine: for a given
initial state and an action, does execution of that action, from
that state, yield the expected results? If the action was
allowed from that state, does the machine transition to the
correct state? If the action was not allowed from that state,
does the machine reject the action (e.g. raise an exception or
return an error code) and remain in the current state?
:param machine: the state machine under test
:type machine: state machine object instance
:param state_under_test: the state from which the
`action_under_test` is being tested
:type state_under_test: string
:param action_under_test: the action being tested from the
`state_under_test`
:type action_under_test: string
:param expected_state: the state to which the machine is
expected to transition, as a result of performing the
`action_under_test` in the `state_under_test`. If None, then
the action should be disallowed and result in no change of
state.
:type expected_state: string
"""
# Put the device into the state under test
self.to_state(machine, state_under_test)
# Check that we are in the state under test
self.assert_state(machine, state_under_test)
# Test that the action under test does what we expect it to
if expected_state is None:
# Action should fail and the state should not change
self.check_action_disallowed(machine, action_under_test)
self.assert_state(machine, state_under_test)
else:
# Action should succeed
self.perform_action(machine, action_under_test)
self.assert_state(machine, expected_state)
def assert_state(self, machine, state):
"""
Abstract method for asserting the current state of the state
machine under test
:param machine: the state machine under test
:type machine: state machine object instance
:param state: the state that we are asserting to be the current
state of the state machine under test
:type state: string
"""
raise NotImplementedError()
def perform_action(self, machine, action):
"""
Abstract method for performing an action on the state machine
:param machine: the state machine under test
:type machine: state machine object instance
:param action: action to be performed on the state machine
:type action: str
"""
raise NotImplementedError()
def check_action_disallowed(self, machine, action):
"""
Abstract method for asserting that an action fails if performed
on the state machine under test in its current state.
:param machine: the state machine under test
:type machine: state machine object instance
:param action: action to be performed on the state machine
:type action: str
"""
raise NotImplementedError()
def to_state(self, machine, target_state):
"""
Abstract method for getting the state machine into a target
state.
:param machine: the state machine under test
:type machine: state machine object instance
:param target_state: the state that we want to get the state
machine under test into
:type target_state: str
"""
raise NotImplementedError()
class TransitionsStateMachineTester(StateMachineTester):
"""
Concrete implementation of a StateMachineTester for a pytransitions
state machine (with autotransitions turned on). The states and
actions in the state machine specification must correspond exactly
with the machine's states and triggers.
"""
def assert_state(self, machine, state):
"""
Assert the current state of the state machine under test.
:param machine: the state machine under test
:type machine: state machine object instance
:param state: the state that we are asserting to be the current
state of the state machine under test
:type state: str
"""
assert machine.state == state
def perform_action(self, machine, action):
"""
Perform a given action on the state machine under test.
:param machine: the state machine under test
:type machine: state machine object instance
:param action: action to be performed on the state machine
:type action: str
"""
machine.trigger(action)
def check_action_disallowed(self, machine, action):
"""
Assert that performing a given action on the state maching under
test fails in its current state.
:param machine: the state machine under test
:type machine: state machine object instance
:param action: action to be performed on the state machine
:type action: str
"""
with pytest.raises(MachineError):
self.perform_action(machine, action)
def to_state(self, machine, target_state):
"""
Transition the state machine to a target state. This
implementation uses autotransitions. If the pytransitions state
machine under test has autotransitions turned off, then this
method will need to be overridden by some other method of
putting the machine into the state under test.
:param machine: the state machine under test
:type machine: state machine object instance
:param target_state: the state that we want to get the state
machine under test into
:type target_state: str
"""
machine.trigger(f"to_{target_state}")
def load_data(name):
"""
Loads a dataset by name. This implementation uses the name to find a
JSON file containing the data to be loaded.
:param name: name of the dataset to be loaded; this implementation
uses the name to find a JSON file containing the data to be
loaded.
:type name: string
"""
with open(f"tests/data/{name}.json", "r") as json_file:
return json.load(json_file)
@pytest.fixture(scope="class")
def tango_context(request):
"""Creates and returns a TANGO DeviceTestContext object.
......@@ -95,6 +320,11 @@ def tango_change_event_helper(tango_context):
"""
class _Callback:
"""
Private callback handler class, an instance of which is returned
by the tango_change_event_helper each time it is used to
subscribe to a change event.
"""
@staticmethod
def subscribe(attribute_name):
"""
......
tests/data/base_device_state_machine.json 0 → 100644
+ 222
− 0
View file @ d6363294
[
[
"UNINITIALISED",
"init_started",
"INIT_ENABLED"
],
[
"INIT_ENABLED",
"to_notfitted",
"INIT_DISABLED"
],
[
"INIT_ENABLED",
"to_offline",
"INIT_DISABLED"
],
[
"INIT_ENABLED",
"to_online",
"INIT_ENABLED"
],
[
"INIT_ENABLED",
"to_maintenance",
"INIT_ENABLED"
],
[
"INIT_ENABLED",
"init_succeeded",
"OFF"
],
[
"INIT_ENABLED",
"init_failed",
"FAULT_ENABLED"
],
[
"INIT_ENABLED",
"fatal_error",
"FAULT_ENABLED"
],
[
"INIT_DISABLED",
"to_notfitted",
"INIT_DISABLED"
],
[
"INIT_DISABLED",
"to_offline",
"INIT_DISABLED"
],
[
"INIT_DISABLED",
"to_online",
"INIT_ENABLED"
],
[
"INIT_DISABLED",
"to_maintenance",
"INIT_ENABLED"
],
[
"INIT_DISABLED",
"init_succeeded",
"DISABLED"
],
[
"INIT_DISABLED",
"init_failed",
"FAULT_DISABLED"
],
[
"INIT_DISABLED",
"fatal_error",
"FAULT_DISABLED"
],
[
"FAULT_DISABLED",
"to_notfitted",
"FAULT_DISABLED"
],
[
"FAULT_DISABLED",
"to_offline",
"FAULT_DISABLED"
],
[
"FAULT_DISABLED",
"to_online",
"FAULT_ENABLED"
],
[
"FAULT_DISABLED",
"to_maintenance",
"FAULT_ENABLED"
],
[
"FAULT_DISABLED",
"reset_succeeded",
"DISABLED"
],
[
"FAULT_DISABLED",
"reset_failed",
"FAULT_DISABLED"
],
[
"FAULT_DISABLED",
"fatal_error",
"FAULT_DISABLED"
],
[
"FAULT_ENABLED",
"to_notfitted",
"FAULT_DISABLED"
],
[
"FAULT_ENABLED",
"to_offline",
"FAULT_DISABLED"
],
[
"FAULT_ENABLED",
"to_online",
"FAULT_ENABLED"
],
[
"FAULT_ENABLED",
"to_maintenance",
"FAULT_ENABLED"
],
[
"FAULT_ENABLED",
"reset_succeeded",
"OFF"
],
[
"FAULT_ENABLED",
"reset_failed",
"FAULT_ENABLED"
],
[
"FAULT_ENABLED",
"fatal_error",
"FAULT_ENABLED"
],
[
"DISABLED",
"to_notfitted",
"DISABLED"
],
[
"DISABLED",
"to_offline",
"DISABLED"
],
[
"DISABLED",
"to_online",
"OFF"
],
[
"DISABLED",
"to_maintenance",
"OFF"
],
[
"DISABLED",
"fatal_error",
"FAULT_DISABLED"
],
[
"OFF",
"to_notfitted",
"DISABLED"
],
[
"OFF",
"to_offline",
"DISABLED"
],
[
"OFF",
"to_online",
"OFF"
],
[
"OFF",
"to_maintenance",
"OFF"
],
[
"OFF",
"on_succeeded",
"ON"
],
[
"OFF",
"on_failed",
"FAULT_ENABLED"
],
[
"OFF",
"fatal_error",
"FAULT_ENABLED"
],
[
"ON",
"off_succeeded",
"OFF"
],
[
"ON",
"off_failed",
"FAULT_ENABLED"
],
[
"ON",
"fatal_error",
"FAULT_ENABLED"
]
]
\ No newline at end of file
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment