alab_management.lab_view module

alab_management.lab_view module#

Each task process will have a LabView instance, through which it can request the lab resources (devices and sample positions).

It can also update the position of a sample in the lab.

exception DeviceRunningException[source]#

Bases: Exception

Raise when a task try to release a device that is still running.

class LabView(task_id)[source]#

Bases: object

LabView is a wrapper over device view and sample view. A task can get access to that to request resources, query sample and update sample positions.

property cancellation_event: Event#

A threading.Event that becomes set once this task is being cancelled.

Backed by a daemon thread polling the database once a second, started on first access, so a task that never asks for it pays nothing. Use it where an Event is more convenient than polling is_cancelling() – most usefully as a cancellable sleep, which wakes early instead of sitting out the full delay:

# returns True as soon as the task is cancelled, False if the full 30s elapsed
if self.lab_view.cancellation_event.wait(timeout=30):
    raise TaskCancelledError("cancelled while waiting")

get_locked_sample_positions()[source]#

Get a list of sample positions that are occupied by this task.

Return type:: list[str]

get_sample(sample)[source]#

Get a sample by either an ObjectId corresponding to sample_id, or as a string corresponding to the sample’s name within the experiment., see also get_sample.

Return type:: Sample

get_sample_position_parent_device(position)[source]#

Get the name of the device that owns the sample position.

Return type:: str | None

is_cancelling()[source]#

Whether someone has asked for this task to be cancelled.

A cancellation reaches a task as a dramatiq_abort.Abort raised in the worker thread, and that cannot interrupt a thread already blocked inside a device call or a time.sleep. Poll this in long waits, and bail out cooperatively, so a cancel takes effect at the next safe point instead of after the blocking call finally returns.

while furnace.is_running():
    if self.lab_view.is_cancelling():
        raise TaskCancelledError("cancelled while waiting for the furnace")
    time.sleep(5)

Return type:: bool

lock_exact_sample_positions(positions)[source]#

Lock exact sample positions as a context manager.

This method locks exact position names (not prefixes) and automatically releases them when exiting the context. This is useful when you need to lock specific positions like “input_rack/slot/1” without prefix matching potentially selecting “input_rack/slot/10”.

Parameters:: positions (list[str]) – List of exact sample position names to lock
Yields:: list[str] (The list of locked positions (same as input))
Raises:: ValueError – If any position is currently occupied or locked by another task.:

Example

# Lock exact positions
with self.lab_view.lock_exact_sample_positions([
    "input_rack/slot/1",
    "input_rack/slot/2"
]) as locked_positions:
    # Use the positions
    self.lab_view.move_sample(
        sample=self.sample,
        position=locked_positions[0]
    )
# Positions are automatically released here

lock_sample_position(position)[source]#

Lock an exact sample position for this task.

Unlike request_resources, this method locks an exact position name rather than using prefix matching. This is useful when you need to lock a specific position like “input_rack/slot/1” without it potentially matching “input_rack/slot/10”.

Parameters:: position (str) – The exact name of the sample position to lock (e.g., “input_rack/slot/1”)
Raises:: ValueError – If the position is currently occupied or locked by another task.:

Example

# Lock an exact position
self.lab_view.lock_sample_position("input_rack/slot/1")

# Use the position
self.lab_view.move_sample(sample=self.sample, position="input_rack/slot/1")

# Release when done
self.lab_view.release_sample_position("input_rack/slot/1")

move_sample(sample, position)[source]#: Move a sample to a new position. sample can be given as either an ObjectId corresponding to sample_id, or as a string corresponding to the sample’s name within the experiment.

See also

move_sample

property priority: int#: Get the priority of the task.

release_all_resources()[source]#

Give back every resource this task holds or has requested.

request_cleanup already does this as its last step. Call it directly when a task has reconciled its own samples and only needs its devices and sample positions released, which is what a task with cleanup_on_cancel = False needs after being cancelled.

release_sample_position(position)[source]#

Release a locked sample position.

Parameters:: position (str) – The exact name of the sample position to release
Raises:: ValueError – If the position is invalid:

Example

self.lab_view.lock_sample_position("input_rack/slot/1")
try:
    # Use the position
    self.lab_view.move_sample(sample=self.sample, position="input_rack/slot/1")
finally:
    self.lab_view.release_sample_position("input_rack/slot/1")

request_cleanup(error_message=None)[source]#

Request cleanup of the task. This function will block until the task is cleaned up.

Parameters:: error_message (Optional[str]) – A pre-formatted, debuggable error report to show the operator (and send to Slack/email). If not provided, a report is built from the exception currently being handled so the prompt always identifies what failed, where (file/line/function), and the full traceback.

request_resources(resource_request, priority=None, timeout=None, exact_positions=None)[source]#

Request devices and sample positions. This function is a context manager, which should be used in a with statement to ensure all the devices are released when the task is done.

resource_request format is: {device: {position: number, …}, …} device can be a name of a specific device (str), a type of device, or None. If device is a type, the resource request will look for any available device of that type. If device is None, the resource request will look for sample positions that do not belong to a device. position is the name of a sample position that should be reserved on the device, and number is the number of such positions that should be reserved. If the device is required but no positions are required, this can be left as an empty dictionary.

Examples ——– {TubeFurnace: {“tray”: 4}, “arm1”: {}} will find the first available TubeFurnace device, then reserve 4 sample positions of “{tubefurnacename}/tray/{tray_index}” on that device. It will also find the device named “arm1”.

By default, position names are matched by prefix (e.g., “input_rack/slot/1” would match both “input_rack/slot/1” and “input_rack/slot/10”). To match exact positions, use the exact_positions parameter to specify which position names should be matched exactly.

Parameters:

resource_request (dict[type[BaseDevice] | str | None, dict[str, str | int]]) – Dictionary mapping devices to position requests
priority (Optional[int]) – Optional priority for the request (0-40, default 20). Higher number = higher priority. Numbers >= 100 are reserved for urgent/error correcting requests.
timeout (Optional[float]) – Optional timeout for the request in seconds
exact_positions (Optional[set[str]]) –
Set of position names that should be matched exactly (not by prefix). Position names should be specified as they appear in the resource_request dictionary, before device prefixes are added. For example:
- For {None: {“input_rack/slot/1”: 1}}, use exact_positions={“input_rack/slot/1”}
- For {Furnace: {“slot/1”: 1}}, use exact_positions={“slot/1”} (device prefix added automatically)
Important: Exact matching only works when number=1. If you request multiple positions (e.g., {“slot”: 11}), you cannot use exact matching for that position. Use exact_positions only for positions where number=1.

If None (default), all positions use prefix matching for backward compatibility.

Example

# Prefix matching (default) - "slot/1" could match "slot/1" or "slot/10"
with self.lab_view.request_resources({Furnace: {"slot/1": 1}}) as (devices, positions):
    ...

# Exact matching - only matches "slot/1"
with self.lab_view.request_resources(
    {Furnace: {"slot/1": 1}},
    exact_positions={"slot/1"}
) as (devices, positions):
    ...

# Mix of prefix and exact matching
with self.lab_view.request_resources(
    {None: {"input_rack/slot/1": 1, "input_rack/slot/2": 1}},
    exact_positions={"input_rack/slot/1"}  # Only slot/1 is exact
) as (devices, positions):
    # slot/1 is exact, slot/2 uses prefix matching
    ...

# Invalid: Cannot use exact matching with number > 1
# This will raise ValueError:
# with self.lab_view.request_resources(
#     {"BFT_input_rack": {"slot": 11}},
#     exact_positions={"slot"}  # ERROR: exact matching requires number=1
# ) as (devices, positions):
#     ...

request_user_input(prompt, options)[source]#

Request user input from the user. This function will block until the user inputs something.

Parameters:

prompt (str) – The prompt to display to the user.
options (list[str]) – A list of options to display to the user.

Return type:

str

Returns:

response (str): The value returned by the user (from the buttons).

request_user_input_with_note(prompt, options)[source]#

Request user input from the user. This function will block until the user inputs something.

Parameters:

prompt (str) – The prompt to display to the user.
options (list[str]) – A list of options to display to the user.

Return type:

tuple[str, str]

Returns:

response (str): The value returned by the user (from the buttons). note (str): The note returned by the user.

run_subtask(task, samples, **kwargs)[source]#

Run a task as a subtask within the task. basically fills in task_id and lab_view for you. this command blocks until the subtask is completed.

Parameters:

task (Type[BaseTask]) – The type/class of the Task to run.
samples (List[Union[ObjectId, str]]) – List of sample IDs or names.
**kwargs – will be passed to the Task method via the parameters entry in the task collection.

set_sample_in_transit(sample, source, destination)[source]#

Record that a sample is being physically moved from source to destination.

Call this right before a robot move begins. It does not change the recorded position, so if the move crashes mid-transfer the last known position plus intended destination remain visible. A subsequent successful move_sample() clears the record automatically.

alab_management.lab_view module

Contents

alab_management.lab_view module#