mxcubecore.HardwareObjects.ICATLIMS#

Classes

DataCollectionMetadataGatherer()

Assembles the metadata for a finished standard MX data collection, in the format expected by ICAT (via pyicat-plus) and metadata.json.

ICATLIMS(name)

class mxcubecore.HardwareObjects.ICATLIMS.DataCollectionMetadataGatherer[source]#

Bases: object

Assembles the metadata for a finished standard MX data collection, in the format expected by ICAT (via pyicat-plus) and metadata.json.

Independent of any LIMS hardware object instance or of how the resulting metadata is subsequently written to disk or uploaded - it only reads beamline/session/queue state (via HWR) and the arguments passed to gather().

gather(datacollection_dict: dict, beamline_config, params: IcatDatasetParameters, extra: dict, scheduled_beamline: str | None = None) → dict[source]#

Assemble the metadata for a finished data collection

Parameters:

  • datacollection_dict (dict) – the collection’s own parameters.

  • beamline_config – beamline configuration object/dict used to add beamline configuration fields to the gathered metadata.

  • params (IcatDatasetParameters) – the partially-filled icat_models.IcatDatasetParameters already produced by gather_common_metadata()

  • extra (dict) – flat ICAT keys with no corresponding model field, as returned alongside params by gather_common_metadata().

  • scheduled_beamline (str | None) – name of the beamline the experiment was scheduled on

Return type:

dict

Returns a dict with keys “metadata”, “file_metadata”, “directory”, “dataset_name”, “beamline”, “proposal” and “snapshot_paths”.

static gather_common_metadata(datacollection_dict: dict, investigation_id: str | None = None, investigation_name: str | None = None, actual_instrument: str | None = None) → Tuple[IcatDatasetParameters, dict][source]#

Assemble the pydantic model fields common to all data collection techniques (energy scans, XFE spectra, and finished data collections alike): sample name, collection start/end time, beam/detector/ energy/transmission/machine/cryo readings.

Returns a tuple (params, extra). params is a partially-filled icat_models.IcatDatasetParameters.blank() instance. extra holds flat ICAT keys with no corresponding model field.

Parameters:

  • datacollection_dict (dict) –

  • investigation_id (str | None) –

  • investigation_name (str | None) –

  • actual_instrument (str | None) –

Return type:

Tuple[IcatDatasetParameters, dict]

class mxcubecore.HardwareObjects.ICATLIMS.ICATLIMS(name)[source]#

Bases: AbstractLims

create_session(session_dict)[source]#

TBD

echo()[source]#

Mockup for the echo method.

finalize_data_collection(datacollection_dict)[source]#

Finalizes the collection with “collection_id”, provided in datacollection_dict.

Structure of datacollection_dict as defined in store_data_collection above.

Parameters:

datacollection_dict –

get_full_user_name()[source]#

Returns the user name of the current user

get_lims_name() → List[Lims][source]#

Returns the LIMS used, name and description

Return type:

List[Lims]

get_proposals_by_user(user_name)[source]#

Returns a list with proposal dictionaries for login_id

Proposal dictionary structure:
{

“Proposal”: proposal, “Person”: , “Laboratory”:, “Session”:,

}

get_samples(lims_name: str) → list[source]#
Retrieve and process sample information from LIMS based on the

provided name: - Retrieves parcel data (containers like UniPucks or SpinePucks). - Retrieves sample sheet data. - Identifies and processes only loaded pucks (those with a ‘sampleChangerLocation’). - Converts each sample in the pucks into internal queue samples using __to_sample.

Parameters:

  • data. (The LIMS name or identifier used to fetch sample-related) –

  • lims_name (str) –

Returns:

A list of processed sample objects ready for queuing.

Return type:

list

get_samples_by_investigation(investigation_id: str) → List[Sample][source]#

Return the sample records associated with an investigation.

Parameters:

investigation_id (str) –

Return type:

List[Sample]

get_user_name()[source]#

Returns the user name of the current user

init()[source]#

Method inherited from baseclass

is_user_login_type() → bool[source]#

Returns True if the login type is user based (not done with proposal)

Return type:

bool

login(username: str, password: str, session_manager: LimsSessionManager | None) → LimsSessionManager[source]#

Login to LIMS, returns a list of Session objects for login_id

Parameters:

  • login_id – username

  • password (str) – password

  • create_session – True if session should be created by LIMS if it does not exist otherwise False

  • username (str) –

  • session_manager (LimsSessionManager | None) –

Return type:

LimsSessionManager

remove_user(user_name: str)[source]#

Drop a signed-out user’s ICAT client/session along with the base-class session-manager bookkeeping. Never evicts the user currently active (matches AbstractLims.remove_user, which refuses to remove a user whose session is the active one).

Parameters:

user_name (str) –

set_active_session_by_id(session_id: str) → Session[source]#

Sets session with session_id to active session

Parameters:

session_id (str) – session id

Return type:

Session

store_beamline_setup(session_id: str, bl_config_dict: dict)[source]#

Stores the beamline setup dict bl_config_dict for session_id

Parameters:

  • session_id (str) – The session id that the beamline_setup should be associated with.

  • bl_config_dict (dict) – The dictionary with beamline settings.

Returns:

The id of the beamline setup.

store_common_data(datacollection_dict: dict) → Tuple[IcatDatasetParameters, dict][source]#

Fill in the pydantic model fields common to all the data collection techniques. :param datacollection_dict: dictionarry from the data collection. :type datacollection_dict: dict

Returns:

A tuple (params, extra). params is a partially-filled icat_models.IcatDatasetParameters.blank() instance. extra holds flat ICAT keys with no corresponding model field.

Parameters:

datacollection_dict (dict) –

Return type:

Tuple[IcatDatasetParameters, dict]

store_data_collection(datacollection_dict, beamline_config_dict=None)[source]#

Store the dictionary with the information about the beamline to be sent when a dataset is produced.

store_energy_scan(energyscan_dict: dict)[source]#

Store energyscan data

Parameters:

energyscan_dict (dict) – Energyscan data to store.

Returns:

int}

Return type:

Dictionary with the energy scan id {“energyScanId”

store_image(image_dict: dict)[source]#

Stores (image parameters) <image_dict>

Parameters:

image_dict (dict) – A dictionary with image pramaters.

store_robot_action(proposal_id: str)[source]#

Stores the robot action dictionary.

Structure of robot_action_dictionary: {

“actionType”:str, “containerLocation”: str, “dewarLocation”:str, “message”:str, “sampleBarcode”:str, “sessionId”:int, “sampleId”:int. “startTime”:str, “endTime”:str, “xtalSnapshotAfter:str”, “xtalSnapshotBefore:str”,

}

Parameters:

  • robot_action_dict – robot action dictionary as defined above

  • proposal_id (str) –

store_workflow(workflow_dict: dict)[source]#

Stores worklflow data workflow_dict

Structure of workflow_dict: {

“workflow_id”: int, “workflow_type”: str, “comments”: str, “log_file_path”: str, “result_file_path”: str, “status”: str, “title”: str, “grid_info_id”: int, “dx_mm”: float, “dy_mm”: float, “mesh_angle”: float, “steps_x”: float, “steps_y”: float, “xOffset”: float, “yOffset”: float,

}

Parameters:

workflow_dict (dict) – worklflow data on the format above

Returns:

Tuple of ints workflow_id, workflow_mesh_id, grid_info_id

store_xfe_spectrum(xfespectrum_dict: dict)[source]#

Stores a XFE spectrum.

Parameters:

xfespectrum_dict (dict) – XFE scan data to store.

Returns:

int}

Return type:

Dictionary with the XFE scan id {“xfeFluorescenceSpectrumId”

update_data_collection(datacollection_dict: dict)[source]#

Update data collection.

Parameters:

datacollection_dict (dict) –