alab_management.task_view.task module

Define the base class of task, which will be used for defining more tasks.

class BaseTask(samples=None, task_id=None, lab_view=None, priority=TaskPriority.NORMAL, _offline_mode=True, *args, **kwargs)

Bases: ABC

The abstract class of task.

All the tasks should inherit from this class.

add_to(samples)

Used to add basetask to a SampleBuilder’s tasklist during Experiment construction.

Args: samples (Union[SampleBuilder, List[SampleBuilder]]): One or more SampleBuilder’s which will have this task appended to their tasklists.

cleanup_on_cancel: bool = True

Whether the task actor should run the default cleanup (prompt the operator, then move every sample of this task out of the lab with position=None) after the task is cancelled. Set to False in a subclass that reconciles its own samples in on_cancel(), e.g. a transport task whose samples are still sitting at a known physical position.

format_error_report(*, exc=None, stage=None, header=None, include_traceback=True)

Build a rich, debuggable error report for this task.

Use this inside an except block to produce a message that identifies what failed, the exact source location (file, line and function), the current process stage, the task and its samples, plus the full traceback. The result is suitable for an operator prompt (request_user_input), a dashboard message, or a Slack/email alert.

Parameters:

• exc (Optional[BaseException]) – The exception to report. If None, the exception currently being handled is used.

• stage (Optional[str]) – The process stage where it failed. Defaults to the task’s last status message.

• header (Optional[str]) – A short title for the report.

• include_traceback (bool) – Whether to append the full traceback.

Return type:

str

classmethod from_kwargs(samples, task_id, **subclass_kwargs)

Used to create a new task object from the provided arguments.

This is used in the add_to and ExperimentBuilder.add_task method to create a new task object and validate it before adding it to an experiment or sample builder.

Return type:

BaseTask

get_message()

Gets the task message to be displayed on the dashboard.

property is_offline: bool

Returns True if this task is in offline, False if it is a live task.

on_cancel()

Bring the lab to a safe, truthful state after this task has been cancelled.

The task actor calls this once, from its Abort handler, before any cleanup. Use it to stop hardware, park it somewhere safe, and reconcile sample positions to where things physically are. Two things are worth knowing:

• run has already been interrupted, so anything it was waiting on is gone. The resources this task held are still held at this point and are released afterwards.

• Raising from here does not un-cancel the task. The exception is logged and the cancellation proceeds, so a failure to clean up never leaves a task stuck.

Set cleanup_on_cancel to False alongside this method when the default cleanup (which moves every sample of the task out of the lab) would lie about where samples are.

By default this does nothing.

property priority: int

Returns the priority of this task.

property result_specification: type[BaseModel] | None

Returns a pydantic model describing the results to be generated by this task. If specified, this model will be used by task_actor to validate the results after the task is completed. If any error occurs, a warning will be printed. If there is a LargeResult in the result, it will ensured to be stored in the database.

Raises:

NotImplementedError – The subclass must implement this method.:

Returns:

BaseModel

Return type:

A Pydantic model type describing the results to be generated by this task.

abstract run()

Run the task. In this function, you can request lab resources from lab manager and log data to database with logger.

request_resources will not return until all the requested resources are available. So the task will pend until it gets the requested resources, which prevent the conflict in the resource allocation.

When a device get the requested device and sample positions, it also takes over the ownership of these resources, i.e., other task cannot use the device or request the sample positions this task has requested.

We use a context manager to manage the ownership of the resources. when a task is completed, all the devices and sample positions will be released automatically.

Here is an example about how to define the task

# request devices and sample positions from lab manager. The `$` represents
# the name of assigned devices in the sample positions we try to request,
# 4 is the number of sample positions.
with self.lab_view.request_resources({Furnace: [("$.inside", 4)]}) as devices_and_positions:
    devices, sample_positions = devices_and_positions
    furnace = devices[Furnace]
    inside_furnace = sample_positions[Furnace]["$.inside"]

    for sample in self.samples:
        # in a task, we can call other tasks, which will share the same
        # task id, requested devices and sample positions.
        moving_task = Moving(sample=sample,
                             task_id=self.task_id,
                             dest=inside_furnace[0],
                             lab_view=self.lab_view,
                             logger=self.logger)
        moving_task.run()

    # send command to device
    furnace.run_program(self.setpoints)

    while furnace.is_running():
        # log the device data, which is current temperature of the furnace
        self.logger.log_device_signal({
            "device": furnace.name,
            "temperature": furnace.get_temperature(),
        })

run_subtask(task, samples=None, **kwargs)

Run a subtask of this current task. Returns the result, if any, of the subtask.

property samples: list[str]

Returns the list of samples associated with this task.

set_message(message)

Sets the task message to be displayed on the dashboard.

validate()

Validate the task.

This function will be called before the task is executed. Should return False if the task has values that make it impossible to execute. For example, a Heating subclass of BaseTask might return False if the set temperature is too high for the furnace.

By default, this function returns True unless it is overridden by a subclass.

Return type:

bool

class LargeResult(**data)

Bases: BaseModel

A Pydantic model for a large result (file >16 MB). Stored in either gridFS or other filesystems (Cloud AWS S3, etc.).

classmethod check_file_like_data(values)

Check if file_like_data has a .read() method.

check_if_stored()

Check if the large result is stored in the storage system.

file_like_data: Optional[Any]

classmethod from_file_like_data(file_like_data, storage_type=<object object>)

Create a LargeResult object from a file-like object. File-like object must have a .read() method. If file is failed to be stored, will raise a ValueError.

Parameters:

• file_like_data (Any) – the file-like data

• storage_type (str) – the storage type, default to the default storage type in the config

Returns:

LargeResult: the LargeResult object

classmethod from_local_file(local_path, storage_type=<object object>)

Create a LargeResult object from a local file and store it. If file is failed to be stored, will raise a ValueError.

Parameters:

• local_path (str | Path) – the path to the local file

• storage_type (str) – the storage type, default to the default storage type in the config

Returns:

LargeResult: the LargeResult object

identifier: str | ObjectId | None

local_path: str | Path | None

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'file_like_data': FieldInfo(annotation=Union[Any, NoneType], required=False, default=None), 'identifier': FieldInfo(annotation=Union[str, ObjectId, NoneType], required=False, default=None), 'local_path': FieldInfo(annotation=Union[str, Path, NoneType], required=False, default=None), 'storage_type': FieldInfo(annotation=str, required=False, default_factory=<lambda>)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

retrieve()

Retrieve the large result from the storage system.

storage_type: str

store()

Store the large result in the storage system. This method should block until the result is confirmed to be stored. This method should have a timeout regardless of the storage system to not block indefinitely.

exception TaskCancelledError

Bases: Exception

Raise from run to end a task as CANCELLED rather than as ERROR.

A cancellation reaches a task as a dramatiq_abort.Abort, which cannot interrupt a thread already blocked inside a device call. So a task that polls LabView.is_cancelling(), or that asks a device to stop and gets told it stopped, has to end itself – and if it does that by raising anything else it is recorded as a failure. It was not a failure: somebody asked for it. Raise this instead, and the task actor treats it exactly like an abort, including calling BaseTask.on_cancel().

add_reroute_task(supported_sample_positions, task, **kwargs)

Register a reroute task.

add_task(task)

Register a task.

get_all_tasks()

Get all the tasks in the registry.

Return type:

dict[str, type[BaseTask]]

get_task_by_name(name)

Get a task by name.

Return type:

type[BaseTask]