abstract

AbstractHardwareProperty2

Implement an automomous property.

Source code in daiquiri/core/hardware/abstract/__init__.py

class AbstractHardwareProperty2(ABC):
    """
    Implement an automomous property.
    """

    def __init__(self, parent: MappedHardwareObject):
        self._object = parent
        self.__update = None

    def _connect_update(self, callback):
        """Called by daiquiri at the initialization to be informed by changes"""
        assert self.__update is None
        self.__update = callback

    def emit_update(self, value):
        """To be called by the property itself when the hardware property have changed"""
        update = self.__update
        if update is not None:
            update(value)

    @abstractmethod
    def read_hardware(self, obj):
        """Read the value from the subsystem"""
        ...

    @abstractmethod
    def write_hardware(self, obj, value):
        """Write the value to the subsystem"""
        ...

    @abstractmethod
    def connect_hardware(self, obj):
        """Do the connection of this property to the subsystem"""
        ...

connect_hardware(self, obj)

Do the connection of this property to the subsystem

Source code in daiquiri/core/hardware/abstract/__init__.py

@abstractmethod
def connect_hardware(self, obj):
    """Do the connection of this property to the subsystem"""
    ...

emit_update(self, value)

To be called by the property itself when the hardware property have changed

Source code in daiquiri/core/hardware/abstract/__init__.py

def emit_update(self, value):
    """To be called by the property itself when the hardware property have changed"""
    update = self.__update
    if update is not None:
        update(value)

read_hardware(self, obj)

Read the value from the subsystem

Source code in daiquiri/core/hardware/abstract/__init__.py

@abstractmethod
def read_hardware(self, obj):
    """Read the value from the subsystem"""
    ...

write_hardware(self, obj, value)

Write the value to the subsystem

Source code in daiquiri/core/hardware/abstract/__init__.py

@abstractmethod
def write_hardware(self, obj, value):
    """Write the value to the subsystem"""
    ...

HardwareObject

Base HardwareObject from which all inherit

The base hardware object defines the objects procotol, type, its properties and callables schema, and mechanisms to subscribe to property changes

Attributes:

Name	Type	Description
_object	obj	The instance of the control system object.
_protocol	str	The protocol for this object, i.e. bliss
_type	str	The object type, i.e. motor, shutter

Source code in daiquiri/core/hardware/abstract/__init__.py

class HardwareObject(ABC):
    """Base HardwareObject from which all inherit

    The base hardware object defines the objects procotol, type, its properties
    and callables schema, and mechanisms to subscribe to property changes

    Attributes:
        _object (obj): The instance of the control system object.
        _protocol (str): The protocol for this object, i.e. bliss
        _type (str): The object type, i.e. motor, shutter

    """

    _object = None
    _online = False
    _state_ok = []

    _protocol = None
    _type = None

    _properties = HardwareSchema()
    _callables = HardwareSchema()

    def schema_name(self):
        ty = self._type
        return ty[0].upper() + ty[1:]

    def __init__(self, *args, **kwargs):
        self._callbacks = {}
        self._online_callbacks = []

        self._id = kwargs.get("id", None)
        self._name = kwargs.get("name", "")
        self._require_staff = kwargs.get("require_staff", False)
        self._alias = None
        self._user_tags = []

    def get(self, prop: str):
        """Get a property from a hardware object

        First checks the property is defined in the objects
        property schema, then delegates to the local getter
        implementation _get

        Arguments:
            prop: The property to retreive.

        Returns:
            The property value if the property exists, otherwise
            rasises an exception

        """
        if not self._online:
            return

        if prop in self._properties:
            value = self._get(prop)
            return value
        else:
            raise KeyError("Unknown property {p}".format(p=prop))

    def set(self, prop: str, value):
        """Set a property on a hardware object

        First checks the property is defined in the objects
        property schema, then delegates to the local setter
        implementation _set

        Arguments:
            prop: The property to set.
            value: The property to set.

        Returns:
            The the result from the object setter if the property exists
            otherwise raises an exception

        """
        if not self._online:
            return

        if prop in self._properties:
            if self._properties.read_only(prop):
                raise AttributeError("Property {p} is read only".format(p=prop))

            value = self._properties.validate(prop, value)
            return self._set(prop, value)
        else:
            raise KeyError("Unknown property {p}".format(p=prop))

    def get_subobject_configs(self):
        """Returns a list of referenced objects own by this object."""
        return []

    def call(self, function, value, **kwargs):
        """Calls a function on a hardware object

        First checks the function is defined in the objects
        callables schema, then delegates to the local call
        implementation _call

        Args:
            function (str): The function to call.
            value: The value to call the function with.

        Returns:
            The the result from the object function if the function exists
            otherwise rasises an exception

        """
        if not self._online:
            return

        if function in self._callables:
            value = self._callables.validate(function, value)
            # try:
            ret = self._call(function, value, **kwargs)
            if ret:
                return ret
            else:
                return True
            # except Exception as e:
            #     return e
        else:
            raise KeyError("Unknown function {fn}".format(fn=function))

    @abstractmethod
    def _get(self, prop: str):
        """Local implementation of getter"""
        pass

    @abstractmethod
    def _set(self, prop: str, value):
        """Local implementation of setter"""
        pass

    @abstractmethod
    def _call(self, function, value):
        """Local implementation of call"""
        pass

    def state(self, **kwargs):
        """Gets the current state of a hardware object

        Builds a dictionary of the basic info of the object, plus its properties, and
        callables.

        Returns:
            A dict of the hardware object status

        """
        info = {
            k: getattr(self, "_" + k)
            for k in ["name", "type", "id", "protocol", "online"]
        }
        info["callables"] = [k for k in self._callables]
        info["errors"] = []
        properties = {}
        for p in self._properties:
            try:
                properties[p] = self.get(p)
            except Exception as e:
                properties[p] = None
                info["online"] = False
                info["errors"].append(
                    {
                        "property": p,
                        "exception": str(e),
                        "traceback": "".join(traceback.format_tb(e.__traceback__)),
                    }
                )
                logger.exception(f"Couldn't get property `{p}` for `{self.name()}`")

        try:
            info["properties"] = self._properties.dump(properties)
        except Exception:
            logger.error(
                "Error while serializing %s (%s)",
                self._name,
                properties,
                exc_info=True,
            )
            info["properties"] = {}
            info["online"] = False

        try:
            info["properties"]["state_ok"] = self.state_ok()
        except Exception:
            logger.debug(
                f"Could not determine `state_ok` for {self._name}", exec_info=True
            )

        info["require_staff"] = self.require_staff()
        info["alias"] = self.alias()
        info["user_tags"] = self.user_tags()

        return info

    def schema(self):
        """Returns the schema for the current hardware object

        The hardware schema is built from the `HardwareObjectBaseSchema`
        and the object specific _property and _callable schema
        (both instances of `HardwareSchema`)

        Returns:
            An instance of a schema

        """
        schema = type(
            f"HW{self.schema_name()}Schema",
            (HardwareObjectBaseSchema,),
            {
                "properties": fields.Nested(self._properties.__class__),
                "callables": fields.Nested(self._callables.__class__),
            },
        )
        return schema()

    def set_online(self, state):
        """Set the online state of the device

        Sets the state and execute any registered callbacks

        Args:
            state (boolean): Set the online state
        """
        self._online = state

        for cb in self._online_callbacks:
            cb(self, self._online)

    def subscribe_online(self, fn):
        """Subscribe to the online state of the hardware object

        Add a function to a list of callbacks for when the online state of the object change

        Args:
            fn: (:callable) The function to call when this property changes.

        """
        if not callable(fn):
            raise AttributeError("Callback function must be callable")

        if not (fn in self._online_callbacks):
            self._online_callbacks.append(fn)

    def id(self):
        return self._id

    def name(self):
        return self._name

    def type(self):
        return self._type

    def object(self):
        return self._object

    def online(self):
        return self._online

    def alias(self) -> typing.Optional[str]:
        return self._alias

    def user_tags(self) -> typing.List[str]:
        return self._user_tags

    def require_staff(self):
        return self._require_staff

    def state_ok(self):
        """Returns if the current object state is `ok`"""
        state = self.get("state")
        if isinstance(state, list):
            st = False
            for ok in self._state_ok:
                if ok in state:
                    st = True
            return st

        else:
            return state in self._state_ok

    def subscribe(self, prop: str, fn):
        """Subscribe to property changes on the hardware object

        Add a function to a list of callbacks for when properties on the object change

        Arguments:
            prop: The property to subscribe to. Can pass 'all' to subscribe to all changes
            fn: (:callable) The function to call when this property changes.

        """
        if not callable(fn):
            raise AttributeError("Callback function must be callable")

        if prop in self._properties or prop == "all":
            if not (prop in self._callbacks):
                self._callbacks[prop] = []

            if not (fn in self._callbacks[prop]):
                self._callbacks[prop].append(fn)

        else:
            raise KeyError(f"No such property: {prop}")

    def unsubscribe(self, prop: str, fn):
        """Unsubscribe from a property change on the hardware object

        Arguments:
            prop: The property to unsubscribe from.
            fn: (:callable) The function to unsubscribe

        """
        if prop in self._callbacks:
            if fn in self._callbacks[prop]:
                # logger.debug("Unsubscribe from property %s, %s", prop, fn)
                self._callbacks[prop].remove(fn)
                return True

        return False

    def _execute_callbacks(self, name, value):
        if name in self._callbacks:
            for cb in self._callbacks[name]:
                cb(self, name, value)

        if "all" in self._callbacks:
            for cb in self._callbacks["all"]:
                cb(self, name, value)

    def _update(self, name: str, prop: HardwareProperty, value):
        """Internal function to call when a property has changed

        This delegates to all subscribes that a property has changed

        Arguments:
            name: Property name of the abstract device
            prop: Property of the control system hardware
            value: The new value (from the control system).

        """
        # logger.debug('{c}._update {n} - {p}: {v}'.format(c=self.__class__.__name__, n=self._address, p=prop, v=value))
        if not isinstance(prop, AbstractHardwareProperty2):
            value = prop.translate_from(value)
        if name in self._properties:
            self._execute_callbacks(name, value)

            # If `state` changes emit an updated `state_ok`
            if name == "state":
                self._execute_callbacks("state_ok", self.state_ok())

call(self, function, value, **kwargs)

Calls a function on a hardware object

First checks the function is defined in the objects callables schema, then delegates to the local call implementation _call

Parameters:

Name	Type	Description	Default
function	str	The function to call.	required
value		The value to call the function with.	required

Returns:

Type	Description
	The the result from the object function if the function exists otherwise rasises an exception

Source code in daiquiri/core/hardware/abstract/__init__.py

def call(self, function, value, **kwargs):
    """Calls a function on a hardware object

    First checks the function is defined in the objects
    callables schema, then delegates to the local call
    implementation _call

    Args:
        function (str): The function to call.
        value: The value to call the function with.

    Returns:
        The the result from the object function if the function exists
        otherwise rasises an exception

    """
    if not self._online:
        return

    if function in self._callables:
        value = self._callables.validate(function, value)
        # try:
        ret = self._call(function, value, **kwargs)
        if ret:
            return ret
        else:
            return True
        # except Exception as e:
        #     return e
    else:
        raise KeyError("Unknown function {fn}".format(fn=function))

get(self, prop)

Get a property from a hardware object

First checks the property is defined in the objects property schema, then delegates to the local getter implementation _get

Parameters:

Name	Type	Description	Default
prop	str	The property to retreive.	required

Returns:

Type	Description
	The property value if the property exists, otherwise rasises an exception

Source code in daiquiri/core/hardware/abstract/__init__.py

def get(self, prop: str):
    """Get a property from a hardware object

    First checks the property is defined in the objects
    property schema, then delegates to the local getter
    implementation _get

    Arguments:
        prop: The property to retreive.

    Returns:
        The property value if the property exists, otherwise
        rasises an exception

    """
    if not self._online:
        return

    if prop in self._properties:
        value = self._get(prop)
        return value
    else:
        raise KeyError("Unknown property {p}".format(p=prop))

get_subobject_configs(self)

Returns a list of referenced objects own by this object.

Source code in daiquiri/core/hardware/abstract/__init__.py

def get_subobject_configs(self):
    """Returns a list of referenced objects own by this object."""
    return []

schema(self)

Returns the schema for the current hardware object

The hardware schema is built from the HardwareObjectBaseSchema and the object specific _property and _callable schema (both instances of HardwareSchema)

Returns:

Type	Description
	An instance of a schema

Source code in daiquiri/core/hardware/abstract/__init__.py

def schema(self):
    """Returns the schema for the current hardware object

    The hardware schema is built from the `HardwareObjectBaseSchema`
    and the object specific _property and _callable schema
    (both instances of `HardwareSchema`)

    Returns:
        An instance of a schema

    """
    schema = type(
        f"HW{self.schema_name()}Schema",
        (HardwareObjectBaseSchema,),
        {
            "properties": fields.Nested(self._properties.__class__),
            "callables": fields.Nested(self._callables.__class__),
        },
    )
    return schema()

set(self, prop, value)

Set a property on a hardware object

First checks the property is defined in the objects property schema, then delegates to the local setter implementation _set

Parameters:

Name	Type	Description	Default
prop	str	The property to set.	required
value		The property to set.	required

Returns:

Type	Description
	The the result from the object setter if the property exists otherwise raises an exception

Source code in daiquiri/core/hardware/abstract/__init__.py

def set(self, prop: str, value):
    """Set a property on a hardware object

    First checks the property is defined in the objects
    property schema, then delegates to the local setter
    implementation _set

    Arguments:
        prop: The property to set.
        value: The property to set.

    Returns:
        The the result from the object setter if the property exists
        otherwise raises an exception

    """
    if not self._online:
        return

    if prop in self._properties:
        if self._properties.read_only(prop):
            raise AttributeError("Property {p} is read only".format(p=prop))

        value = self._properties.validate(prop, value)
        return self._set(prop, value)
    else:
        raise KeyError("Unknown property {p}".format(p=prop))

set_online(self, state)

Set the online state of the device

Sets the state and execute any registered callbacks

Parameters:

Name	Type	Description	Default
state	boolean	Set the online state	required

Source code in daiquiri/core/hardware/abstract/__init__.py

def set_online(self, state):
    """Set the online state of the device

    Sets the state and execute any registered callbacks

    Args:
        state (boolean): Set the online state
    """
    self._online = state

    for cb in self._online_callbacks:
        cb(self, self._online)

state(self, **kwargs)

Gets the current state of a hardware object

Builds a dictionary of the basic info of the object, plus its properties, and callables.

Returns:

Type	Description
	A dict of the hardware object status

Source code in daiquiri/core/hardware/abstract/__init__.py

def state(self, **kwargs):
    """Gets the current state of a hardware object

    Builds a dictionary of the basic info of the object, plus its properties, and
    callables.

    Returns:
        A dict of the hardware object status

    """
    info = {
        k: getattr(self, "_" + k)
        for k in ["name", "type", "id", "protocol", "online"]
    }
    info["callables"] = [k for k in self._callables]
    info["errors"] = []
    properties = {}
    for p in self._properties:
        try:
            properties[p] = self.get(p)
        except Exception as e:
            properties[p] = None
            info["online"] = False
            info["errors"].append(
                {
                    "property": p,
                    "exception": str(e),
                    "traceback": "".join(traceback.format_tb(e.__traceback__)),
                }
            )
            logger.exception(f"Couldn't get property `{p}` for `{self.name()}`")

    try:
        info["properties"] = self._properties.dump(properties)
    except Exception:
        logger.error(
            "Error while serializing %s (%s)",
            self._name,
            properties,
            exc_info=True,
        )
        info["properties"] = {}
        info["online"] = False

    try:
        info["properties"]["state_ok"] = self.state_ok()
    except Exception:
        logger.debug(
            f"Could not determine `state_ok` for {self._name}", exec_info=True
        )

    info["require_staff"] = self.require_staff()
    info["alias"] = self.alias()
    info["user_tags"] = self.user_tags()

    return info

state_ok(self)

Returns if the current object state is ok

Source code in daiquiri/core/hardware/abstract/__init__.py

def state_ok(self):
    """Returns if the current object state is `ok`"""
    state = self.get("state")
    if isinstance(state, list):
        st = False
        for ok in self._state_ok:
            if ok in state:
                st = True
        return st

    else:
        return state in self._state_ok

subscribe(self, prop, fn)

Subscribe to property changes on the hardware object

Add a function to a list of callbacks for when properties on the object change

Parameters:

Name	Type	Description	Default
prop	str	The property to subscribe to. Can pass 'all' to subscribe to all changes	required
fn		(:callable) The function to call when this property changes.	required

Source code in daiquiri/core/hardware/abstract/__init__.py

def subscribe(self, prop: str, fn):
    """Subscribe to property changes on the hardware object

    Add a function to a list of callbacks for when properties on the object change

    Arguments:
        prop: The property to subscribe to. Can pass 'all' to subscribe to all changes
        fn: (:callable) The function to call when this property changes.

    """
    if not callable(fn):
        raise AttributeError("Callback function must be callable")

    if prop in self._properties or prop == "all":
        if not (prop in self._callbacks):
            self._callbacks[prop] = []

        if not (fn in self._callbacks[prop]):
            self._callbacks[prop].append(fn)

    else:
        raise KeyError(f"No such property: {prop}")

subscribe_online(self, fn)

Subscribe to the online state of the hardware object

Add a function to a list of callbacks for when the online state of the object change

Parameters:

Name	Type	Description	Default
fn		(:callable) The function to call when this property changes.	required

Source code in daiquiri/core/hardware/abstract/__init__.py

def subscribe_online(self, fn):
    """Subscribe to the online state of the hardware object

    Add a function to a list of callbacks for when the online state of the object change

    Args:
        fn: (:callable) The function to call when this property changes.

    """
    if not callable(fn):
        raise AttributeError("Callback function must be callable")

    if not (fn in self._online_callbacks):
        self._online_callbacks.append(fn)

unsubscribe(self, prop, fn)

Unsubscribe from a property change on the hardware object

Parameters:

Name	Type	Description	Default
prop	str	The property to unsubscribe from.	required
fn		(:callable) The function to unsubscribe	required

Source code in daiquiri/core/hardware/abstract/__init__.py

def unsubscribe(self, prop: str, fn):
    """Unsubscribe from a property change on the hardware object

    Arguments:
        prop: The property to unsubscribe from.
        fn: (:callable) The function to unsubscribe

    """
    if prop in self._callbacks:
        if fn in self._callbacks[prop]:
            # logger.debug("Unsubscribe from property %s, %s", prop, fn)
            self._callbacks[prop].remove(fn)
            return True

    return False

HardwareProperty

Describe a property from the device controls system.

Provides translation function between the controls system specific nomenclature and the abstract one. This can be overwritten.

Parameters:

Name	Type	Description	Default
name	str	Name used by the controls system	required

Source code in daiquiri/core/hardware/abstract/__init__.py

class HardwareProperty:
    """Describe a property from the device controls system.

    Provides translation function between the controls system specific
    nomenclature and the abstract one. This can be overwritten.

    Arguments:
        name: Name used by the controls system
    """

    def __init__(self, name: str, getter: callable = None):
        self._name = name
        self._getter = getter

    @property
    def name(self) -> str:
        return self._name

    @property
    def getter(self) -> typing.Optional[callable]:
        return self._getter

    def translate_from(self, value):
        """Translate the value from the controls layer to the abstraction layer
        i.e for getters

        Arguments:
            value: The property value to translate.

        Returns:
            The translated value
        """
        return value

    def translate_to(self, value):
        """Translate the value to the controls layer from the abstraction layer
        i.e for setters

        Arguments:
            value: The property value to translate.

        Returns:
            The translated value
        """
        return value

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/abstract/__init__.py

def translate_from(self, value):
    """Translate the value from the controls layer to the abstraction layer
    i.e for getters

    Arguments:
        value: The property value to translate.

    Returns:
        The translated value
    """
    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/abstract/__init__.py

def translate_to(self, value):
    """Translate the value to the controls layer from the abstraction layer
    i.e for setters

    Arguments:
        value: The property value to translate.

    Returns:
        The translated value
    """
    return value

MappedHardwareObject

Hardware object that maps properties via a simple dict

HardwareObject that has a simple map between abstract properties and their actual properties on the object with fallback to a function on the parent

Source code in daiquiri/core/hardware/abstract/__init__.py

class MappedHardwareObject(HardwareObject):
    """Hardware object that maps properties via a simple dict

    HardwareObject that has a simple map between abstract properties and their
    actual properties on the object with fallback to a function on the parent
    """

    PROPERTY_MAP: typing.Dict[str, HardwareProperty] = {}
    CALLABLE_MAP: typing.Dict[str, str] = {}

    def __init__(self, *args, **kwargs):
        HardwareObject.__init__(self, *args, **kwargs)
        self._property_map: typing.Dict[
            str, HardwareProperty | AbstractHardwareProperty2
        ] = self._create_properties()
        self._callable_map: typing.Dict[str, str] = dict(self.CALLABLE_MAP)

        for name, prop in self._property_map.items():
            if isinstance(prop, AbstractHardwareProperty2):
                prop._connect_update(functools.partial(self._update, name, prop))

    def _create_properties(
        self,
    ) -> typing.Dict[str, HardwareProperty | AbstractHardwareProperty2]:
        """Return the properties to be used for this hardware object

        The default implementation reads the descriptions from the
        class attribute `PROPERTY_MAP`.
        """
        return dict(self.PROPERTY_MAP)

    def _set(self, prop: str, value):
        """Set a property on the child object

        First try from the simple property map which maps properties to attributes
        on the child object. Delegates to _do_set which locally implements the setter

        Second, if not in the map, try calling the function _get_<prop> on the parent

        Args:
            prop: The property to set.
            value: Its value.
        """
        hprop = self._property_map.get(prop)
        if hprop is not None:
            if not isinstance(hprop, AbstractHardwareProperty2):
                value = hprop.translate_to(value)
            self._do_set(hprop, value)
        else:
            raise KeyError(
                f"Couldnt find a setter for property `{prop}` on `{self.name()}`"
            )

    @abstractmethod
    def _do_set(self, prop, value):
        """Local implementation of how to set the property on the object"""
        pass

    def _get(self, prop: str):
        """Get a property from the child object

        First try from the simple property map which maps properties to attributes
        on the child object. Delegates to _do_get which locally implements the getter

        Second, if not in the map, try calling the function _get_<prop> on the parent

        Arguments:
            prop: The property to set.

        Returns:
            The property value
        """
        hprop = self._property_map.get(prop)
        if hprop is not None:
            try:
                hvalue = self._do_get(hprop)
                if not isinstance(hprop, AbstractHardwareProperty2):
                    hvalue = hprop.translate_from(hvalue)
                return hvalue
            except NotImplementedError:
                logger.info("Object %s does not implement %s", self._id, prop)

        # `state_ok` is dynamically generated from `state`
        elif prop == "state_ok":
            pass
        else:
            raise KeyError(
                f"Couldnt find a getter for property `{prop}` on `{self.name()}`"
            )

    @abstractmethod
    def _do_get(self, prop):
        """Local implementation of how to get the property from the object"""
        pass

    def _call(self, function, value, **kwargs):
        """Call a function on the child object

        First try from the simple function map which maps to function names
        on the child object. Delegates to _do_call which locally implements the getter

        Second, if not in the map, try calling the function _call_<fn> on the parent

        Args:
            function (str): The function to call.
            value: The value to call the function with

        Returns:
            True if function successfully called
        """
        if function in self._callable_map:
            ret = self._do_call(function, value, **kwargs)
        elif hasattr(self, "_call_{fn}".format(fn=function)):
            ret = getattr(self, "_call_{fn}".format(fn=function))(value, **kwargs)

        else:
            raise KeyError(
                f"Couldnt find a handler for function `{function}` on `{self.name()}`"
            )

        if ret is None:
            # When a function returns nothing
            ret = True

        return ret

    @abstractmethod
    def _do_call(self, function, value, **kwargs):
        """
        Local implementation of how to get the function from the object
        """
        pass

ProtocolHandler

A protocol handler instantiates a hardware object from a specific protocol

i.e. initialise a motor object from bliss, or a shutter from tango

Source code in daiquiri/core/hardware/abstract/__init__.py

class ProtocolHandler:
    """A protocol handler instantiates a hardware object from a specific protocol

    i.e. initialise a motor object from bliss, or a shutter from tango
    """

    library = None

    def __init__(self, *args, **kwargs):
        self._app = kwargs.get("app")

    def disconnect(self):
        """Called at the termination of the handler.

        Implement it if the protocol hold resources.
        """
        pass

    @abstractmethod
    def get(self, *args, **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:
            The hardware object instance for the specific protocol

        """
        pass

disconnect(self)

Called at the termination of the handler.

Implement it if the protocol hold resources.

Source code in daiquiri/core/hardware/abstract/__init__.py

def disconnect(self):
    """Called at the termination of the handler.

    Implement it if the protocol hold resources.
    """
    pass

get(self, *args, **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/abstract/__init__.py

@abstractmethod
def get(self, *args, **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:
        The hardware object instance for the specific protocol

    """
    pass

camera

Camera

Source code in daiquiri/core/hardware/abstract/camera.py

class Camera(HardwareObject):
    _type = "camera"
    _state_ok = [CameraStatuses[0], CameraStatuses[1]]

    _properties = CameraPropertiesSchema()
    _callables = CameraCallablesSchema()

    def frame(self):
        """Return a frame from the camera"""
        return np.array([])

    def _call_save(self, path, type="PNG"):
        """Return a frame from the camera to disk"""
        frame = self.frame()
        if frame is None:
            raise RuntimeError("Could not get frame from camera")

        # Possible the frame will not be uint8, convert it
        if frame.dtype != np.uint8:
            # opencv does not support uint32?
            # https://github.com/MouseLand/cellpose/issues/937
            if frame.dtype == np.uint32:
                frame = frame.astype(np.float32)
            frame = cv2.normalize(frame, None, 0, 255, cv2.NORM_MINMAX, dtype=cv2.CV_8U)
            frame = frame.astype(np.uint8)

        im = Image.fromarray(frame)
        if im.mode != "RGB":
            im = im.convert("RGB")

        im.save(path, type, quality=100)

frame(self)

Return a frame from the camera

Source code in daiquiri/core/hardware/abstract/camera.py

def frame(self):
    """Return a frame from the camera"""
    return np.array([])

presetmanager

PresetmanagerCallablesSchema marshmallow-model

Source code in daiquiri/core/hardware/abstract/presetmanager.py

class PresetmanagerCallablesSchema(HardwareSchema):
    apply = fields.Str(required=True, metadata={"description": "The preset to apply"})

Fields:

Name	Type	Properties	Description
apply	String	Required	The preset to apply

scansource

ScanSource

The scan source interface

Source code in daiquiri/core/hardware/abstract/scansource.py

class ScanSource(ABC):
    """The scan source interface"""

    def __init__(
        self,
        config: Dict[str, Any],
        app=None,
        source_config: Dict[str, Any] = {},
    ):
        self._app = app
        self._config = config
        self._source_config = source_config
        self._session_name = source_config.get("session")

        self._new_scan_watchers = []
        self._new_data_watchers = []
        self._end_scan_watchers = []
        self._new_0d_data_watchers = []

    def close(self):
        """Clean up the service at the end.

        After this call, the component should not be accessed anymore
        """
        pass

    @abstractmethod
    def get_scans(self, scanid=None):
        """Get a list of scans

        Returns a list of scan dicts, can be filtered to only scanid to return
        details of a specific scan

        Args:
            scanid (int): Scan id to return

        Returns:
            scans (list): A list of valid `ScanSchema` dicts
        """
        pass

    # @marshal_with(ScanDataSchema())
    @abstractmethod
    def get_scan_data(self, scanid, per_page=25, page=1):
        """Return scan data for the specific scan

        Args:
            scanid (int): The scan id
            per_page (int): Number of items to return per page
            page (int): Page of data to return

        Returns:
            data (dict): A valid `ScanDataSchema` dict

        """
        pass

    @abstractmethod
    def get_scan_spectra(self, scanid, point=0, allpoints=False):
        """Return a scan specific for the specific scan

        Args:
            scanid (int): The scan id
            point (int): The image number to return

        Returns:
            data (dict): A valid `ScanSpectraSchema` dict

        """
        pass

    @abstractmethod
    def get_scan_image(self, scanid, node_name, image_no):
        """Return a scan image for the specific scan

        Args:
            scanid (int): The scan id
            node_name (str): The node for the requested image
            image_no (int): The image number to return

        Returns:
            data (dict): A valid `ScanDataSchema` dict

        """
        pass

    def watch_new_scan(self, fn):
        """Register a callback for when a new scan is started"""
        if not callable(fn):
            raise AttributeError("Callback function must be callable")

        if not (fn in self._new_scan_watchers):
            self._new_scan_watchers.append(fn)
        else:
            logger.warning(
                "Function {f} is already subscribed to new scan".format(f=fn)
            )

    def watch_new_data(self, fn):
        """Register a callback for when there is new data for a scan"""
        if not callable(fn):
            raise AttributeError("Callback function must be callable")

        if not (fn in self._new_data_watchers):
            self._new_data_watchers.append(fn)
        else:
            logger.warning(
                "Function {f} is already subscribed to new data".format(f=fn)
            )

    def watch_new_0d_data(self, fn):
        """Register a callback for when there is new 0d data for a scan"""
        if not callable(fn):
            raise AttributeError("Callback function must be callable")

        if not (fn in self._new_0d_data_watchers):
            self._new_0d_data_watchers.append(fn)
        else:
            logger.warning(
                "Function {f} is already subscribed to new 0d data".format(f=fn)
            )

    def watch_end_scan(self, fn):
        """Register a callback for a scan ends"""
        if not callable(fn):
            raise AttributeError("Callback function must be callable")

        if not (fn in self._end_scan_watchers):
            self._end_scan_watchers.append(fn)
        else:
            logger.warning(
                "Function {f} is already subscribed to end scan".format(f=fn)
            )

    def _emit_new_scan_event(
        self, scanid: int, type: str, title: str, metadata: Dict[str, Any]
    ) -> None:
        """Emit an event when a new scan starts

        Kwargs:
            scanid: The id of the new scan
            type: The scan type, ascan, amesh, etc
            title: A short description of the scan
            metadata: Any related metadata (not currently used)
        """
        for cb in self._new_scan_watchers:
            try:
                cb(scanid, type, type, metadata=metadata)
            except Exception:
                logger.error("Error during scan start callback", exc_info=True)

    def _emit_end_scan_event(self, scanid: int, metadata: Dict[str, Any]) -> None:
        """Emit an event when a scan ends

        Kwargs:
            scanid: The id of the scan that has ended
            metadata: Any related metadata (not currently used)
        """
        for cb in self._end_scan_watchers:
            try:
                cb(scanid, metadata=metadata)
            except Exception:
                logger.error("Error during scan end callback", exc_info=True)

    def _emit_new_scan_data_0d_event(
        self, scanid: int, channel_name: str, channel_size: int, continue_: bool = False
    ) -> None:
        """Emit an event when there is new 0d scan data

        Kwargs:
            scanid: The id of the scan in progress
            channel_name: The channel that has new data
            channel_size: The length of the new channel
        """
        for cb in self._new_0d_data_watchers:
            try:
                cb(scanid, channel_name=channel_name, channel_size=channel_size)
            except Exception:
                logger.error("Error during 0d data callback", exc_info=True)
                if continue_:
                    continue

    def _emit_new_scan_data_event(
        self,
        scanid: int,
        master_channel: str,
        progress: float,
        channel_name: str,
        channel_size: int,
        channel_progress: float,
    ) -> None:
        """Emit an event when there is new scan data

        Kwargs:
            scanid: The id of the new scan
            master_channel: The master channel of the scan
            progress: A value between 0 and 100 for the scan progress
            channel_name: The channel that has new data
            channel_size: The length of the new channel
            channel_progress: The progres between 0 and 100 for the triggering channel
        """
        for cb in self._new_data_watchers:
            try:
                cb(
                    scanid,
                    master_channel,
                    progress,
                    channel_name,
                    channel_size,
                    channel_progress,
                )
            except Exception:
                logger.error("Error during data callback", exc_info=True)

    def _create_scanid(self, scan_key: str) -> int:
        """Construct a valid DB key from the scan key

        Converts any string based scan id into a hashed number for database compatibility

        Kwargs:
            scan_key: The scan engine scan key
        """
        return mmh3.hash(scan_key) & 0xFFFFFFFF

    def get_conversion(self) -> dict:
        """Get the mca conversion factors to convert bins to energy

        energy = zero + bin * scale
        """
        if "mca" not in self._config:
            return {}
        return {
            "zero": self._config["mca"]["conversion"]["zero"],
            "scale": self._config["mca"]["conversion"]["scale"],
        }

close(self)

Clean up the service at the end.

After this call, the component should not be accessed anymore

Source code in daiquiri/core/hardware/abstract/scansource.py

def close(self):
    """Clean up the service at the end.

    After this call, the component should not be accessed anymore
    """
    pass

get_conversion(self)

Get the mca conversion factors to convert bins to energy

energy = zero + bin * scale

Source code in daiquiri/core/hardware/abstract/scansource.py

def get_conversion(self) -> dict:
    """Get the mca conversion factors to convert bins to energy

    energy = zero + bin * scale
    """
    if "mca" not in self._config:
        return {}
    return {
        "zero": self._config["mca"]["conversion"]["zero"],
        "scale": self._config["mca"]["conversion"]["scale"],
    }

get_scan_data(self, scanid, per_page=25, page=1)

Return scan data for the specific scan

Parameters:

Name	Type	Description	Default
scanid	int	The scan id	required
per_page	int	Number of items to return per page	25
page	int	Page of data to return	1

Returns:

Type	Description
data (dict)	A valid ScanDataSchema dict

Source code in daiquiri/core/hardware/abstract/scansource.py

@abstractmethod
def get_scan_data(self, scanid, per_page=25, page=1):
    """Return scan data for the specific scan

    Args:
        scanid (int): The scan id
        per_page (int): Number of items to return per page
        page (int): Page of data to return

    Returns:
        data (dict): A valid `ScanDataSchema` dict

    """
    pass

get_scan_image(self, scanid, node_name, image_no)

Return a scan image for the specific scan

Parameters:

Name	Type	Description	Default
scanid	int	The scan id	required
node_name	str	The node for the requested image	required
image_no	int	The image number to return	required

Returns:

Type	Description
data (dict)	A valid ScanDataSchema dict

Source code in daiquiri/core/hardware/abstract/scansource.py

@abstractmethod
def get_scan_image(self, scanid, node_name, image_no):
    """Return a scan image for the specific scan

    Args:
        scanid (int): The scan id
        node_name (str): The node for the requested image
        image_no (int): The image number to return

    Returns:
        data (dict): A valid `ScanDataSchema` dict

    """
    pass

get_scan_spectra(self, scanid, point=0, allpoints=False)

Return a scan specific for the specific scan

Parameters:

Name	Type	Description	Default
scanid	int	The scan id	required
point	int	The image number to return	0

Returns:

Type	Description
data (dict)	A valid ScanSpectraSchema dict

Source code in daiquiri/core/hardware/abstract/scansource.py

@abstractmethod
def get_scan_spectra(self, scanid, point=0, allpoints=False):
    """Return a scan specific for the specific scan

    Args:
        scanid (int): The scan id
        point (int): The image number to return

    Returns:
        data (dict): A valid `ScanSpectraSchema` dict

    """
    pass

get_scans(self, scanid=None)

Get a list of scans

Returns a list of scan dicts, can be filtered to only scanid to return details of a specific scan

Parameters:

Name	Type	Description	Default
scanid	int	Scan id to return	None

Returns:

Type	Description
scans (list)	A list of valid ScanSchema dicts

Source code in daiquiri/core/hardware/abstract/scansource.py

@abstractmethod
def get_scans(self, scanid=None):
    """Get a list of scans

    Returns a list of scan dicts, can be filtered to only scanid to return
    details of a specific scan

    Args:
        scanid (int): Scan id to return

    Returns:
        scans (list): A list of valid `ScanSchema` dicts
    """
    pass

watch_end_scan(self, fn)

Register a callback for a scan ends

Source code in daiquiri/core/hardware/abstract/scansource.py

def watch_end_scan(self, fn):
    """Register a callback for a scan ends"""
    if not callable(fn):
        raise AttributeError("Callback function must be callable")

    if not (fn in self._end_scan_watchers):
        self._end_scan_watchers.append(fn)
    else:
        logger.warning(
            "Function {f} is already subscribed to end scan".format(f=fn)
        )

watch_new_0d_data(self, fn)

Register a callback for when there is new 0d data for a scan

Source code in daiquiri/core/hardware/abstract/scansource.py

def watch_new_0d_data(self, fn):
    """Register a callback for when there is new 0d data for a scan"""
    if not callable(fn):
        raise AttributeError("Callback function must be callable")

    if not (fn in self._new_0d_data_watchers):
        self._new_0d_data_watchers.append(fn)
    else:
        logger.warning(
            "Function {f} is already subscribed to new 0d data".format(f=fn)
        )

watch_new_data(self, fn)

Register a callback for when there is new data for a scan

Source code in daiquiri/core/hardware/abstract/scansource.py

def watch_new_data(self, fn):
    """Register a callback for when there is new data for a scan"""
    if not callable(fn):
        raise AttributeError("Callback function must be callable")

    if not (fn in self._new_data_watchers):
        self._new_data_watchers.append(fn)
    else:
        logger.warning(
            "Function {f} is already subscribed to new data".format(f=fn)
        )

watch_new_scan(self, fn)

Register a callback for when a new scan is started

Source code in daiquiri/core/hardware/abstract/scansource.py

def watch_new_scan(self, fn):
    """Register a callback for when a new scan is started"""
    if not callable(fn):
        raise AttributeError("Callback function must be callable")

    if not (fn in self._new_scan_watchers):
        self._new_scan_watchers.append(fn)
    else:
        logger.warning(
            "Function {f} is already subscribed to new scan".format(f=fn)
        )

service

ServiceStates

See http://supervisord.org/subprocess.html#process-states