AFL.automation.sample.SampleDriver

Functions

cdist(XA, XB[, metric, out])	Compute distance between each pair of the two collections of inputs.
from_uri(uri[, structure_clients, cache, ...])	Connect to a Node on a local or remote server.

Classes

Client([ip, port, username, interactive])	Communicate with APIServer
Driver(name[, defaults, overrides, useful_links])
Eq(key, value)	Query equality of a given key's value to the specified value.
SampleDriver([camera_urls, ...])	PersistentConfig Values client: dict Contains APIServer uris (url:port) where the keys will be used as the accessor names.

class AFL.automation.sample.SampleDriver.SampleDriver(camera_urls: List[str] | None = None, snapshot_directory: str | None = None, overrides: Dict | None = None)

PersistentConfig Values

client: dict

Contains APIServer uris (url:port) where the keys will be used as the accessor names.

instrument: dict

Description and execution/access/location information of each instrument to be used

ternary: bool

If true, process coordinates as ternary, Barycentric values

data_tag: str

Label for current measurements or active learning run

defaults = {'AL_components': [], 'camera_urls': [], 'client': {}, 'components': [], 'composition_var_name': 'comps', 'concat_dim': 'sample', 'custom_stock_settings': [], 'data_path': './', 'data_tag': 'default', 'grid_blank_interval': None, 'grid_blank_sample': None, 'grid_file': None, 'instrument': {}, 'max_sample_transmission': 0.6, 'mix_order': [], 'next_samples_variable': 'next_samples', 'sample_composition_tol': 0.0, 'snapshot_directory': '/github/home/snaps', 'ternary': False}

__init__(camera_urls: List[str] | None = None, snapshot_directory: str | None = None, overrides: Dict | None = None)

validate_config()

validate_config_grid()

Validate configuration specific to grid-based sample processing.

property tiled_client

status()

update_status(value)

get_client(name, refresh=False)

take_snapshot(prefix)

reset_deck()

add_container(name, slot)

add_catch(name, slot)

add_pipette(name, mount, tipracks)

send_deck_config(home=True)

add_stock(stock_dict, loc)

set_catch_protocol(**kwargs)

fix_protocol_order(mix_order: List, custom_stock_settings: Dict)

mfrac_to_mass(mass_fractions: Dict, fixed_conc: Dict, sample_volume, output_units: str = 'mg')

Convert ternary/Barycentric mass fractions to mass

process_sample(composition: Dict, sample_volume: Dict, fixed_concs: Dict, prepare_mfrac_split: Dict | None = None, predict_combine_comps: Dict | None = None, predict_next: bool = False, enqueue_next: bool = False, calibrate_sensor: bool = False, name: str | None = None, sample_uuid: str | None = None, AL_campaign_name: str | None = None, AL_uuid: str | None = None)

Make protocol for, mix, load, and measure sample. Potentially query Agent for next sample and enqueue

Parameters:

• composition (Dict) – Dict should be of the form composition[“component_name”] = {“value”:value, “units”:units}

• sample_volume (dict) – Dict should be of the form sample_volume = {“value”:value, “units”:units}

• fixed_concs (Optional[Dict]) – Dict should be of the form fixed_concs[component] = {“value”:value, “units”:units}

• mfrac_split (Dict) – Dict should be of the form mfrac_split = {‘component_to_split’:{‘component_A’:’mfrac_A’,’component_B’:’mfrac_B’}}

• predict_next (bool) – If True, will trigger predict call to the agent

• enqueue_next (bool) – If True, will pull the next sample from the dropbox of the agent

• calibrate_sensor (bool) – If True, trigger a load stopper sensor recalibration before the next measurement

• name (str) – The name of the sample, if not generated, it will be auto generated from the self.config[‘data_tag’] and uuid

• sample_uuid (str) – uuid of sample, if not specified it will be auto-generated

• AL_uuid (str) – uuid of AL campaign

• AL_campaign_name (str) – name of AL campaign

compute_prep_protocol(composition: Dict, sample_volume: Dict, fixed_concs: Dict, mfrac_split: Dict | None = None)

Parameters:

• composition (Dict) – Dict should be of the form composition[“component_name”] = {“value”:value, “units”:units}

• sample_volume (Dict) – Dict should be of the form sample_volume = {“value”:value, “units”:units}

• mfrac_split (Dict) – Dict should be of the form mfrac_split = {‘component_to_split’:{‘component_A’:’mfrac_A’,’component_B’:’mfrac_B’}}

make_and_measure(name: str, prep_protocol: dict, catch_protocol: dict, calibrate_sensor: bool = False)

measure(name: str, empty: bool = False, wait: bool = True)

instrument = dict (

load_kw = {‘load_dest_label’:’AfterSANS’} client_name = ‘larmor’ measure_base_kw = {‘task_name’: expose, block:True, exposure: 3600} empty_base_kw = {‘task_name’: expose, block:True, exposure: 3600}

)

reset_sample_env(wait: bool = True)

construct_datasets(combine_comps=None)

Construct AL manifest from measurement and call predict

add_new_data_to_agent(combine_comps=None)

Construct AL manifest from measurement and call predict

predict_next_sample()

validate_measurements()

process_sample_grid(sample, name: str | None = None, sample_uuid: str | None = None, AL_campaign_name: str | None = None, AL_uuid: str | None = None, predict_next: bool = False, enqueue_next: bool = False, reset_grid: bool = False)

Process a sample from a grid of samples.

Parameters:

• sample (Dict, optional) – Dictionary containing sample coordinates or properties. If None, will be fetched from the grid or agent.

• name (str, optional) – The name of the sample. If not provided, will be auto-generated.

• sample_uuid (str, optional) – UUID for the sample. If not provided, will be auto-generated.

• AL_campaign_name (str, optional) – Name of the active learning campaign.

• AL_uuid (str, optional) – UUID for the active learning campaign.

• predict_next (bool) – If True, triggers a predict call to the agent.

• enqueue_next (bool) – If True, will enqueue the next sample for measurement.

measure_grid_sample(sample, name, empty=False)

Measure a sample using the grid-based workflow.

instrument = {

‘client_name’: ‘APSUSAXS’, ‘empty_base_kw’: {‘task_name’: ‘expose’}, ‘measure_base_kw’: {‘task_name’: ‘expose’}, ‘select_sample_base_kw’: {‘task_name’: ‘setPosition’, ‘y_offset’: 2}, ‘sample_select_kwargs’: [‘plate’, ‘row’, ‘col’], ‘sample_comps_variable’: ‘sample_composition’

}

Parameters:

• sample (Dict) – Dictionary containing sample coordinates and properties

• name (str) – Sample name for the measurement

construct_grid_datasets(sample: dict)

Construct datasets from grid-based measurements

set_sample(sample_name: str, sample_uuid: str, AL_campaign_name: str | None = None, AL_uuid: str | None = None, AL_components: List | None = None, sample_composition: Dict | None = None)

Set sample information for all clients

Parameters:

• sample_name (str) – Name of the sample

• sample_uuid (str) – UUID of the sample

• AL_campaign_name (str, optional) – Name of the AL campaign

• AL_uuid (str, optional) – UUID of the AL campaign

• AL_components (List, optional) – List of components for AL

• sample_composition (Dict, optional) – Composition of the sample

Returns:

Sample data for client communication

Return type:

Dict

reset_grid()

Reload the grid from file and reset grid_sample_count.