Writing a custom scan¶

This section presents the fundamental concepts and objects involved in a scan procedure. It will describe how to write a custom scan through examples.

The acquisition chain¶

In Bliss, a scanning procedure is managed by the Scan object (bliss.scanning.scan).

The Scan object works on top of an AcquisitionChain object containing AcquisitionObject objects (bliss.scanning.chain).

Screenshot

The acquisition chain is a tree of acquisition objects organized in a masters and slaves hierarchy.

Screenshot

There are two kind of objects built on top of the AcquisitionObject base class: AcquisitionMaster and AcquisitionSlave (bliss.scanning.chain).

The AcquisitionMaster class is able to trigger the acquisition slaves below itself.
The AcquisitionSlave class is always at the end of a branch of the acquisition chain.

The role of the AcquisitionObject is to encapsulate a CounterController in order to use it in the context of a scan.

The AcquisitionObject defines:

how to behave while receiving incoming triggers (software and/or hardware)
how to acquire data
how to publish data

The underlying CounterController is the one who knows how to read the data from the hardware device.

The acquisition chain can be conceptually split in two regions.

Screenshot

On the left, the static part containing the top level masters. This part must be entirely described by the author of the scan procedure.

On the right, the dynamic part which depends on the list of counters given to this scan procedure. The construction of this part is partially automated thanks to the ChainBuilder object. From the given list of counters, the ChainBuilder object automatically finds all the counter controllers associated to given counters.

All counter controllers are able to return the special AcquisitionObject associated with themselves (see CounterController.get_acquisition_object()).

Also, if a counter controller has a master controller on top of it, the chain builder will find it and register the links (like LimaMaster on top of LimaRoi and LimaBPM in the figure above).

Simple scan example¶

Example of a step by step scan with one axis

To keep the example short and simple, in this scan we will only handle counters form Lima controllers.

If other counters from other controllers are passed to this scan, they are simply ignored.

def scan_demo( motor, start, stop, npoints, count_time, *counters ):
    chain = AcquisitionChain()

    #=== The left side of the chain (STATIC MASTERS) =========================
    # MotorMaster
    acq_master = LinearStepTriggerMaster(npoints, motor, start, stop)

    #=== The right side of the chain (FROM COUNTERS LIST INTROSPECTION) ======
    builder = ChainBuilder(counters)

    #--- handles counters from Lima controllers (Image, Rois, BPM)
    lima_params = {
        "acq_nb_frames": npoints,
        "acq_expo_time": count_time * 0.9,
        "acq_trigger_mode": "INTERNAL_TRIGGER_MULTI",
        "prepare_once": True,
        "start_once": False,
    }

    for node in builder.get_nodes_by_controller_type(Lima):
        # setting the parameters of the LimaMaster is enough
        # the children slaves under the LimaMaster will try to find
        # their parameters from the LimaMaster parameters
        node.set_parameters(acq_params=lima_params)

        # adding the LimaMaster to chain is enough
        # the children slaves (ROI, BPM) are automatically placed below
        # their master
        chain.add(acq_master, node)

    #----- print some information about the chain construction --------
    #print the result of the introspection
    builder.print_tree(not_ready_only=False)
    #print the chain that has been built
    print(chain._tree)

    #----- finalize the scan construction ----------------------------
    scan_info = {
        "npoints": npoints,
        "count_time": count_time,
        "start": start,
        "stop": stop,
        "type": "continous_scan_demo",
    }

    sc = Scan(
        chain,
        name="scan_demo",
        scan_info=scan_info,
        #save=False,
        #save_images=False,
        #scan_saving=None,
        #data_watch_callback=StepScanDataWatch(),
    )

    #----- start the scan ----------------------------
    sc.run()

Define top masters¶

In the first part of this example, the chain object and the motor top master are instantiated.

Instantiate the chain and declare a top master

def scan_demo( motor, start, stop, npoints, count_time, *counters ):

    #----- Initialize the chain object -------------------------------
    chain = AcquisitionChain()

    #----- Initialize the top master
    acq_master = LinearStepTriggerMaster(npoints, motor, start, stop)

The motor top master (LinearStepTriggerMaster) will perform npoints steps from start position to stop position.

As a top master, it will be the first in the chain (root) and it will be the one who triggers all other acquisition objects that are placed under him.

The top master is not put in the chain yet. It will be done later on when adding other acquisition objects in the chain under this one.

Introspect counters list¶

In the second part of this example, the chain builder object is instantiated with the list of counters given to that scan:

Instantiate the chain builder and introspect counters list

builder = ChainBuilder(counters)

Notice that, at that time, counters may contains different objects of different types such as Counter, CounterController, MeasurementGroup.

During its initialization, the chain builder will:

obtain a flatten list of Counters by retrieving counters from CounterControllers and MeasurementGroups.
remove duplicated counters and sort the counters by name.
sort the counters by dependency level (for example CalcCounters are stored at the end of the counter list because they depend on others real counters).
introspect the counter list in order to find the CounterControllers on top of the different counters and create one ChainNode object per CounterController.
check if the CounterControllers have a master_controller on top of them. If true a ChainNode object is created for the master_controller and the node registers the CounterControllers that are attached to itself (see node.children). Notice that a master_controller is an instance of a CounterController.

After this, the builder has created a dictionary of ChainNodes (one per CounterController) (see builder.nodes). The role of the ChainNode object is to store the information required to instantiate the AcquisitionObject associated to each CounterController.

Set acquisition parameters¶

Once the builder is initialized, the acquisition parameters required for each AcquisitionObject associated to each CounterController must be defined. To do so, the builder is able to return all nodes that hold a CounterController of a given type. In this example the nodes of Lima controllers are returned (builder.get_nodes_by_controller_type(Lima)). Then the lima acquisition parameters are declared and stored into the node (node.set_parameters(acq_params=lima_params)).

Define acquisition parameters for Lima controllers

# define acquisition parameters for Lima controllers
lima_params = {
    "acq_nb_frames": npoints,
    "acq_expo_time": count_time * 0.9,
    "acq_trigger_mode": "INTERNAL_TRIGGER_MULTI",
    "prepare_once": True,
    "start_once": False,
}
# set acquisition parameters to all Lima controllers
for node in builder.get_nodes_by_controller_type(Lima):
    # setting the parameters of the LimaMaster is enough
    # the children slaves under the LimaMaster will try to find
    # their parameters from the LimaMaster parameters
    node.set_parameters(acq_params=lima_params)

Set children parameters¶

In the case where it is necessary to set the children parameters explicitly, the children of a node can be managed like this:

Handle children nodes

lima_params = {
    "acq_nb_frames": npoints,
    "acq_expo_time": count_time * 0.9,
    "acq_trigger_mode": "INTERNAL_TRIGGER_MULTI",
    "prepare_once": True,
    "start_once": False,
}

lima_children_params = {"count_time": count_time * 0.8, "npoints": npoints }

for node in builder.get_nodes_by_controller_type(Lima):

    node.set_parameters(acq_params=lima_params)

    for child_node in node.children:
        child_node.set_parameters(acq_params=lima_children_params)

Now, the possible counters related to the Lima controller have been handled.

Add to acquisition chain¶

The last thing to do with the Lima node is to define where to put it in the acquisition chain. In this example it is placed under the LinearStepTriggerMaster:

Add Lima controller node to the chain

# adding the LimaMaster to chain is enough
# the children slaves (ROI, BPM) are automatically placed below
# their master
chain.add(acq_master, node)

The LinearStepTriggerMaster will trigger the LimaAcquisitionMaster and the LimaAcquisitionMaster will trigger slaves under himself.

The LimaAcquisitionMaster associated to that node is created only at that time, when putting it into the chain.

The LimaAcquisitionMaster is instantiated using the acquisition parameters that have been stored into the node (seenode.acquisition_parameters).

If this node has children nodes (example LimaRoi and LimaBPM), the AcquisitionObject of each child node will be instantiated at this time and placed in the chain below the AcquisitionObject of this node (i.e. below the LimaAcquisitionMaster).

If the acquisition parameters of the children nodes have not been set, the children nodes will try to find their parameters in the parent parameters.

Handle other controllers¶

To handle other type of controllers/counters, just repeat what has been done above but adapt the acquisition parameters and the type of controller.

Handle other controller’s counters

xxx_params = {
    "abc": ... ,
    "foo": ... }

for node in builder.get_nodes_by_controller_type( xxx ):
    node.set_parameters(acq_params=xxx_params)

Check the acquisition chain¶

In order to check if all the ChainNodes of the builder have been managed properly, use builder.print_tree() to show/print the tree representation of the chain nodes. If the argument not_ready_only is True, it will be printed only if some nodes have not been treated.

In order to check if the chain has been built properly and looks as expected, use print(chain._tree) to show/print the tree representation of the acquisition chain.

Print builder and chain info

#----- print some information about the chain construction --------

# print the result of the introspection (ChainNodes of the builder)
builder.print_tree(not_ready_only=False)

# print the chain that has been built (AcquisitionChain as a tree)
print(chain._tree)

Finalize and run your scan¶

To finalize the writing of the scan procedure, create the Scan object passing the chain and the scan_info. The scan_info is a dictionary containing the information to be exposed to the outer world. Finally, start the scan by calling its run() method.

Instantiate scan object and run

#----- finalize the scan construction ----------------------------
scan_info = {
    "npoints": npoints,
    "count_time": count_time,
    "start": start,
    "stop": stop,
    "type": "continous_scan_demo",
}

sc = Scan(
    chain,
    name="scan_demo",
    scan_info=scan_info,
    #save=False,
    #save_images=False,
    #scan_saving=None,
    #data_watch_callback=StepScanDataWatch(),
)

#----- start the scan ----------------------------
sc.run()

Two top master example¶

In most common cases, the need for two top master in a scan comes from the fact that some counters could not follow the trigger speed of the fast acquisition chain. But still those counters need to be monitored. The monitored part is like a timescan aside the fast acquisition chain.

Screenshot

In the following example, all Lima controllers will be part of the fast acquisition chain. All sampling counters given as argument will be part of the monitoring (slow acquisition chain).

Writing a scanning with two top masters (fast and slow chain branches)

from bliss import setup_globals
from bliss.controller import lima
from bliss.scanning import chain
from bliss.scanning import scan
from bliss.scanning import toolbox
from bliss.scanning.acquisition.motor import MotorMaster
from bliss.scanning.acquisition.musst import MusstAcquisitionMaster
from bliss.scanning.acquisition.timer import SoftwareTimerMaster

def fast_and_monitor(motor, start, stop, npoints, count_time, *counters, monitor_time=0.5):

    builder = toolbox.ChainBuilder(counters)

    #=== Fast chain part: MotorMaster => MusstAcquisitionMaster => Lima  ===

    # creation of the fast chain's top master with the motor passed as argument
    fast_top_master = MotorMaster(motor,start,stop,time=count_time*npoints)

    # creation of musst master which will be trigger by the motor master
    musst_master = MusstAcquisitionMaster(setup_globals.musst,
                                        program="my_fast_scan_prog"
                                        vars={"START_POS":int(start*motor.steps_per_unit),
                                                "STOP_POS":int(stop*motor.steps_per_unit),
                                                "NPOINTS":npoints})
    # Add them to the acquisition chain
    chain.add(fast_top_master,musst_master)

    # Handle Lima controllers
    for lima_node in builder.get_nodes_by_controller_type(lima.Lima):
        lima_node.set_parameters(acq_params={"acq_nb_frames": npoints,
                                            "acq_expo_time": count_time * 0.9,
                                            "acq_trigger_mode": "EXTERNAL_TRIGGER_MULTI",
                                            "prepare_once": True,
                                            "start_once": True})
        # Add lima node under the musst
        chain.add(musst_master,lima_node)



    #=== Slow chain part: SoftwareTimerMaster => any sampling counters  ===

    # First create the software timer
    monitor_timer = SoftwareTimerMaster(count_time=monitor_time)

    # Handle sampling counter controlelrs
    for sampling_node in builder.get_nodes_by_controller_type(SamplingCounterController):
        sampling_node.set_parameters({'count_time':monitor_time,
                                    'npoints':0})

        #Add sampling node under the software timer
        chain.add(monitor_time,sampling_node)


    # Initialize scan object and run
    s = Scan(chain,
            name='fast_and_monitor',
            scan_info={'npoints':npoints,
                        'count_time':count_time,
                        'start':start,
                        'stop':stop})
    s.run()
    return s