diff --git a/mxcubecore/HardwareObjects/TangoShutter.py b/mxcubecore/HardwareObjects/TangoShutter.py
index 16c7141ee0..ef13b1dbc8 100644
--- a/mxcubecore/HardwareObjects/TangoShutter.py
+++ b/mxcubecore/HardwareObjects/TangoShutter.py
@@ -31,24 +31,35 @@
Open
Close
State
- {"open": "OPEN", "cloded": "CLOSED", "DISABLE" : "DISABLE"}
+ {"OPEN": "MYOPEN", "NEWSTATE": ["MYSTATE", "BUSY"]}
-In this example the tag contains a json dictionary that maps spectific tango shutter states to the
-convantional states defined in the TangoShutter Class. This tag is not necessay in cases where the tango shutter states
-are all covered by the TangoShuter class conventional states.
+In the example the property contains a dictionary that redefines or
+adds specific tango shutter states.
+When redefining a known state, only the VALUES Enum will be updated.
+When defining a new state (new key), the dictionary value should be a
+list. The new state is added to both the VALUES and the SPECIFIC_STATES Enum.
+Attention:
+ - do not use tuples or the python json parser will fail!
+ - make sure only double quotes are used inside the values dictionary. No single quotes (') are allowed !
+ - the second element of the list should be a standard HardwareObjectState name
+ (UNKNOWN, WARNING, BUSY, READY, FAULT, OFF - see in BaseHardwareObjects.py)!
+The property is optional.
"""
import json
+import logging
from enum import (
Enum,
unique,
)
+from tango import DevState
+
from mxcubecore.BaseHardwareObjects import HardwareObjectState
from mxcubecore.HardwareObjects.abstract.AbstractShutter import AbstractShutter
-__copyright__ = """ Copyright © 2023 by the MXCuBE collaboration """
+__copyright__ = """ Copyright © by the MXCuBE collaboration """
__license__ = "LGPLv3+"
@@ -63,6 +74,7 @@ class TangoShutterStates(Enum):
AUTOMATIC = HardwareObjectState.READY, "RUNNING"
UNKNOWN = HardwareObjectState.UNKNOWN, "RUNNING"
FAULT = HardwareObjectState.WARNING, "FAULT"
+ STANDBY = HardwareObjectState.WARNING, "STANDBY"
class TangoShutter(AbstractShutter):
@@ -70,13 +82,13 @@ class TangoShutter(AbstractShutter):
SPECIFIC_STATES = TangoShutterStates
- def __init__(self, name):
+ def __init__(self, name: str):
super().__init__(name)
self.open_cmd = None
self.close_cmd = None
self.state_channel = None
- def init(self):
+ def init(self) -> None:
"""Initilise the predefined values"""
super().init()
self.open_cmd = self.get_command_object("Open")
@@ -86,26 +98,18 @@ def init(self):
self.state_channel.connect_signal("update", self._update_value)
self.update_state()
- try:
- self.config_values = json.loads(self.get_property("values"))
- except:
- self.config_values = None
-
- def _update_value(self, value):
+ def _update_value(self, value: DevState) -> None:
"""Update the value.
Args:
- value(str): The value reported by the state channel.
+ value: The value reported by the state channel.
"""
- if self.config_values:
- value = self.config_values[str(value)]
- else:
- value = str(value)
-
- super().update_value(self.value_to_enum(value))
+ super().update_value(self.value_to_enum(str(value)))
- def _initialise_values(self):
- """Add the tango states to VALUES"""
+ def _initialise_values(self) -> None:
+ """Add specific tango states to VALUES and, if configured
+ in the xml file, to SPECIFIC_STATES"""
values_dict = {item.name: item.value for item in self.VALUES}
+ states_dict = {item.name: item.value for item in self.SPECIFIC_STATES}
values_dict.update(
{
"MOVING": "MOVING",
@@ -114,36 +118,41 @@ def _initialise_values(self):
"FAULT": "FAULT",
}
)
+ try:
+ config_values = json.loads(self.get_property("values"))
+ for key, val in config_values.items():
+ if isinstance(val, (tuple, list)):
+ values_dict.update({key: val[1]})
+ states_dict.update({key: (HardwareObjectState[val[1]], val[0])})
+ else:
+ values_dict.update({key: val})
+ except (ValueError, TypeError):
+ logging.exception("Exception in _initialise_values()")
+
self.VALUES = Enum("ValueEnum", values_dict)
+ self.SPECIFIC_STATES = Enum("TangoShutterStates", states_dict)
- def get_state(self):
+ def get_state(self) -> HardwareObjectState:
"""Get the device state.
Returns:
(enum 'HardwareObjectState'): Device state.
"""
try:
- if self.config_values:
- _state = self.config_values[str(self.state_channel.get_value())]
- else:
- _state = str(self.state_channel.get_value())
-
+ _state = self.get_value().name
+ return self.SPECIFIC_STATES[_state].value[0]
except (AttributeError, KeyError):
+ logging.exception("Exception in get_state()")
return self.STATES.UNKNOWN
- return self.SPECIFIC_STATES[_state].value[0]
-
- def get_value(self):
+ def get_value(self) -> TangoShutterStates:
"""Get the device value
Returns:
(Enum): Enum member, corresponding to the 'VALUE' or UNKNOWN.
"""
- if self.config_values:
- _val = self.config_values[str(self.state_channel.get_value())]
- else:
- _val = str(self.state_channel.get_value())
+ _val = str(self.state_channel.get_value())
return self.value_to_enum(_val)
- def _set_value(self, value):
+ def _set_value(self, value: TangoShutterStates) -> None:
if value.name == "OPEN":
self.open_cmd()
elif value.name == "CLOSED":
diff --git a/ruff.toml b/ruff.toml
index 3c56f9706b..ea85fbc2f5 100644
--- a/ruff.toml
+++ b/ruff.toml
@@ -7118,3 +7118,6 @@ convention = "google"
"PT022",
"S108",
]
+"test/pytest/test_hwo_tango_shutter.py" = [
+ "N802"
+]
diff --git a/test/pytest/test_hwo_tango_shutter.py b/test/pytest/test_hwo_tango_shutter.py
new file mode 100644
index 0000000000..e731b3b788
--- /dev/null
+++ b/test/pytest/test_hwo_tango_shutter.py
@@ -0,0 +1,123 @@
+"""Test the HardwareObjects.TangoShutter shutter hardware object."""
+
+from typing import Callable
+
+import gevent
+import pytest
+from gevent import Timeout
+from tango import (
+ DeviceProxy,
+ DevState,
+)
+from tango.server import (
+ Device,
+ command,
+)
+from tango.test_context import DeviceTestContext
+
+from mxcubecore.HardwareObjects import TangoShutter
+
+TANGO_SHUTTER_STATES_MAPPING = """
+{"OPEN": "OPEN", "CLOSED": "CLOSE", "MOVING" : "MOVING"}
+"""
+
+
+class Shutter(Device):
+ """Very simple Tango shutter device, that only goes
+ between 'open' and 'close' states."""
+
+ def __init__(self, *args, **kwargs):
+ self._is_open = False
+ super().__init__(*args, **kwargs)
+
+ @command()
+ def Open(self):
+ self._is_open = True
+
+ @command()
+ def Close(self):
+ self._is_open = False
+
+ def dev_state(self):
+ return DevState.OPEN if self._is_open else DevState.CLOSE
+
+
+@pytest.fixture
+def shutter():
+ tangods_test_context = DeviceTestContext(Shutter, process=True)
+ tangods_test_context.start()
+
+ #
+ # set up the TangoShutter hardware object
+ #
+ hwo_shutter = TangoShutter.TangoShutter("/random_name")
+ hwo_shutter.tangoname = tangods_test_context.get_device_access()
+ hwo_shutter.set_property("values", TANGO_SHUTTER_STATES_MAPPING)
+ hwo_shutter.add_channel(
+ {
+ "name": "State",
+ "type": "tango",
+ },
+ "State",
+ add_now=True,
+ )
+ hwo_shutter.add_command(
+ {
+ "name": "Open",
+ "type": "tango",
+ },
+ "Open",
+ add_now=True,
+ )
+ hwo_shutter.add_command(
+ {
+ "name": "Close",
+ "type": "tango",
+ },
+ "Close",
+ add_now=True,
+ )
+
+ hwo_shutter.init()
+
+ yield hwo_shutter
+
+ tangods_test_context.stop()
+ tangods_test_context.join()
+
+
+def _wait_until(condition: Callable, condition_desc: str):
+ with Timeout(1.2, Exception(f"timed out while waiting for {condition_desc}")):
+ while not condition():
+ gevent.sleep(0.01)
+
+
+def test_open(shutter):
+ """
+ test opening the shutter
+ """
+ dev = DeviceProxy(shutter.tangoname)
+
+ assert dev.State() == DevState.CLOSE
+ assert not shutter.is_open
+
+ shutter.open()
+
+ _wait_until(lambda: shutter.is_open, "shutter to open")
+ assert dev.State() == DevState.OPEN
+
+
+def test_close(shutter):
+ """
+ test closing the shutter
+ """
+ dev = DeviceProxy(shutter.tangoname)
+ dev.Open()
+
+ assert dev.State() == DevState.OPEN
+ assert shutter.is_open
+
+ shutter.close()
+
+ _wait_until(lambda: not shutter.is_open, "shutter to close")
+ assert dev.State() == DevState.CLOSE