Skip to content

Actors

Actors allow daiquiri to decouple the server from things that actually happen. They are a base python class that define a method. Actors can be executed immediately, or placed into a queue. Components will define which actors they need implementing.

Actors are dynamically reloaded before each invocation meaning the server need not be restarted if the actor method is changed

# Actor named 'example'
# implementors/exampleactor.py
from daiquiri.core.components import ComponentActor, ComponentActorSchema


class ExampleSchema(ComponentActorSchema):
    ...


class ExampleActor(ComponentActor):
    schema = ExampleSchema
    name = "example"
    metatype = "experiment"

    def method(self, **kwargs):
        print(self["key1"])  # get actor arguments
        self["key2"] = 10  # set actor arguments (the initial ones cannot be overwritten)

Schema

Actors can define schemas, these are actually just marshmallow schemas. A schema is used to validate whatever data is passed to the actor before execution, marshmallow provides a simple way to define fields, type, and validation. See the marshmallow documentation for a full list of field types. These schems can be automatically transformed by the client into a form that can be filled and submitted.

A simple actor schema may look like:

from marshmallow import Schema, fields, validate
from daiquiri.core.schema.validators import OneOf

class SimpleScanSchema(ComponentActorSchema):
    motor = OneOf(["robz", "roby"], required=True, metadata={"title": "Motor"})
    motor_start = fields.Float(required=True, metadata={"title": "Start Position"})
    motor_end = fields.Float(required=True, metadata={"title": "End Position"})
    npoints = fields.Int(required=True, metadata={"title": "No. Points"})
    time = fields.Float(
        validate=validate.Range(min=0.1, max=5),
        required=True,
        metadata={"title": "Time per Point", "unit": "s"},
    )
    detectors = fields.List(
        OneOf(["diode", "simu1", "lima_simulator"]),
        required=True,
        metadata={"title": "Detectors", "uniqueItems": True, "minItems": 1},
    )

Title will be used on the frontend form, along with the unit if specified. A series of additional field types and validators can be found in daiquiri.core.schema.validators

Asynchronous Validation

The schema for an actor can be validated in real time (asynchronously) as the data is filled in. This requires that a method with @validates_schema is defined in the schema. Raising a ValidationError from this method will throw an error in the UI. This will stop the form being submitted. The form data is passed into the data argument as a dictionary. The currently selected objects in the UI will be passed into data.objects, multiple objects may be selected. This allows calculated to be made based on the object information, i.e. roi size.

A class attribute DEBUG can be set to true, to log server side information during the validation.

For example:

from marshmallow import fields, validate, validates_schema, ValidationError

class SimpleScanSchema(ComponentActorSchema):
    DEBUG = False
    """If true, the server will log debug information during form interaction"""

    ...

    @validates_schema
    def schema_validate(self, data, **kwargs):
        intervals = [["step_size_x", "x", "x2"], ["step_size_y", "y", "y2"]]

        for keys in intervals:
            if data.get("objects"):
                for obj in data.get("objects"):
                    steps = self._steps(obj[keys[1]], obj[keys[2]], data[keys[0]])
                    if not steps == steps.to_integral_value():
                        raise ValidationError(
                            f"{keys[0]} must be an integer value: {steps}"
                        )

Warnings

The actor schema can also send warnings back to the UI. These will not stop the form being submitted, but will draw attention. To do this define a function called warnings in the actor schema. The form data is passed into the data argument as a dictionary. Warnings should be passed back with a key corresponding to the field throwing the warning

class SimpleScanSchema(ComponentActorSchema):
    ...

    def warnings(self, data, **kwargs):
        warnings = {}
        if data.get("objects"):
            for obj in data.get("objects"):
                size_x = (obj["x2"] - obj["x"]) * 1e-9 / 1e-6
                size_y = (obj["y2"] - obj["y"]) * 1e-9 / 1e-6

                if size_x > 100 or size_y > 100:
                    warnings[
                        obj["subsampleid"]
                    ] = f"Object {obj['subsampleid']} will use stepper rather than piezo as size is {size_x:.0f}x{size_y:.0f} um"

        return warnings

Calculated Parameters

An actor schema can also generated calculated parameters from data provided to the actor. For example the number of steps can be calculated from the size of the object and the step size. Pass back a dictionary with key value pairs corresponding to the field key defined in the schema.

class SimpleScanSchema(ComponentActorSchema):
    ...

    def calculated(self, data, **kwargs):
        calculated = {}

        intervals = {
            "steps_x": ["step_size_x", "x", "x2"],
            "steps_y": ["step_size_y", "y", "y2"],
        }

        for iv, keys in intervals.items():
            if data.get(keys[0]):
                steps = []
                if data.get("objects"):
                    for obj in data["objects"]:
                        step = self._steps(obj[keys[1]], obj[keys[2]], data[keys[0]])
                        # print('calculated', key, step, step.to_integral_value() == step, obj[keys[2]] - obj[keys[1]])
                        step = (
                            step.to_integral_value()
                            if step.to_integral_value() == step
                            else round(step, 2)
                        )

                        steps.append(step)

                calculated[iv] = ", ".join(map(str, steps))

        return calculated

Time estimate

Actor schemas can return a time estimate, this will be used if the actor is queued to give an estimate of the total queue time. To do this define a method called time_estimate, as per the other schema functions the form data will be passed into data as a dictionary. The time should be returned as an int/float from the function

class SimpleScanSchema(ComponentActorSchema):
    ...

    def time_estimate(self, data):
        fudge = 1.5
        print("time_estimate", data)
        if data.get("step_size_x") and data.get("step_size_y"):
            if data.get("objects"):
                for obj in data["objects"]:
                    steps_x = (obj["x2"] - obj["x"]) * 1e-9 / data["step_size_x"] / 1e-6
                    steps_y = (obj["y2"] - obj["y"]) * 1e-9 / data["step_size_y"] / 1e-6
                    return data["dwell"] * steps_x * steps_y * fudge

Presets

Actor schemas can provide methods to load and save presets. The simplest way to do this is to provide a presets key in the schema Meta, this should contain a dictionary of key, value pairs to be applied to the schema.

class SimpleScanSchema(ComponentActorSchema):
    ...

    class Meta:
        presets = {
            "Preset1": {
                key: value
                ...
            }
        }

Schemas can also generate the list of presets at run time rather than initialisation as well using the get_presets function:

class SimpleScanSchema(ComponentActorSchema):
    ...

    def get_presets(self):
        ...

        return {
            "Preset1": { 
                key: value,
                ...
            }
            ...
        }
The get_presets function will always take presedence over the instantiated presets property.

A schema can finally save the current data as a new preset. When called from the UI the save_preset function will be invoked on the schema with the preset name and associated data:

class SimpleScanSchema(ComponentActorSchema):
    ...

    def save_preset(self, name, data):
        ...

It is up to the schema to determine how this data is actually persisted.

Beamline Parameters

A special type of schema ParamSchema can be used to execute arbritary actions before and after an actor, for example, move a detector to a specific position, or open and close a fast shutter.

A simple example of a ParamSchema is shown below:

# beamlineparams.py
from daiquiri.core.components.params import ParamHandler, ParamsHandler, ParamSchema

class FastShutterParam(ParamHandler):
    def before(self, value):
        if value:
            shut = cfg.get("safshut")
            shut.open()

    def after(self, value):
        if value:
            shut = cfg.get("safshut")
            shut.close()


class DetectorDistanceParam(ParamHandler):
    def before(self, value):
        det = cfg.get("omega")
        det.move(value)


class BeamlineParamsHandler(ParamsHandler):
    fast_shutter = FastShutterParam()
    detector_distance = DetectorDistanceParam()


class BeamlineParamsSchema(ParamSchema):
    handler = BeamlineParamsHandler()

    detector_distance = fields.Float(
        metadata={"title": "Detector Distance"},
        validate=validate.Range(min=20, max=400),
        metadata={"description": "Move the detector to specified distance", "unit": "mm"},
    )

    fast_shutter = fields.Bool(
        metadata={"title": "Fast Shutter", "description": "Open fast shutter before and close after"}
    )

This can be kept in a separate file and then imported into each actor. This way the ParamSchema can be reused between different types of actors:

from .beamlineparams import BeamlineParamsSchema

class MyactorSchema(ComponentActorSchema):
    ...
    beamlineparams = fields.Nested(BeamlineParamsSchema, metadata={"title": "Beamline Parameters"})

Controls Session

If a controls session is defined in app.yml then the session can be accessed directly from within the actor. For example if the controls_session_type is bliss:

from daiquiri.core.hardware.bliss.session import *

global_from_bliss_session

Metadata and experiment types

Actors launched from the imageviewer and samplescan components will by default create an entry in daiquiri's database in order to allow online and downstream processing of said data. The metatype key tells daiquiri which DataCollectionGroup.experimentType to populate in the database. This list is fairly extensive and can be used to describe a variety of experiment types. The default if left unspecified is the generic experiment type, this can be easily changed, for example to an XRF map:

class ExampleActor(ComponentActor):
    metatype = "XRF map"

Additional metadata can be used to populate the relevant fields of the database which can aid with downstream dataprocessing, for example the beamline energy, or step size. These can be populated by calling update_datacollection from within the actor method, for example for the roiscan actor:

kwargs["update_datacollection"](
    self,
    emit_start=True,
    datacollectionnumber=mmh3.hash(mesh._scan_data.key) & 0xFFFFFFFF,
    imagecontainersubpath="1.1/measurement",
    dx_mm=kwargs["step_size_x"] * 1e-3,
    dy_mm=kwargs["step_size_y"] * 1e-3,
    numberofimages=steps_x * steps_y,
    exposuretime=kwargs["dwell"],
    wavelength=to_wavelength(energy),
    steps_x=steps_x,
    steps_y=steps_y,
    orientation="horizontal",
)

This function is injected from both the imageviewer and samplescan components.

It is sometimes useful to be able to disable this behaviour so that an executed actor is not recorded into the database, for example maybe an alignment scan that is run frequently. This can be achieved by setting the metatype to None:

class ExampleActor(ComponentActor):
    metatype = None

Downstream handling

In addition to the experimentType arbritary metadata can also be passed through the actor using the additional_metadata keyword, depending on the saving type handling this metadata can be ingested into other handlers, for example ICAT:

class ExampleActor(ComponentActor):
    metatype = "XRF map"
    additional_metadata = {"definition": "XRF_2Dmap"}

In this case, used along side saving configured as so:

saving_metadata:
    dataset_definition: "{metadata_definition}"

XRF_2Dmap will be set as the definition in ICAT. See Configuration for more details.