diff --git a/mxcubecore/HardwareObjects/PlateManipulator.py b/mxcubecore/HardwareObjects/PlateManipulator.py index 60f59a883e..5b25c1b6fe 100644 --- a/mxcubecore/HardwareObjects/PlateManipulator.py +++ b/mxcubecore/HardwareObjects/PlateManipulator.py @@ -46,7 +46,7 @@ class Xtal(Sample): __LOGIN_PROPERTY__ = "Login" def __init__(self, drop, index): - super().__init__(drop, Xtal._get_xtal_address(drop, index), False) + super().__init__(drop, self._get_xtal_address(drop, index), False) self._drop = drop self._index = index self._set_image_x(None) @@ -106,19 +106,15 @@ def get_container(self): return self.get_cell().get_container() def get_name(self): - return "%s%d:%d" % ( - self.get_cell().get_row_chr(), - self.get_cell().get_index() + 1, - self._drop.get_index() + 1, - ) + return f"{self.get_cell().get_row_chr()}{self.get_cell().get_index() + 1}:{self._drop.get_index() + 1}" class Drop(Container): __TYPE__ = "Drop" def __init__(self, cell, drops_num): - super(Drop, self).__init__( - self.__TYPE__, cell, Drop._get_drop_address(cell, drops_num), False + super().__init__( + self.__TYPE__, cell, self._get_drop_address(cell, drops_num), False ) self._cell = cell self._drops_num = drops_num @@ -202,6 +198,7 @@ def __init__(self, *args, **kwargs): self.reference_pos_x = None self.stored_pos_x = None self.stored_pos_y = None + self.plate_location = None self.crims_url = None self.crims_user_agent = None self.plate_barcode = None @@ -212,7 +209,7 @@ def __init__(self, *args, **kwargs): self.diffr = None def init(self): - """Initialise the properties""" + """Initialise the properties.""" self.diffr = HWR.beamline.diffractometer try: ( @@ -246,18 +243,21 @@ def init(self): except AttributeError: self.log.exception("plate_location is not configured") - self.plate_location_changed() + self.plate_location_changed(self.diffr.plate_location.get_value()) + super().init() - def change_plate_barcode(self, barcode): + def change_plate_barcode(self, barcode: str) -> bool: + """Set the plate barcode. + Args: + barcode: The barcode. + """ if self._load_data(barcode): self.plate_barcode = barcode return True - else: - raise Exception("barcode unknown") + raise Exception("Unknown barcode") def hw_get_loaded_sample_location(self): - loaded_sample = None if hasattr(self.diffr, "drop_location"): loaded_sample = self.diffr.drop_location.get_value() return ( @@ -267,14 +267,18 @@ def hw_get_loaded_sample_location(self): + str(loaded_sample[2] + 1) + "-0" ) - return loaded_sample + return None - def plate_location_changed(self, plate_location=None): + def plate_location_changed(self, plate_location: tuple | None = None): self.plate_location = plate_location or self.diffr.plate_location.get_value() self._update_loaded_sample() self.update_info() def state_changed(self, state): + """Change the state + Args: + state: the new state to be updated. + """ try: self.plate_location_changed() self._on_state_changed(state) @@ -293,7 +297,7 @@ def _on_state_changed(self, state): self._set_state(SampleChangerState.Alarm) elif state == "Fault": self._set_state(SampleChangerState.Fault) - elif state == "Moving" or state == "Running": + elif state in ("Moving", "Running"): self._set_state(SampleChangerState.Moving) elif state == "Ready": if self.diffr.get_phase() == self.diffr.get_phase_enum.TRANSFER: @@ -369,7 +373,7 @@ def load(self, sample: tuple | None = None, wait: bool = True): self.plate_location_changed() return res - def _load_sample(self, sample_location=None, pos_x=None, pos_y=None, wait=True): + def _load_sample(self, sample_location=None, pos_x=None, pos_y=None): """ Location is estimated by sample location and reference positions. """ @@ -398,10 +402,9 @@ def _load_sample(self, sample_location=None, pos_x=None, pos_y=None, wait=True): # No actual move cmd defined. Act like a mockup self.plate_location = [row, col, pos_x, pos_y] col += 1 - cell = self.get_component_by_address("%s%d" % (chr(65 + row), col)) - drop = cell.get_component_by_address( - "%s%d:%d" % (chr(65 + row), col, drop) - ) + _addr = f"{chr(65 + row)}{col}" + cell = self.get_component_by_address(_addr) + drop = cell.get_component_by_address(f"{_addr}:{drop}") new_sample = drop.get_sample() old_sample = self.get_loaded_sample() new_sample = drop.get_sample() @@ -429,28 +432,21 @@ def _do_reset(self): self._wait_device_ready() def _do_scan(self, component, recursive): - """ - Descript. : - """ + """Scan the plate to get all the samples.""" if not isinstance(component, PlateManipulator): - raise Exception("Not supported") + raise ValueError("Not supported") self._initializeData() if self.get_token() is None: - raise Exception("No plate barcode defined") + raise ValueError("No plate barcode defined") self._load_data(self.get_token()) def sync_with_crims(self): - """ - Descript. : - Get Crims information - """ + """Get plate information from Crims""" self.processing_plan = self._load_data(self.plate_barcode) return self.processing_plan def _do_select(self, component): - """ - Descript. : - """ + """Select a sample.""" if isinstance(component, Xtal): self._select_sample( component.get_cell().get_row_index(), @@ -488,7 +484,7 @@ def _do_select(self, component): cell = self.get_component_by_address(Cell._get_cell_address(row, col)) cell._set_selected(True) else: - raise Exception("Invalid selection") + raise RuntimeError("Invalid selection") self._reset_loaded_sample() self._wait_device_ready() @@ -497,15 +493,10 @@ def _load_data(self, plate_barcode): plate_barcode, self.crims_url, self.crims_user_agent, self.harvester_key ) if processing_plan is None: - msg = ( - "No information about plate with barcode %s found in CRIMS" - % plate_barcode - ) + msg = f"No information for plate with barcode {plate_barcode} in CRIMS" logging.getLogger("user_level_log").error(msg) else: - msg = ( - "Information about plate with barcode %s found in CRIMS" % plate_barcode - ) + msg = "Information for plate with barcode {plate_barcode} in CRIMS" logging.getLogger("user_level_log").info(msg) self._set_info(True, processing_plan.plate.barcode, True) @@ -568,19 +559,22 @@ def get_loaded_sample(self): return sample def get_sample(self, plate_location): + """Convert the plate location to sample name + Args: + plate_location: The location + """ row = int(plate_location[0]) col = int(plate_location[1]) y_pos = float(plate_location[3]) drop_index = abs(y_pos * self.num_drops) + 1 - if drop_index > self.num_drops: - drop_index = self.num_drops + drop_index = min(drop_index, self.num_drops) - cell = self.get_component_by_address("%s%d" % (chr(65 + row), col + 1)) + _addr = f"{chr(65 + row)}{(col + 1)}" + cell = self.get_component_by_address(_addr) if cell: - drop = cell.get_component_by_address( - "%s%d:%d" % (chr(65 + row), col + 1, drop_index) - ) + drop = cell.get_component_by_address(f"{_addr}:{drop_index}") return drop.get_sample() + return () def get_sample_list(self): """ diff --git a/mxcubecore/HardwareObjects/PlateManipulatorMaintenance.py b/mxcubecore/HardwareObjects/PlateManipulatorMaintenance.py index bf69c1edc0..c42cccc56a 100644 --- a/mxcubecore/HardwareObjects/PlateManipulatorMaintenance.py +++ b/mxcubecore/HardwareObjects/PlateManipulatorMaintenance.py @@ -16,40 +16,14 @@ class PlateManipulatorMaintenance(HardwareObject): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) + self._scan_limits = "" self._sc = None def init(self): self._sc = HWR.beamline.sample_changer - self._scan_limits = "" - - def _do_abort(self): - """ - Abort current command - - :returns: None - :rtype: None - """ - return self._sc._do_abort() - - def _move_to_crystal_position(self, args): - """ - command to move MD head to x, y crystal position - argrs: crystall uuid or none for current drop position - - :returns: None if exception - :rtype: None - """ - return self._sc.move_to_crystal_position(args) - - def _do_change_mode(self, args): - self._sc._do_change_mode(args) def set_plate_barcode(self, args): - """ - Plate barcode is set in config, but can be update here - :returns: None - :rtype: None - """ + """Plate barcode is set in config, but can be update here""" ret = self._sc.change_plate_barcode(args) if ret: self._update_global_state() @@ -59,16 +33,13 @@ def _update_global_state(self): self.emit("globalStateChanged", (state_dict, cmd_state, message)) def get_global_state(self): - """ """ - state = self._sc._read_state().upper() - scan_limits = self._scan_limits - running = state in ("RUNNING",) - plate_info_dict = self._sc.get_plate_info() + """Return the gloobal state.""" + state = self._sc._read_state() state_dict = { - "running": running, - "scan_limits": scan_limits, + "running": state.upper() == "RUNNING", + "scan_limits": self._scan_limits, "state": state, - "plate_info": plate_info_dict, + "plate_info": self._sc.get_plate_info(), } cmd_state = { @@ -81,7 +52,9 @@ def get_cmd_info(self): """return information about existing commands for this object the information is organized as a list with each element contains - [cmd_id, cmd_display_name, nb_args, cmd_category, description ]""" + [ cmd_name, display_name, category ] + [cmd_id, cmd_display_name, nb_args, cmd_category, description ] + """ cmd_list = [ [ @@ -95,12 +68,20 @@ def get_cmd_info(self): return cmd_list def send_command(self, cmdname, args=None): + """Send command to the sample changer. + Args: + cmdname: The command name + args: Arguments, if any + """ if cmdname in ["getOmegaMotorDynamicScanLimits"]: - self._get_scan_limits(args) + self._scan_limits = self._sc.diffr.scan_limits() + self._update_global_state() if cmdname in ["moveToCrystalPosition"]: - self._move_to_crystal_position(args) + self._sc.move_to_crystal_position(args) if cmdname == "abort": - self._do_abort() + self._sc._do_abort() if cmdname == "setPlateBarcode": self.set_plate_barcode(args) + # if cmdname == "change_mode": + # self._sc._do_change_mode(args) return True diff --git a/ruff.toml b/ruff.toml index 1944d84c05..8e19cbd653 100644 --- a/ruff.toml +++ b/ruff.toml @@ -617,6 +617,9 @@ convention = "google" "E501", "S301", ] +"mxcubecore/HardwareObjects/PlateManipulator.py" = [ + "E501", +] "mxcubecore/HardwareObjects/ProposalTypeISPyBLims.py" = [ "E402", "E501",