From 8c8a90741f44d1a6c2236e5d81547ceeaae43a9e Mon Sep 17 00:00:00 2001 From: Alejandro De Maria Antolinos Date: Mon, 24 Feb 2025 08:07:14 +0100 Subject: [PATCH 1/3] Added get_samples function to retrieve the acronym --- mxcubecore/HardwareObjects/ICATLIMS.py | 78 +++++++++++++++++++++----- mxcubecore/model/lims_session.py | 48 ++++++++++++++++ 2 files changed, 111 insertions(+), 15 deletions(-) diff --git a/mxcubecore/HardwareObjects/ICATLIMS.py b/mxcubecore/HardwareObjects/ICATLIMS.py index 896809fec4..5848b50881 100644 --- a/mxcubecore/HardwareObjects/ICATLIMS.py +++ b/mxcubecore/HardwareObjects/ICATLIMS.py @@ -21,6 +21,7 @@ from mxcubecore.model.lims_session import ( Lims, LimsSessionManager, + SampleSheet, Session, ) @@ -127,7 +128,10 @@ def get_samples(self, lims_name): self.session_manager.active_session.proposal_name, lims_name, ) - parcels = self.get_parcels_by_investigation_id() + parcels = self.get_parcels() + + sample_sheets = self.get_samples_sheets() + queue_samples = [] for parcel in parcels: pucks = parcel["content"] @@ -150,7 +154,7 @@ def get_samples(self, lims_name): ) for tracking_sample in tracking_samples: queue_samples.append( - self.__to_sample(tracking_sample, puck) + self.__to_sample(tracking_sample, puck, sample_sheets) ) except Exception as e: @@ -169,8 +173,28 @@ def find(self, arr, atribute_name): return x["value"] return "" - def __to_sample(self, tracking_sample, puck): + def get_sample_sheet_by_id(self, samples: List[SampleSheet], sample_id: int) -> Optional[SampleSheet]: + """ + Retrieves a sample by its unique ID. + + Args: + samples (List[Sample]): A list of Sample objects. + sample_id (int): The unique identifier of the sample to retrieve. + + Returns: + Optional[Sample]: The Sample object if found, otherwise None. + """ + return next((sample for sample in samples if sample.id == sample_id), None) + + def __to_sample(self, tracking_sample, puck, sample_sheets: List[SampleSheet]): """Converts the sample tracking into the expected sample data structure""" + + sample_name = str(tracking_sample["name"]) + protein_acronym = sample_name + sample_sheet = self.get_sample_sheet_by_id(sample_sheets, tracking_sample["sampleId"]) + if sample_sheet is not None: + protein_acronym = sample_sheet.name + experiment_plan = tracking_sample["experimentPlan"] return { "cellA": self.find(experiment_plan, "unit_cell_a"), @@ -201,10 +225,10 @@ def __to_sample(self, tracking_sample, puck): "requiredResolution": self.find(experiment_plan, "requiredResolution"), }, "experimentType": self.find(experiment_plan, "workflowType"), - "proteinAcronym": tracking_sample["name"], + "proteinAcronym": protein_acronym, "sampleId": tracking_sample["sampleId"], "sampleLocation": tracking_sample["sampleContainerPosition"], - "sampleName": str(tracking_sample["name"]), + "sampleName": sample_name, "smiles": None, } @@ -481,7 +505,7 @@ def get_user_name(self): def to_sessions(self, investigations): return [self.__to_session(investigation) for investigation in investigations] - def get_parcels_by_investigation_id(self): + def get_parcels(self): """Returns the parcels associated to an investigation""" try: logging.getLogger("HWR").debug( @@ -491,6 +515,7 @@ def get_parcels_by_investigation_id(self): parcels = self.icatClient.get_parcels_by( self.session_manager.active_session.session_id ) + print(parcels[0]) logging.getLogger("HWR").debug( "[ICAT] Successfully retrieved %s parcels" % (len(parcels)) ) @@ -501,6 +526,29 @@ def get_parcels_by_investigation_id(self): ) return [] + def get_samples_sheets(self) -> List[SampleSheet]: + """Returns the samples associated to an investigation""" + try: + logging.getLogger("HWR").debug( + "[ICAT] Retrieving samples by investigation_id %s " + % (self.session_manager.active_session.session_id) + ) + samples = self.icatClient.get_samples_by( + self.session_manager.active_session.session_id + ) + logging.getLogger("HWR").debug( + "[ICAT] Successfully retrieved %s samples" % (len(samples)) + ) + # Convert to object + sample_sheets = [SampleSheet.parse_obj(sample) for sample in samples] + + return sample_sheets + except Exception as e: + logging.getLogger("HWR").error( + "[ICAT] get_samples_by_investigation_id %s " % (str(e)) + ) + return [] + def echo(self): """Mockup for the echo method.""" return True @@ -578,12 +626,12 @@ def add_sample_metadata(self, metadata, collection_parameters): self.__add_sample_changer_position(cell, puck, metadata) metadata["SampleTrackingContainer_position"] = sample_position - metadata["SampleTrackingContainer_type"] = ( - "UNIPUCK" # this could be read from the configuration file somehow - ) - metadata["SampleTrackingContainer_capaticy"] = ( - "16" # this could be read from the configuration file somehow - ) + metadata[ + "SampleTrackingContainer_type" + ] = "UNIPUCK" # this could be read from the configuration file somehow + metadata[ + "SampleTrackingContainer_capaticy" + ] = "16" # this could be read from the configuration file somehow self.__add_protein_acronym(sample_node, metadata) @@ -713,9 +761,9 @@ def finalize_data_collection(self, collection_parameters): # This forces the ingester to associate the dataset to the experiment by ID if self.session_manager.active_session.session_id: - metadata["investigationId"] = ( - self.session_manager.active_session.session_id - ) + metadata[ + "investigationId" + ] = self.session_manager.active_session.session_id # Store metadata on disk self.add_sample_metadata(metadata, collection_parameters) diff --git a/mxcubecore/model/lims_session.py b/mxcubecore/model/lims_session.py index c268d68522..0cb0954e30 100644 --- a/mxcubecore/model/lims_session.py +++ b/mxcubecore/model/lims_session.py @@ -3,6 +3,7 @@ timedelta, ) from typing import ( + Any, Dict, List, Optional, @@ -70,6 +71,43 @@ class Session(BaseModel): logbook_URL: Optional[str] = None +class Instrument(BaseModel): + name: str + id: int + instrumentScientists: List[Any] + + +class Investigation(BaseModel): + name: str + startDate: datetime + endDate: datetime + id: int + title: str + visitId: str + summary: str + parameters: Dict[str, Any] + instrument: Instrument + investigationUsers: List[Any] + + +class Parameter(BaseModel): + name: str + value: str + id: int + units: str + + +class MetaPage(BaseModel): + totalWithoutFilters: int + total: int + totalPages: int + currentPage: int + + +class Meta(BaseModel): + page: MetaPage + + class LimsUser(BaseModel): user_name: str = "" sessions: Optional[List[Session]] = [] @@ -79,3 +117,13 @@ class LimsSessionManager(BaseModel): active_session: Optional[Session] = None sessions: Optional[List[Session]] = [] users: Optional[Dict[str, LimsUser]] = {} + + +class SampleSheet(BaseModel): + id: int + name: str + investigation: Investigation + modTime: datetime + parameters: List[Parameter] + datasets: List[Any] + meta: Meta From 82191139cf9a31c16eb3617dc849807f84350f6c Mon Sep 17 00:00:00 2001 From: Alejandro De Maria Antolinos Date: Mon, 24 Feb 2025 08:35:18 +0100 Subject: [PATCH 2/3] Improved doc inline and removed print --- mxcubecore/HardwareObjects/ICATLIMS.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/mxcubecore/HardwareObjects/ICATLIMS.py b/mxcubecore/HardwareObjects/ICATLIMS.py index 5848b50881..862afb1979 100644 --- a/mxcubecore/HardwareObjects/ICATLIMS.py +++ b/mxcubecore/HardwareObjects/ICATLIMS.py @@ -175,11 +175,11 @@ def find(self, arr, atribute_name): def get_sample_sheet_by_id(self, samples: List[SampleSheet], sample_id: int) -> Optional[SampleSheet]: """ - Retrieves a sample by its unique ID. + Retrieves a sample sheet by its unique ID. Args: - samples (List[Sample]): A list of Sample objects. - sample_id (int): The unique identifier of the sample to retrieve. + samples (List[SampleSheet]): A list of Sample objects. + sample_id (int): The unique identifier of the sample sheet to retrieve. Returns: Optional[Sample]: The Sample object if found, otherwise None. @@ -515,7 +515,7 @@ def get_parcels(self): parcels = self.icatClient.get_parcels_by( self.session_manager.active_session.session_id ) - print(parcels[0]) + logging.getLogger("HWR").debug( "[ICAT] Successfully retrieved %s parcels" % (len(parcels)) ) @@ -527,7 +527,7 @@ def get_parcels(self): return [] def get_samples_sheets(self) -> List[SampleSheet]: - """Returns the samples associated to an investigation""" + """Returns the samples sheets associated to an investigation""" try: logging.getLogger("HWR").debug( "[ICAT] Retrieving samples by investigation_id %s " From 53174cbe9e08d4b079f463d9c20971cb201e0d81 Mon Sep 17 00:00:00 2001 From: Alejandro De Maria Antolinos Date: Mon, 24 Feb 2025 08:54:47 +0100 Subject: [PATCH 3/3] Fix mxcubecore/HardwareObjects/ICATLIMS.py:545:20: RET504 Unnecessary assignment to before statement --- mxcubecore/HardwareObjects/ICATLIMS.py | 30 ++++++++++++++------------ 1 file changed, 16 insertions(+), 14 deletions(-) diff --git a/mxcubecore/HardwareObjects/ICATLIMS.py b/mxcubecore/HardwareObjects/ICATLIMS.py index 862afb1979..c279362f78 100644 --- a/mxcubecore/HardwareObjects/ICATLIMS.py +++ b/mxcubecore/HardwareObjects/ICATLIMS.py @@ -173,7 +173,9 @@ def find(self, arr, atribute_name): return x["value"] return "" - def get_sample_sheet_by_id(self, samples: List[SampleSheet], sample_id: int) -> Optional[SampleSheet]: + def get_sample_sheet_by_id( + self, samples: List[SampleSheet], sample_id: int + ) -> Optional[SampleSheet]: """ Retrieves a sample sheet by its unique ID. @@ -191,7 +193,9 @@ def __to_sample(self, tracking_sample, puck, sample_sheets: List[SampleSheet]): sample_name = str(tracking_sample["name"]) protein_acronym = sample_name - sample_sheet = self.get_sample_sheet_by_id(sample_sheets, tracking_sample["sampleId"]) + sample_sheet = self.get_sample_sheet_by_id( + sample_sheets, tracking_sample["sampleId"] + ) if sample_sheet is not None: protein_acronym = sample_sheet.name @@ -540,9 +544,7 @@ def get_samples_sheets(self) -> List[SampleSheet]: "[ICAT] Successfully retrieved %s samples" % (len(samples)) ) # Convert to object - sample_sheets = [SampleSheet.parse_obj(sample) for sample in samples] - - return sample_sheets + return [SampleSheet.parse_obj(sample) for sample in samples] except Exception as e: logging.getLogger("HWR").error( "[ICAT] get_samples_by_investigation_id %s " % (str(e)) @@ -626,12 +628,12 @@ def add_sample_metadata(self, metadata, collection_parameters): self.__add_sample_changer_position(cell, puck, metadata) metadata["SampleTrackingContainer_position"] = sample_position - metadata[ - "SampleTrackingContainer_type" - ] = "UNIPUCK" # this could be read from the configuration file somehow - metadata[ - "SampleTrackingContainer_capaticy" - ] = "16" # this could be read from the configuration file somehow + metadata["SampleTrackingContainer_type"] = ( + "UNIPUCK" # this could be read from the configuration file somehow + ) + metadata["SampleTrackingContainer_capaticy"] = ( + "16" # this could be read from the configuration file somehow + ) self.__add_protein_acronym(sample_node, metadata) @@ -761,9 +763,9 @@ def finalize_data_collection(self, collection_parameters): # This forces the ingester to associate the dataset to the experiment by ID if self.session_manager.active_session.session_id: - metadata[ - "investigationId" - ] = self.session_manager.active_session.session_id + metadata["investigationId"] = ( + self.session_manager.active_session.session_id + ) # Store metadata on disk self.add_sample_metadata(metadata, collection_parameters)