abstract

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

    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
        info["require_staff"] = self.require_staff()
        info["alias"] = self.alias()

        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 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 _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))
        value = prop.translate_from(value)
        if name in self._properties:
            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)

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
    info["require_staff"] = self.require_staff()
    info["alias"] = self.alias()

    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 _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:
            value = 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)
                value = hprop.translate_from(hvalue)
                return value
            except NotImplementedError:
                logger.info("Object %s does not implement %s", self._id, prop)
        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 (only if not grayscale)
        if frame.dtype != np.uint8 and len(frame.shape) > 2:
            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, session: str, config: Dict[str, Any], app=None):
        self._app = app
        self._config = config
        self._session_name = 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