# -*- coding: utf-8 -*-
#
# This file is part of the SKA Control System project.
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE.txt for more info.
"""This module defines an enumerated type and state model for admin mode."""
from __future__ import annotations
import enum
import logging
from typing import Any, Callable
from transitions.extensions import LockedMachine as Machine
from ska_control_model.faults import StateModelError
from ska_control_model.utils import for_testing_only
[docs]
class AdminMode(enum.IntEnum):
"""
Python enumerated type for device admin mode.
An admin mode represents user intent as to how the component under
control will be used.
"""
ONLINE = 0
"""
The component under control can be used for normal operations, such
as observing. While in this mode, the control system actively
monitors and controls the component under control.
Control system elements that implement admin mode as a read-only
attribute shall always report the admin mode to be ``ONLINE``.
"""
OFFLINE = 1
"""
The component under control shall not be monitored or controlled by
the control system.
Either the component under control shall not be used at all, or it
is under external control (such as the local control of a field
technician).
While in this mode, the control system reports its state as
``DISABLE``. Since monitoring of the component is not occurring,
the control system does not issue alarms, alerts and other events.
"""
ENGINEERING = 2
"""
The component under control can be used for engineering purposes,
such as testing, debugging or commissioning, as part of an
"engineering subarray". It may not be used for normal operations.
While in this mode, the control system actively monitors and
controls its component, but may only support a subset of normal
functionality. Alarms and alerts will usually be suppressed.
``ENGINEERING`` mode has different meaning for different components,
depending on the context and functionality. Some entities may
implement different behaviour when in ``ENGINEERING`` mode. For each
Tango device, the difference in behaviour and functionality in
``ENGINEERING`` mode shall be documented.
"""
NOT_FITTED = 3
"""
The component cannot be used for any purposes because it is not
fitted; for example, faulty equipment has been removed and not
yet replaced, leaving nothing `in situ` to monitor.
While in this mode, the control system reports state ``DISABLED``.
All monitoring and control functionality is disabled because there
is no component to monitor.
"""
RESERVED = 4
"""
The component is fitted, but only for redundancy purposes. It is
additional equipment that does not take part in operations at this
time, but is ready to take over when the operational
equipment fails.
While in this mode, the control system reports state ``DISABLED``.
All monitoring and control functionality is disabled.
"""
class _AdminModeMachine(Machine):
"""
The state machine governing admin modes.
For documentation of states and transitions, see the documentation
of the public :py:class:`.AdminModeModel` class.
"""
def __init__(
self,
callback: Callable[[str], None] | None = None,
**extra_kwargs: Any,
) -> None:
"""
Initialise the admin mode state machine model.
:param callback: A callback to be called whenever there is a
transition to a new admin mode value
:param extra_kwargs: Additional keywords arguments to pass to
super class initialiser (useful for graphing)
"""
self._callback = callback
states = ["RESERVED", "NOT_FITTED", "OFFLINE", "ENGINEERING", "ONLINE"]
transitions = [
{
"source": ["NOT_FITTED", "RESERVED", "OFFLINE"],
"trigger": "to_reserved",
"dest": "RESERVED",
},
{
"source": ["RESERVED", "NOT_FITTED", "OFFLINE"],
"trigger": "to_notfitted",
"dest": "NOT_FITTED",
},
{
"source": [
"RESERVED",
"NOT_FITTED",
"OFFLINE",
"ENGINEERING",
"ONLINE",
],
"trigger": "to_offline",
"dest": "OFFLINE",
},
{
"source": ["OFFLINE", "ENGINEERING", "ONLINE"],
"trigger": "to_engineering",
"dest": "ENGINEERING",
},
{
"source": ["OFFLINE", "ENGINEERING", "ONLINE"],
"trigger": "to_online",
"dest": "ONLINE",
},
]
super().__init__(
states=states,
initial="OFFLINE",
transitions=transitions,
after_state_change=self._state_changed,
**extra_kwargs,
)
self._state_changed()
def _state_changed(self) -> None:
"""
Handle change in machine state.
Responsible for ensuring that callbacks are called.
"""
if self._callback is not None:
self._callback(self.state)
[docs]
class AdminModeModel:
"""
This class implements the state model for admin mode.
The model supports the five admin modes defined by the values of the
:py:class:`AdminMode` enum. It allows for:
* any transition between the modes NOT_FITTED, RESERVED and OFFLINE
(e.g. an unfitted device being fitted as a redundant or
non-redundant device, a redundant device taking over when another
device fails, etc.)
* any transition between the modes OFFLINE, ENGINEERING and ONLINE
(e.g. an online device being taken offline or put into engineering
mode to diagnose a fault, a faulty device moving between
engineering and offline mode as it undergoes sporadic periods of
diagnosis.)
The actions supported are:
* **to_not_fitted**
* **to_reserved**
* **to_offline**
* **to_engineering**
* **to_online**
A diagram of the admin mode model, as designed, is shown below
.. uml:: admin_mode_model.uml
:caption: Diagram of the admin mode model
"""
def __init__(
self,
logger: logging.Logger,
callback: Callable[[AdminMode], None] | None = None,
state_machine_factory: Callable[..., Machine] = _AdminModeMachine,
) -> None:
"""
Initialise the state model.
:param logger: the logger to be used by this state model.
:param callback: A callback to be called when the state machine
for admin_mode reports a change of state
:param state_machine_factory: a callable that returns a
state machine for this model to use
"""
self.logger = logger
self._admin_mode = AdminMode.OFFLINE
self._callback = callback
self._admin_mode_machine = state_machine_factory(
callback=self._admin_mode_changed
)
@property
def admin_mode(self) -> AdminMode:
"""
Return the admin_mode.
:returns: admin_mode of this state model
"""
return self._admin_mode
def _admin_mode_changed(self, machine_state: str) -> None:
"""
Handle a change in admin mode.
This is a helper method that updates admin mode, ensuring that
the callback is called if one exists.
:param machine_state: the new state of the admin mode machine
"""
admin_mode = AdminMode[machine_state]
if self._admin_mode != admin_mode:
self._admin_mode = admin_mode
if self._callback is not None:
self._callback(admin_mode)
def is_action_allowed(self, action: str, raise_if_disallowed: bool = False) -> bool:
"""
Return whether a given action is allowed in the current state.
:param action: an action, as given in the transitions table
:param raise_if_disallowed: whether to raise an exception if the
action is disallowed, or merely return False (optional,
defaults to False)
:raises StateModelError: if the action is unknown to the state
machine
:return: whether the action is allowed in the current state
"""
if action in self._admin_mode_machine.get_triggers(
self._admin_mode_machine.state
):
return True
if raise_if_disallowed:
raise StateModelError(
f"Action {action} is not allowed in admin mode " f"{self.admin_mode}."
)
return False
def perform_action(self, action: str) -> None:
"""
Perform an action on the state model.
:param action: an action, as given in the transitions table
"""
_ = self.is_action_allowed(action, raise_if_disallowed=True)
self._admin_mode_machine.trigger(action)
@for_testing_only
def _straight_to_state(self, admin_mode_name: str) -> None:
"""
Take this AdminMode state model straight to the specified AdminMode.
This method exists to simplify testing; for example, if testing
that a command may be run in a given AdminMode, you can push
this state model straight to that AdminMode, 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.
:param admin_mode_name: name of the target admin mode
"""
getattr(self._admin_mode_machine, f"to_{admin_mode_name}")()