alab_management.resource_manager.resource_requester module

TaskLauncher is the core module of the system, which actually executes the tasks.

exception CombinedTimeoutError

Bases: TimeoutError, TimeoutError

Combined TimeoutError.

If you catch either TimeoutError or concurrent.futures.TimeoutError, this will catch both.

class DeviceRequest(**data)

Bases: BaseModel

Pydantic model for device request.

content: str

identifier: str

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

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

model_config: ClassVar[ConfigDict] = {}

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

model_fields: ClassVar[dict[str, FieldInfo]] = {'content': FieldInfo(annotation=str, required=True), 'identifier': 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.

exception RequestCanceledError

Bases: Exception

Request Canceled Error.

class RequestMixin

Bases: object

Simple wrapper for the request collection.

get_request(request_id, **kwargs)

Get a request by request_id.

Return type:

dict[str, Any] | None

get_requests_by_status(status)

Get all requests by status.

get_requests_by_task_id(task_id)

Get all requests by task_id.

update_request_status(request_id, status, original_status=None)

Update the status of a request by request_id.

class ResourceRequestItem(**data)

Bases: BaseModel

Pydantic model for resource request item.

device: DeviceRequest

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

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

model_config: ClassVar[ConfigDict] = {}

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

model_fields: ClassVar[dict[str, FieldInfo]] = {'device': FieldInfo(annotation=DeviceRequest, required=True), 'sample_positions': FieldInfo(annotation=list[SamplePositionRequest], 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.

sample_positions: list[SamplePositionRequest]

class ResourceRequester(task_id)

Bases: RequestMixin

Class for request lab resources easily. This class will insert a request into the database, and then the task manager will read from the database and assign the resources.

It is used in LabView.

get_concurrent_result(f, timeout=None)

Wait for a resource request, and abort if this task is being cancelled.

Polls in short slices so a cancel does not sit behind an unbounded Future.result wait, and so a cancelled request raises TaskCancelledError instead of hanging until Dramatiq abort arrives.

release_all_resources()

Release all requests by task_id, used for error recovery.

For the requests that are not fulfilled, they will be marked as CANCELED.

For the request that have been fulfilled, they will be marked as NEED_RELEASE.

For the request that have been errored, release assigned resources.

release_resources(request_id)

Release a request by request_id.

request_resources(resource_request, timeout=None, priority=None, exact_positions=None)

Request lab resources.

Write the request into the database, and then the task manager will read from the database and assign the resources.

Parameters:

• resource_request (dict[type[BaseDevice] | str | None, dict[str, int]]) – Dictionary mapping devices to position requests

• timeout (Optional[float]) – Optional timeout for the request

• priority (Union[TaskPriority, int, None]) – Optional priority for the request

• 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, if requesting {None: {“input_rack/slot/1”: 1}}, use exact_positions={“input_rack/slot/1”}. If requesting {Furnace: {“slot/1”: 1}}, use exact_positions={“slot/1”} (the device prefix will be added automatically).

Return type:

dict[str, Any]

class ResourcesRequest(root=PydanticUndefined, **data)

Bases: RootModel

This class is used to validate the resource request. Each request should have a format like this.

[
    {
        "device":{
            "identifier": "name" or "type" or "nodevice",
            "content": string corresponding to identifier
        },
        "sample_positions": [
            {
                "prefix": prefix of sample position,
                "number": integer number of such positions requested.
            },
            ...
        ]
    },
    ...
].

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

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

model_config: ClassVar[ConfigDict] = {}

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

model_fields: ClassVar[dict[str, FieldInfo]] = {'root': FieldInfo(annotation=list[ResourceRequestItem], 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.

classmethod preprocess(values)

Preprocess the request.

root: list[ResourceRequestItem]