alab_management.sample_view.sample_view module

A wrapper over the samples and sample_positions collections.

class SamplePositionRequest(**data)

Bases: BaseModel

The class is used to request sample position.

You need to specify the prefix of the sample position (will be used to match by startwith method) and the number you request. By default, the number is set to be 1.

classmethod from_py_type(sample_position)

Create a SamplePositionRequest from a string or a dict.

classmethod from_str(sample_position_prefix)

Create a SamplePositionRequest from a string.

Return type:

SamplePositionRequest

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

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

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

model_fields: ClassVar[dict[str, FieldInfo]] = {'number': FieldInfo(annotation=int, required=False, default=1, metadata=[None, Interval(gt=None, ge=0, lt=None, le=None), None]), 'prefix': FieldInfo(annotation=str, required=True)}

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

This replaces Model.__fields__ from Pydantic V1.

number: Annotated[int]

prefix: str

class SamplePositionStatus(value)

Bases: Enum

The status of a sample position.

• EMPTY: the sample position is neither locked nor occupied

• OCCUPIED: there is a sample in the sample position

• LOCKED: the sample position is reserved by a task

EMPTY = 1

LOCKED = 3

OCCUPIED = 2

class SampleView

Bases: object

Sample view manages the samples and their positions.

add_sample_positions_to_db(sample_positions, parent_device_name=None)

Insert sample positions info to db, which includes position name and description.

If one sample position’s name has already appeared in the database, we will just skip it.

Parameters:

• sample_positions (list[SamplePosition]) – some sample position instances

• parent_device_name (Optional[str]) – name of the parent device to these sample_positions.

clean_up_sample_position_collection()

Drop the sample position collection.

clear_sample_in_transit(sample_id)

Clear the in-transit record for a sample (e.g. for manual recovery).

create_sample(name, position=None, sample_id=None, tags=None, metadata=None)

Create a sample and return its uid in the database.

Samples with the same name can exist in the database

Return type:

ObjectId

exists(sample_id)

Check if a sample exists in the database.

Parameters:

sample_id (ObjectId) – id of the sample within sample collection

Return type:

bool

Returns:

bool: True if sample exists in the database

get_all_sample_positions_from_db()

Get all sample positions from the database directly.

Returns a dictionary mapping position names to their database entries. This includes both standalone and device-associated sample positions.

Return type:

dict[str, dict[str, Any]]

get_available_sample_position(task_id, position_prefix, exact_match=False)

Check if the position is occupied.

The structure of returned list is {"name": str, "need_release": bool}. The entry need_release indicates whether a sample position needs to be released when __exit__ method is called in the SamplePositionsLock.

Parameters:

• task_id (ObjectId) – The task ID requesting the position

• position_prefix (str) – The position name or prefix to match

• exact_match (bool) – If True, match the exact position name. If False, use prefix matching.

Return type:

list[dict[str, str | bool]]

get_in_transit_samples()

Return summary info for all samples currently recorded as in transit.

Return type:

list[dict[str, Any]]

get_sample(sample_id)

Get a sample by its id.

Parameters:

sample_id (ObjectId) – id of the sample within sample collection

Raises:

ValueError – no sample found with given id:

Return type:

Sample

Returns:

Sample: Sample object for given id

get_sample_by_position(position)

Get the sample object at a given position.

Parameters:

position (str) – The position to search for

Return type:

Sample | None

Returns:

The Sample object if a sample exists at the position, None otherwise

get_sample_name_by_position(position)

Get the sample name at a given position.

Parameters:

position (str) – The position to search for

Return type:

str | None

Returns:

The sample name if a sample exists at the position, None otherwise

get_sample_position(position)

Get the sample position entry in the database.

if the position is not a valid position (not defined in the database), return None

Return type:

dict[str, Any] | None

get_sample_position_max_number_by_prefix(prefix)

Get the maximum number of sample positions for a given prefix from the database.

Parameters:

prefix (str) – The prefix to search for (e.g., “furnace_temp”)

Return type:

int

Returns:

The maximum number found for positions with this prefix

get_sample_position_parent_device(position)

Get the parent device of a sample position.

If no parent device is defined, returns None. If the “position” query is a prefix, will look for a single parent device across all matched positions (ie a query for position=”furnace_1/tray” will properly return “furnace_1” even if “furnace_1/tray/1” and “furnace_1/tray/2” are in the database _as long as “furnace_1” is the parent device of both!_).

Return type:

str | None

get_sample_position_status(position)

Get the status of a sample position.

If there is a sample in the position, return OCCUPIED; else if the sample position is locked by a task, return LOCKED; return EMPTY

Parameters:

position (str) – the name of the sample position

Return type:

tuple[SamplePositionStatus, ObjectId | None]

Returns:

if the position is occupied by a sample, return OCCUPIED and the task id of the sample if the position is locked by a task, return LOCKED and the task id else, return EMPTY and None

get_sample_positions_by_task(task_id)

Get the list of sample positions that is locked by a task (given task id).

Return type:

list[str]

get_sample_positions_names_by_device(device_name)

Get all the sample positions names that are related to a device.

Return type:

list[str]

get_sample_positions_names_by_prefix(prefix)

Get all the sample positions names that are related to a device.

Return type:

list[str]

get_samples_on_device(device_name)

Get all the samples on a device.

Return type:

dict[str, list[ObjectId]]

is_locked_position(position)

Tell if a sample position is locked or not.

Return type:

bool

is_unoccupied_position(position)

Tell if a sample position is unoccupied or not.

Return type:

bool

lock_sample_position(task_id, position)

Lock a sample position.

move_sample(sample_id, position)

Update the sample with new position.

A successful move also clears any in_transit record, since the sample has arrived at a well-defined position.

release_sample_position(position)

Unlock a sample position.

remove_sample_position_by_prefix(prefix)

Remove a sample position from the database.

request_sample_positions(task_id, sample_positions, exact_positions=None)

Request a list of sample positions, this function will return until all the sample positions are available.

Parameters:

• task_id (ObjectId) – the task id that requests these resources

• sample_positions (list[SamplePositionRequest | str | dict[str, Any]]) – the list of sample positions, which is requested by their names. The sample position name is actually the prefix of a sample position, which we will try to match all the sample positions will the name

• exact_positions (Optional[set[str]]) – Set of position names that should be matched exactly (not by prefix). If None, all positions use prefix matching (default behavior).

Return type:

dict[str, list[dict[str, Any]]] | None

set_sample_in_transit(sample_id, source, destination)

Mark a sample as being physically moved from source to destination.

This is meant to be called right before a robot move begins. It does NOT change the recorded position (the sample is still physically at/near source until the move completes). If the move crashes mid-transfer, this record persists, so the last known position plus the intended destination remain visible. move_sample clears it once the sample arrives.

update_sample_metadata(sample_id, metadata)

Update the metadata for a sample. This adds new metadata or updates existing metadata.

update_sample_task_id(sample_id, task_id)

Update the task id for a sample.