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.

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_available_sample_position(task_id, position_prefix)

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.

Return type:

list[dict[str, str | bool]]

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_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_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_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.

release_sample_position(position)

Unlock a sample position.

request_sample_positions(task_id, sample_positions)

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

Return type:

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

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.