Skip to content

tango

TangoHOConfig marshmallow-model

The Tango Hardware Object Config Schema

Source code in daiquiri/core/hardware/tango/__init__.py
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
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
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
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
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
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
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
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
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
@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
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
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
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
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
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
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
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
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"