Skip to content

tango

TangoHOConfig marshmallow-model

The Tango Hardware Object Config Schema

Source code in daiquiri/core/hardware/tango/__init__.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class TangoHOConfig(HOConfigSchema):
    """The Tango Hardware Object Config Schema"""

    type = fields.Str(required=True, metadata={"description": "The object type"})
    tango_url = fields.Str(metadata={"description": "URL to tango device/attribute"})
    attrname = fields.Str(
        required=False, metadata={"description": "Used by TangoAttr object"}
    )

    @validates_schema
    def schema_validate(self, data, **kwargs):
        if not (data.get("tango_url") or data.get("url")):
            raise ValidationError(
                "Object must have either a `tango_url` or `url` defined"
            )

    @post_load
    def populate(self, data, **kwargs):
        """Maintain backwards compatibility with `tango_url`

        TODO: Remove me
        """
        url = data.get("url")
        if url is not None:
            # Normalize URL to a valid tango URL
            if url.startswith("tango:///"):
                # short schema without host and port
                url = url[len("tango:///") :]
            data["tango_url"] = url

        return data

Fields:

Name Type Properties Description
attrname String Used by TangoAttr object
tango_url String URL to tango device/attribute
type String Required The object type

TangoHandler

Source code in daiquiri/core/hardware/tango/__init__.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
class TangoHandler(ProtocolHandler):
    library = "tango"

    def get(self, **kwargs):
        # Discover tangoattr using from url
        if "url" in kwargs:
            url = kwargs["url"]

            if "type" not in kwargs or kwargs["type"] == "tangoattr":
                if url.count("/") == 6:
                    url, attrname = url.rsplit("/", 1)
                    kwargs["url"] = url
                    kwargs["attrname"] = attrname
                    if "type" not in kwargs:
                        kwargs["type"] = "tangoattr"

        try:
            kwargs = TangoHOConfig().load(kwargs)
        except ValidationError as err:
            raise InvalidYAML(
                {
                    "message": "Tango hardware object definition is invalid",
                    "file": "hardware.yml",
                    "obj": kwargs,
                    "errors": err.messages,
                }
            ) from err

        try:
            return loader(
                "daiquiri.core.hardware.tango",
                "",
                kwargs.get("type"),
                app=self._app,
                **kwargs
            )
        except ModuleNotFoundError:
            logger.error(
                "Could not find class for tango object {type}".format(
                    type=kwargs.get("type")
                )
            )
            raise

get(self, **kwargs)

Get the specific object from the protocol handler.

This function checks that kwargs conforms to Schema defined for the protocol handler (see core/hardware/bliss/init.py for a concrete example)

Returns:

Type Description

The hardware object instance for the specific protocol
Source code in daiquiri/core/hardware/tango/__init__.py 
def get(self, **kwargs):
    # Discover tangoattr using from url
    if "url" in kwargs:
        url = kwargs["url"]

        if "type" not in kwargs or kwargs["type"] == "tangoattr":
            if url.count("/") == 6:
                url, attrname = url.rsplit("/", 1)
                kwargs["url"] = url
                kwargs["attrname"] = attrname
                if "type" not in kwargs:
                    kwargs["type"] = "tangoattr"

    try:
        kwargs = TangoHOConfig().load(kwargs)
    except ValidationError as err:
        raise InvalidYAML(
            {
                "message": "Tango hardware object definition is invalid",
                "file": "hardware.yml",
                "obj": kwargs,
                "errors": err.messages,
            }
        ) from err

    try:
        return loader(
            "daiquiri.core.hardware.tango",
            "",
            kwargs.get("type"),
            app=self._app,
            **kwargs
        )
    except ModuleNotFoundError:
        logger.error(
            "Could not find class for tango object {type}".format(
                type=kwargs.get("type")
            )
        )
        raise

frontend

FrontendStateProperty

Source code in daiquiri/core/hardware/tango/frontend.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class FrontendStateProperty(HardwareProperty):
    def translate_from(self, value):
        val_map = {
            DevState.OPEN: "OPEN",
            DevState.RUNNING: "RUNNING",
            DevState.CLOSE: "CLOSED",
            DevState.STANDBY: "STANDBY",
            DevState.FAULT: "FAULT",
            DevState.UNKNOWN: "UNKNOWN",
        }
        for k, v in val_map.items():
            if k == value:
                return v

        return "UNKNOWN"

translate_from(self, value)

Translate the value from the controls layer to the abstraction layer i.e for getters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/frontend.py 
def translate_from(self, value):
    val_map = {
        DevState.OPEN: "OPEN",
        DevState.RUNNING: "RUNNING",
        DevState.CLOSE: "CLOSED",
        DevState.STANDBY: "STANDBY",
        DevState.FAULT: "FAULT",
        DevState.UNKNOWN: "UNKNOWN",
    }
    for k, v in val_map.items():
        if k == value:
            return v

    return "UNKNOWN"

ItlkStateProperty

Source code in daiquiri/core/hardware/tango/frontend.py 
14
15
16
17
18
19
20
21
22
class ItlkStateProperty(HardwareProperty):
    def translate_from(self, value):
        val_map = {DevState.ON: "ON", DevState.FAULT: "FAULT"}

        for k, v in val_map.items():
            if k == value:
                return v

        return "UNKNOWN"

translate_from(self, value)

Translate the value from the controls layer to the abstraction layer i.e for getters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/frontend.py 
def translate_from(self, value):
    val_map = {DevState.ON: "ON", DevState.FAULT: "FAULT"}

    for k, v in val_map.items():
        if k == value:
            return v

    return "UNKNOWN"

gauge

GaugeStateProperty

Source code in daiquiri/core/hardware/tango/gauge.py 
14
15
16
17
18
19
20
21
class GaugeStateProperty(HardwareProperty):
    def translate_from(self, value):
        val_map = {DevState.ON: "ON", DevState.OFF: "OFF", DevState.UNKNOWN: "UNKNOWN"}
        for k, v in val_map.items():
            if k == value:
                return v

        return "UNKNOWN"

translate_from(self, value)

Translate the value from the controls layer to the abstraction layer i.e for getters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/gauge.py 
def translate_from(self, value):
    val_map = {DevState.ON: "ON", DevState.OFF: "OFF", DevState.UNKNOWN: "UNKNOWN"}
    for k, v in val_map.items():
        if k == value:
            return v

    return "UNKNOWN"

lima

ImageFormatNotSupported

Raised when the RAW data from a Lima device can't be decoded as an RGB numpy array.

Source code in daiquiri/core/hardware/tango/lima.py 
29
30
class ImageFormatNotSupported(Exception):
    """Raised when the RAW data from a Lima device can't be decoded as an RGB numpy array."""

Lima

Source code in daiquiri/core/hardware/tango/lima.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class Lima(TangoObject, AbstractCamera):
    property_map = {
        "state": LimaStateProperty("acq_status"),
        "exposure": HardwareProperty("video_exposure"),
        "gain": HardwareProperty("video_gain"),
        "mode": HardwareProperty("video_mode"),
        "live": HardwareProperty("video_live"),
        "height": HardwareProperty("image_height"),
        "width": HardwareProperty("image_width"),
    }

    def frame(self) -> numpy.ndarray:
        _, raw_data = self._object.video_last_image
        if len(raw_data) > HEADER_SIZE:
            try:
                rotation = int(self._object.image_rotation)
            except ValueError:
                rotation = 0
            return parse_video_frame(
                raw_data, int(self._object.image_type.replace("Bpp", "")), rotation
            )

frame(self)

Return a frame from the camera

Source code in daiquiri/core/hardware/tango/lima.py 
def frame(self) -> numpy.ndarray:
    _, raw_data = self._object.video_last_image
    if len(raw_data) > HEADER_SIZE:
        try:
            rotation = int(self._object.image_rotation)
        except ValueError:
            rotation = 0
        return parse_video_frame(
            raw_data, int(self._object.image_type.replace("Bpp", "")), rotation
        )

LimaStateProperty

Source code in daiquiri/core/hardware/tango/lima.py 
166
167
168
169
170
171
172
173
class LimaStateProperty(HardwareProperty):
    def translate_from(self, value):
        vals = {"Ready": "READY", "Running": "ACQUIRING"}

        if value in vals:
            return vals[value]

        return "UNKNOWN"

translate_from(self, value)

Translate the value from the controls layer to the abstraction layer i.e for getters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/lima.py 
def translate_from(self, value):
    vals = {"Ready": "READY", "Running": "ACQUIRING"}

    if value in vals:
        return vals[value]

    return "UNKNOWN"

VIDEO_MODES

An enumeration.

Source code in daiquiri/core/hardware/tango/lima.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class VIDEO_MODES(enum.Enum):
    # From https://github.com/esrf-bliss/Lima/blob/master/common/include/lima/Constants.h#L118
    Y8 = 0
    Y16 = 1
    Y32 = 2
    Y64 = 3
    RGB555 = 4
    RGB565 = 5
    RGB24 = 6
    RGB32 = 7
    BGR24 = 8
    BGR32 = 9
    BAYER_RG8 = 10
    BAYER_RG16 = 11
    BAYER_BG8 = 12
    BAYER_BG16 = 13
    I420 = 14
    YUV411 = 15
    YUV422 = 16
    YUV444 = 17
    YUV411PACKED = 18
    YUV422PACKED = 19
    YUV444PACKED = 20

VideoCodec dataclass

VideoCodec(dtype: numpy.dtype, shape: Callable, opencv_code: int)

Source code in daiquiri/core/hardware/tango/lima.py 
58
59
60
61
62
@dataclass
class VideoCodec:
    dtype: numpy.dtype
    shape: typing.Callable
    opencv_code: int

parse_video_frame(raw_data, bpp=None, rotation=0)

Parse a lima video frame and convert to rgb

Lima frame decoding in bliss https://gitlab.esrf.fr/bliss/bliss/-/blob/master/bliss/data/lima_image.py

Available lima pixel formats: https://gitlab.esrf.fr/limagroup/lima/-/blob/master/common/include/lima/VideoUtils.h

OpenCV formats: https://docs.opencv.org/master/d1/d4f/imgproc_2include_2opencv2_2imgproc_8hpp.html

Source code in daiquiri/core/hardware/tango/lima.py 
def parse_video_frame(
    raw_data: bytes, bpp: int = None, rotation: int = 0
) -> numpy.ndarray:
    """Parse a lima video frame and convert to rgb

    Lima frame decoding in bliss
    https://gitlab.esrf.fr/bliss/bliss/-/blob/master/bliss/data/lima_image.py

    Available lima pixel formats:
    https://gitlab.esrf.fr/limagroup/lima/-/blob/master/common/include/lima/VideoUtils.h

    OpenCV formats:
    https://docs.opencv.org/master/d1/d4f/imgproc_2include_2opencv2_2imgproc_8hpp.html
    """
    (
        magic,
        header_version,
        image_mode,
        image_frame_number,
        image_width,
        image_height,
        endian,
        header_size,
        pad0,
        pad1,
    ) = struct.unpack(VIDEO_HEADER_FORMAT, raw_data[:HEADER_SIZE])

    if magic != VIDEO_MAGIC:
        raise ImageFormatNotSupported(f"Magic header not supported (found {magic:0x})")

    if header_version != 1:
        raise ImageFormatNotSupported(
            f"Image header version not supported (found {header_version})"
        )
    if image_frame_number < 0:
        raise IndexError("Image from lima video_live interface not available yet.")

    try:
        mode = VIDEO_MODES(image_mode)
    except KeyError:
        raise ImageFormatNotSupported(f"Video format unsupported (found {image_mode})")

    try:
        codec = VIDEO_CODECS[mode]
    except KeyError:
        raise ImageFormatNotSupported(f"Video codec unsupported (found {mode})")

    data = numpy.frombuffer(raw_data[HEADER_SIZE:], dtype=codec.dtype)

    # flip w and h as this is how images are saved
    shape = list(codec.shape(image_width, image_height))
    shape[0], shape[1] = shape[1], shape[0]
    data.shape = shape

    if bpp is not None:
        if codec.dtype == numpy.uint16:
            if bpp < 16:
                logger.debug(f"Scaling to 16bit from {bpp}")
                data = data << (16 - bpp)

        if codec.dtype == numpy.uint8:
            if bpp > 8:
                logger.warning(f"Scaling to 8bit from {bpp}, trimming {16-bpp} bits")
                data = data >> (bpp - 8)

    if cv2:
        if codec.opencv_code is not None:
            logger.debug(
                f"Decoding frame with mode {image_mode} using {codec.opencv_code}"
            )
            data = cv2.cvtColor(data, codec.opencv_code)

    if rotation != 0:
        ninties = rotation / 90
        data = numpy.rot90(data, -ninties)

    return data

object

TangoAttrProperty

Source code in daiquiri/core/hardware/tango/object.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class TangoAttrProperty(HardwareProperty):
    def __init__(self, name: str, attrname: callable):
        """
        Attribute
            name: Name of the attribute property
            attrname: Callable returning the attribute name
        """
        HardwareProperty.__init__(self, name=name)
        self._attrname = attrname

    def translate_from(self, value):
        if self.name == "quality":
            return value.name
        if self.name == "data_type":
            return value.name
        if self.name == "display_unit":
            if value is None or value == "No display unit":
                # This values are part of the Tango specification
                return ""
        return value

    def translate_to(self, value):
        return HardwareProperty.translate_from(self, value)

    def get_attrname(self, obj) -> str:
        """Returns the attrname used from a mapping object"""
        return self._attrname(obj)

__init__(self, name, attrname) special

Attribute name: Name of the attribute property attrname: Callable returning the attribute name

Source code in daiquiri/core/hardware/tango/object.py 
def __init__(self, name: str, attrname: callable):
    """
    Attribute
        name: Name of the attribute property
        attrname: Callable returning the attribute name
    """
    HardwareProperty.__init__(self, name=name)
    self._attrname = attrname

get_attrname(self, obj)

Returns the attrname used from a mapping object

Source code in daiquiri/core/hardware/tango/object.py 
def get_attrname(self, obj) -> str:
    """Returns the attrname used from a mapping object"""
    return self._attrname(obj)

translate_from(self, value)

Translate the value from the controls layer to the abstraction layer i.e for getters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/object.py 
def translate_from(self, value):
    if self.name == "quality":
        return value.name
    if self.name == "data_type":
        return value.name
    if self.name == "display_unit":
        if value is None or value == "No display unit":
            # This values are part of the Tango specification
            return ""
    return value

translate_to(self, value)

Translate the value to the controls layer from the abstraction layer i.e for setters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/object.py 
def translate_to(self, value):
    return HardwareProperty.translate_from(self, value)

TangoObject

Source code in daiquiri/core/hardware/tango/object.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
class TangoObject(MappedHardwareObject):
    _protocol = "tango"

    def __init__(self, app, tango_url, **kwargs):
        self._app = app
        self._subscriptions = {}
        self._tango_url = tango_url
        self._object = DeviceProxy(self._tango_url)
        super().__init__(**kwargs)

        self._watchdog = gevent.spawn(self.try_device)
        self._watchdog.name = f"{__name__}.{tango_url}"

        # TODO: Temporary fix for
        # https://github.com/tango-controls/TangoTickets/issues/34
        time.sleep(1)

    def __repr__(self) -> str:
        return f"<Tango: {self._tango_url} ({self.__class__.__name__})>"

    def _connect(self):
        """Try to connect this daiquiri hardware to the remote tango hardware

        raises:
            tango.ConnectionFailed: Connection have failed
            tango.DevFailed: Another tango thing have failed
            Exception: Another thing have failed
        """
        logger.info(f"Trying to connect to {self._tango_url}")
        self._object.ping()
        self._subscribe_device()
        self.set_online(True)
        try:
            self._broadcast()
        except Exception:
            self.set_online(False)
            raise

    def _disconnect(self):
        """Disconnect this daiquiri hardware from the remote tango hardware"""
        self.set_online(False)
        self._unsubscribe_device()

    def try_device(self):
        """Watch dog trying to reconnect to the remote hardware if it was
        disconnected"""
        while True:
            log_failure = logger.exception if self._app.debug else logger.info
            if not self._online:
                try:
                    self._connect()
                except Exception:
                    log_failure(
                        f"Could not connect to {self._tango_url}. Retrying in 10s"
                    )
            gevent.sleep(10)

    def _broadcast(self):
        """Read the connected hardware properties from the proxy and update this
        daiquiri hardware"""
        for name, prop in self.property_map.items():
            self._update(name, prop, self._do_get(prop))

    def _force_subscribe_event(self, tango_attr_name, event_type, callback):
        """Force polling a tango attribute.

        Trying first to subscribe.

        If it fails, setup the server side events.

        raises:
            tango.DevFailed: If the subscription failed
            RuntimeError: If the event configuration have failed
        """
        obj = self._object
        try:
            return self._object.subscribe_event(
                tango_attr_name, event_type, callback, green_mode=tango.GreenMode.Gevent
            )
        except tango.DevFailed as e:
            if hasattr(e, "reason"):
                if e.reason not in [
                    "API_EventPropertiesNotSet",
                    "API_AttributePollingNotStarted",
                ]:
                    raise
            else:
                raise

        info: tango.AttributeInfoEx = obj.get_attribute_config(tango_attr_name)
        changes = []
        valid_period = is_int_str(info.events.per_event.period)
        if not valid_period:
            changes += ["polling period"]
            info.events.per_event.period = "1000"
        valid_rel_change = is_float_str(info.events.ch_event.rel_change)
        valid_abs_change = is_float_str(info.events.ch_event.abs_change)
        if not valid_abs_change and not valid_rel_change:
            changes += ["rel_change"]
            info.events.ch_event.rel_change = "0.001"

        if changes != []:
            msg = " + ".join(changes)
            logger.info(f"Active {msg} for {self._tango_url} {tango_attr_name}")
            try:
                info.name = tango_attr_name
                obj.set_attribute_config(info)
            except tango.DevFailed:
                raise RuntimeError(
                    f"Failed to configure events {self._tango_url} {tango_attr_name}"
                )

        logger.info(f"Try using event for {self._tango_url} {tango_attr_name}")
        return self._object.subscribe_event(
            tango_attr_name, event_type, callback, green_mode=tango.GreenMode.Gevent
        )

    def _subscribe_device(self):
        """Subscribe events for this hardware"""
        log_failure = logger.exception if self._app.debug else logger.info
        for prop in self.property_map.values():
            if isinstance(prop, TangoAttrProperty):
                # FIXME: This have to be properly implemented
                continue
            try:
                tango_attr_name = prop.name
                event_id = self._force_subscribe_event(
                    prop.name, tango.EventType.CHANGE_EVENT, self._push_event
                )
            except Exception:
                log_failure(
                    f"Could not subscribe to property {self._tango_url} {prop.name}"
                )
            else:
                self._subscriptions[tango_attr_name] = event_id

    def _unsubscribe_device(self):
        """Unsubscribe registred events for this daiquiri hardware"""
        log_failure = logger.exception if self._app.debug else logger.info
        for tango_attr_name, eid in self._subscriptions.items():
            try:
                self._object.unsubscribe_event(eid)
            except Exception:
                log_failure(f"Couldnt unsubscribe from {tango_attr_name}")

        self._subscriptions = {}

    def _push_event(self, event: tango.EventData):
        """Callback triggered when the remote tango hardware fire an event"""
        try:
            self._protected_push_event(event)
        except Exception:
            logger.error("Error while processing push_event", exc_info=True)

    def _protected_push_event(self, event: tango.EventData):
        """Callback triggered when the remote tango hardware fire an event

        Any exceptions raised are catched by `_push_event`.
        """
        log_failure = logger.exception if self._app.debug else logger.info

        if not self._online:
            return

        if event.errors:
            error = event.errors[0]
            log_failure(f"Error in push_event for {event.attr_name}: {error.desc}")
            # if error.reason == 'API_EventTimeout':
            self._disconnect()
            return

        if event.attr_value is not None:
            for name, prop in self.property_map.items():
                if event.attr_value.name == prop.name:
                    self._update(name, prop, event.attr_value.value)
                    break

    def _do_set(self, prop: HardwareProperty, value):
        obj = self._object
        if isinstance(prop, TangoAttrProperty):
            if prop.name != "value":
                raise RuntimeError(
                    f"Setter for '{prop.name}' attribute config is not implemented"
                )
            attrname = prop.get_attrname(self)
            return obj.write_attribute(attrname, value)
        else:
            return obj.write_attribute(prop.name, value)

    def _do_get(self, prop: HardwareProperty):
        obj = self._object
        if isinstance(prop, TangoAttrProperty):
            attrname = prop.get_attrname(self)
            if prop.name == "name":
                return attrname
            if prop.name == "value":
                return obj.read_attribute(attrname).value
            elif prop.name == "data_type":
                return obj.read_attribute(attrname).type
            elif prop.name == "quality":
                return obj.read_attribute(attrname).quality
            elif prop.name == "unit":
                return obj.get_attribute_config(attrname).unit
            elif prop.name == "format":
                return obj.get_attribute_config(attrname).format
            elif prop.name == "display_unit":
                return obj.get_attribute_config(attrname).display_unit
            elif prop.name == "label":
                return obj.get_attribute_config(attrname).label
            elif prop.name == "description":
                return obj.get_attribute_config(attrname).description
            else:
                raise RuntimeError(f"Unsupported {prop.name} attribute")

        getter = prop.getter
        if getter is not None:
            return getter(self)
        return obj.read_attribute(prop.name).value

    def _do_call(self, function, value, **kwargs):
        return self._object.command_inout(function, value, **kwargs)

try_device(self)

Watch dog trying to reconnect to the remote hardware if it was disconnected

Source code in daiquiri/core/hardware/tango/object.py 
def try_device(self):
    """Watch dog trying to reconnect to the remote hardware if it was
    disconnected"""
    while True:
        log_failure = logger.exception if self._app.debug else logger.info
        if not self._online:
            try:
                self._connect()
            except Exception:
                log_failure(
                    f"Could not connect to {self._tango_url}. Retrying in 10s"
                )
        gevent.sleep(10)

TangoStateProperty

Source code in daiquiri/core/hardware/tango/object.py 
32
33
34
35
36
37
38
39
40
class TangoStateProperty(HardwareProperty):
    def translate_from(self, value):
        return value.name

    def translate_to(self, value):
        state = tango.DevState.__members__.get(value)
        if state is None:
            return tango.DevState.UNKNOWN
        return state

translate_from(self, value)

Translate the value from the controls layer to the abstraction layer i.e for getters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/object.py 
def translate_from(self, value):
    return value.name

translate_to(self, value)

Translate the value to the controls layer from the abstraction layer i.e for setters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/object.py 
def translate_to(self, value):
    state = tango.DevState.__members__.get(value)
    if state is None:
        return tango.DevState.UNKNOWN
    return state

pump

PumpStateProperty

Source code in daiquiri/core/hardware/tango/pump.py 
14
15
16
17
18
19
20
21
class PumpStateProperty(HardwareProperty):
    def translate_from(self, value):
        val_map = {DevState.ON: "ON", DevState.OFF: "OFF", DevState.UNKNOWN: "UNKNOWN"}
        for k, v in val_map.items():
            if k == value:
                return v

        return "UNKNOWN"

translate_from(self, value)

Translate the value from the controls layer to the abstraction layer i.e for getters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/pump.py 
def translate_from(self, value):
    val_map = {DevState.ON: "ON", DevState.OFF: "OFF", DevState.UNKNOWN: "UNKNOWN"}
    for k, v in val_map.items():
        if k == value:
            return v

    return "UNKNOWN"

shutter

ShutterStateProperty

Source code in daiquiri/core/hardware/tango/shutter.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class ShutterStateProperty(HardwareProperty):
    def translate_from(self, value):
        val_map = {
            DevState.CLOSE: "CLOSED",
            DevState.DISABLE: "DISABLED",
            DevState.OPEN: "OPEN",
            DevState.FAULT: "FAULT",
            DevState.UNKNOWN: "UNKNOWN",
            DevState.RUNNING: "OPEN",  # The RUNNING state is not very useful in user point of view
            DevState.STANDBY: "STANDBY",
            DevState.MOVING: "MOVING",
        }
        for k, v in val_map.items():
            if k == value:
                return v

        return "UNKNOWN"

translate_from(self, value)

Translate the value from the controls layer to the abstraction layer i.e for getters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/shutter.py 
def translate_from(self, value):
    val_map = {
        DevState.CLOSE: "CLOSED",
        DevState.DISABLE: "DISABLED",
        DevState.OPEN: "OPEN",
        DevState.FAULT: "FAULT",
        DevState.UNKNOWN: "UNKNOWN",
        DevState.RUNNING: "OPEN",  # The RUNNING state is not very useful in user point of view
        DevState.STANDBY: "STANDBY",
        DevState.MOVING: "MOVING",
    }
    for k, v in val_map.items():
        if k == value:
            return v

    return "UNKNOWN"

valve

ValveStateProperty

Source code in daiquiri/core/hardware/tango/valve.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
class ValveStateProperty(HardwareProperty):
    def translate_from(self, value):
        val_map = {
            DevState.OPEN: "OPEN",
            DevState.CLOSE: "CLOSED",
            DevState.UNKNOWN: "UNKNOWN",
            DevState.FAULT: "FAULT",
        }
        for k, v in val_map.items():
            if k == value:
                return v

        return "UNKNOWN"

translate_from(self, value)

Translate the value from the controls layer to the abstraction layer i.e for getters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/valve.py 
def translate_from(self, value):
    val_map = {
        DevState.OPEN: "OPEN",
        DevState.CLOSE: "CLOSED",
        DevState.UNKNOWN: "UNKNOWN",
        DevState.FAULT: "FAULT",
    }
    for k, v in val_map.items():
        if k == value:
            return v

    return "UNKNOWN"

wago

WagoStateProperty

Source code in daiquiri/core/hardware/tango/wago.py 
15
16
17
18
19
20
21
class WagoStateProperty(HardwareProperty):
    def translate_from(self, value):
        val_map = {DevState.ON: "ON", DevState.OFF: "OFF", DevState.UNKNOWN: "UNKNOWN"}
        for k, v in val_map.items():
            if k == value:
                return v
        return "UNKNOWN"

translate_from(self, value)

Translate the value from the controls layer to the abstraction layer i.e for getters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/wago.py 
def translate_from(self, value):
    val_map = {DevState.ON: "ON", DevState.OFF: "OFF", DevState.UNKNOWN: "UNKNOWN"}
    for k, v in val_map.items():
        if k == value:
            return v
    return "UNKNOWN"

WagoStateProperty

Source code in daiquiri/core/hardware/tango/wago.py 
15
16
17
18
19
20
21
class WagoStateProperty(HardwareProperty):
    def translate_from(self, value):
        val_map = {DevState.ON: "ON", DevState.OFF: "OFF", DevState.UNKNOWN: "UNKNOWN"}
        for k, v in val_map.items():
            if k == value:
                return v
        return "UNKNOWN"

translate_from(self, value)

Translate the value from the controls layer to the abstraction layer i.e for getters

Parameters:

Name Type Description Default
value

The property value to translate.

 required

Returns:

Type Description

The translated value
Source code in daiquiri/core/hardware/tango/wago.py 
def translate_from(self, value):
    val_map = {DevState.ON: "ON", DevState.OFF: "OFF", DevState.UNKNOWN: "UNKNOWN"}
    for k, v in val_map.items():
        if k == value:
            return v
    return "UNKNOWN"