Skip to content

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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
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._locked_callbacks = []
        self._locked: str | None = None

        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(f"Unknown property '{prop}' from object {self._name}")

    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", "locked"]
        }
        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 set_locked(self, reason: str | None):
        """Set the device locked for a reason.

        Argument:
            reason: Locking reason. If none the device is not locked.
        """
        if self._locked == reason:
            return
        self._locked = reason
        for cb in self._locked_callbacks:
            cb(self, self._locked)

    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 fn not in self._online_callbacks:
            self._online_callbacks.append(fn)

    def subscribe_locked(self, fn):
        """Subscribe to the locked state of the hardware object

        Add a function to a list of callbacks for when the locked 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 fn not in self._locked_callbacks:
            self._locked_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 locked(self) -> str | None:
        return self._locked

    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(f"Unknown property '{prop}' from object {self._name}")

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_locked(self, reason)

Set the device locked for a reason.

Argument

reason: Locking reason. If none the device is not locked.

Source code in daiquiri/core/hardware/abstract/__init__.py
def set_locked(self, reason: str | None):
    """Set the device locked for a reason.

    Argument:
        reason: Locking reason. If none the device is not locked.
    """
    if self._locked == reason:
        return
    self._locked = reason
    for cb in self._locked_callbacks:
        cb(self, self._locked)

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", "locked"]
    }
    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_locked(self, fn)

Subscribe to the locked state of the hardware object

Add a function to a list of callbacks for when the locked 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_locked(self, fn):
    """Subscribe to the locked state of the hardware object

    Add a function to a list of callbacks for when the locked 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 fn not in self._locked_callbacks:
        self._locked_callbacks.append(fn)

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 fn not 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

    def create_monitor(self):
        """Create a global monitor at global state of the handler"""
        return None

create_monitor(self)

Create a global monitor at global state of the handler

Source code in daiquiri/core/hardware/abstract/__init__.py
def create_monitor(self):
    """Create a global monitor at global state of the handler"""
    return None

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