"""
[Name] PlateManipulator
[Description]
Plate manipulator hardware object is used to use diffractometer in plate mode.
It is compatible with md2, md3 diffractometers. Class is based on
SampleChanger, so it has all the sample changed functionalities, like
mount, unmount sample (in this case move to plate position).
Plate is organized in rows and columns. Each cell (Cell) contains drop (Drop).
Each drop could contain several crystals (Xtal). If CRIMS is available then
each drop could have several crystals.
Commands and channels defined in the diffractometer
[Emitted signals]
- emitted signals defined in SampleChanger class
[Included Hardware Objects]
-----------------------------------------------------------------------
| name | signals | functions
-----------------------------------------------------------------------
-----------------------------------------------------------------------
"""
import logging
from gevent import Timeout, sleep
from mxcubecore import HardwareRepository as HWR
from mxcubecore.HardwareObjects.abstract.AbstractSampleChanger import (
SampleChanger,
SampleChangerMode,
SampleChangerState,
)
from mxcubecore.HardwareObjects.abstract.sample_changer import Crims
from mxcubecore.HardwareObjects.abstract.sample_changer.Container import (
Basket,
Container,
Sample,
)
[docs]class Xtal(Sample):
__NAME_PROPERTY__ = "Name"
__LOGIN_PROPERTY__ = "Login"
def __init__(self, drop, index):
super().__init__(drop, Xtal._get_xtal_address(drop, index), False)
self._drop = drop
self._index = index
self._set_image_x(None)
self._set_image_y(None)
self._set_image_url(None)
self._set_name(None)
self._set_login(None)
self._set_info_url(None)
self._set_info(False, False, False)
self._set_loaded(False, False)
self.present = True
def _set_name(self, value):
self._set_property(self.__NAME_PROPERTY__, value)
def _set_login(self, value):
self._set_property(self.__LOGIN_PROPERTY__, value)
def get_login(self):
return self.get_property(self.__LOGIN_PROPERTY__)
def get_drop(self):
return self._drop
def get_cell(self):
return self.get_drop().get_cell()
[docs] def get_basket_no(self):
"""
In this cas we assume a drop is a basket or puck
"""
return self.get_drop().get_index() + 1
[docs] def get_cell_no(self):
"""
In this cas we assume wells in the row is a cell
"""
return self.get_cell().get_row_index() + 1
@staticmethod
def _get_xtal_address(drop, index):
return str(drop.get_address()) + "-" + str(index)
[docs] def get_index(self):
"""
Descript. : Sample index is calculated relative to the row (Basket)
In this case we assume that in drop is one xtal
This should be changed to various num of xtals in the drop
"""
cell_index = self.get_cell().get_index()
drops_in_cell_num = self.get_cell().get_drops_no()
drop_index = self._drop.get_index()
return cell_index * drops_in_cell_num + drop_index
[docs] 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,
)
[docs]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
)
self._cell = cell
self._drops_num = drops_num
@staticmethod
def _get_drop_address(cell, drop_num):
return str(cell.get_address()) + ":" + str(drop_num)
def get_cell(self):
return self._cell
def get_well_no(self):
return self.get_index() + 1
[docs] def is_loaded(self):
"""
Returns if the sample is currently loaded for data collection
:rtype: bool
"""
sample = self.get_sample()
return sample.is_loaded()
[docs] def get_sample(self):
"""
In this cas we assume that there is one crystal per drop
"""
sample = self.get_components()
return sample[0]
[docs]class Cell(Container):
__TYPE__ = "Cell"
def __init__(self, row, row_chr, col_index, drops_num):
super().__init__(
self.__TYPE__, row, self._get_cell_address(row_chr, col_index), False
)
self._row = row
self._row_chr = row_chr
self._col_index = col_index
self._drops_num = drops_num
for drop_index in range(self._drops_num):
drop = Drop(self, drop_index + 1)
self._add_component(drop)
xtal = Xtal(drop, drop.get_number_of_components())
drop._add_component(xtal)
self._transient = True
def get_row(self):
return self._row
def get_row_chr(self):
return self._row_chr
def get_row_index(self):
return ord(self._row_chr.upper()) - ord("A")
def get_col(self):
return self._col_index
def get_drops_no(self):
return self._drops_num
@staticmethod
def _get_cell_address(row, col):
return str(row) + str(col)
class PlateManipulator(SampleChanger):
""" """
__TYPE__ = "PlateManipulator"
def __init__(self, *args, **kwargs):
super().__init__(self.__TYPE__, False, *args, **kwargs)
self.plate_label = None
self.num_cols = None
self.num_rows = None
self.num_drops = None
self.reference_pos_x = None
self.stored_pos_x = None
self.stored_pos_y = None
self.crims_url = None
self.crims_user_agent = None
self.plate_barcode = None
self.harvester_key = None
self.processing_plan = None
self.plate_locatiom = None
self.diffr = None
def init(self):
"""Initialise the properties"""
self.diffr = HWR.beamline.diffractometer
try:
(
self.num_rows,
self.num_cols,
self.num_drops,
) = self.diffr.plate_topology.get_value()
except ValueError:
self.log.exception("")
except AttributeError:
self.num_rows = self.get_property("numRows")
self.num_cols = self.get_property("numCols")
self.num_drops = self.get_property("numDrops")
self.reference_pos_x = self.get_property("referencePosX", 0.5)
self.stored_pos_x = self.reference_pos_x
self.stored_pos_y = 0.5
self.plate_label = self.get_property("plate_label", "")
self.crims_url = self.get_property("crimsWsRoot")
self.crims_user_agent = self.get_property("crimsUserAgent")
self.plate_barcode = self.get_property("PlateBarcode")
self.harvester_key = self.get_property("harvesterKey")
self._init_sc_contents()
try:
self.diffr.plate_location.connect_signal(
"update", self.plate_location_changed
)
except AttributeError:
self.log.exception("plate_location is not configured")
self.plate_location_changed()
super().init()
def change_plate_barcode(self, barcode):
if self._load_data(barcode):
self.plate_barcode = barcode
return True
else:
raise Exception("barcode unknown")
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 (
chr(65 + loaded_sample[0])
+ str(loaded_sample[1] + 1)
+ ":"
+ str(loaded_sample[2] + 1)
+ "-0"
)
return loaded_sample
def plate_location_changed(self, plate_location=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):
try:
self.plate_location_changed()
self._on_state_changed(state)
except AttributeError:
self.log.exception("")
def _on_state_changed(self, state):
"""
Descript. : state change callback. Based on diffractometer state
sets PlateManipulator state.
"""
if state is None:
self._set_state(SampleChangerState.Unknown)
else:
if state == "Alarm":
self._set_state(SampleChangerState.Alarm)
elif state == "Fault":
self._set_state(SampleChangerState.Fault)
elif state == "Moving" or state == "Running":
self._set_state(SampleChangerState.Moving)
elif state == "Ready":
if self.diffr.get_phase() == self.diffr.get_phase_enum.TRANSFER:
self._set_state(SampleChangerState.Charging)
elif self.diffr.get_phase() == self.diffr.get_phase_enum.CENTRE:
self._set_state(SampleChangerState.Ready)
else:
self._set_state(SampleChangerState.StandBy)
elif state == "Initializing":
self._set_state(SampleChangerState.Initializing)
def _init_sc_contents(self):
"""
Descript. : Initializes content of plate.
"""
if self.num_rows is None:
return
self._set_info(False, None, False)
self._clear_components()
for row in range(self.num_rows):
# row is like a basket
basket = Basket(self, row + 1, samples_num=0, name="Row")
present = True
datamatrix = ""
scanned = False
basket._set_info(present, datamatrix, scanned)
self._add_component(basket)
for col in range(self.num_cols):
cell = Cell(basket, chr(65 + row), col + 1, self.num_drops)
basket._add_component(cell)
def _do_abort(self):
"""Local abort command"""
self.diffr.abort()
def _do_change_mode(self, mode: SampleChangerMode):
"""Change the diffractometer phase as function of sample changer mode.
Args:
mode: Sample changer mode
"""
if mode == SampleChangerMode.Charging:
self.diffr.set_phase(self.diffr.get_phase_enum.TRANSFER)
elif mode == SampleChangerMode.Normal:
self.diffr.set_phase(self.diffr.get_phase_enum.CENTRE)
def _do_load(self, sample=None):
"""Actual load action.
Args:
sample: If None, use the selected sample.
"""
selected = self.get_selected_sample()
if sample is None:
sample = self.get_selected_sample()
if sample is not None:
if sample != selected:
self._do_select(sample)
self._set_loaded_sample(sample)
def load(self, sample: tuple | None = None, wait: bool = True):
"""Load sample.
Args:
sample: sample address on the form
(component1, ... ,component_N-1, component_N
wait: True - wait for the load to finish.
"""
comp = self._resolve_component(sample)
coords = comp.get_coords()
res = self._load_sample(coords)
if res:
self._set_loaded_sample(comp)
comp._set_loaded(True, True)
self.plate_location_changed()
return res
def _load_sample(self, sample_location=None, pos_x=None, pos_y=None, wait=True):
"""
Location is estimated by sample location and reference positions.
"""
try:
row = sample_location[0] - 1
col = int((sample_location[1] - 1) / self.num_drops)
drop = sample_location[1] - self.num_drops * col
if pos_x is None:
pos_x = self.stored_pos_x
else:
self.stored_pos_x = pos_x
if pos_y is None:
pos_y = self.stored_pos_y
else:
self.stored_pos_y = pos_y
if hasattr(self.diffr, "move_to_drop"):
self.diffr.move_to_drop(row, col, drop - 1)
self._wait_ready(60)
elif hasattr(self.diffr, "move_to_location"):
self.diffr.move_to_location(row, col, pos_x, pos_y)
self._wait_ready(60)
else:
# 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)
)
new_sample = drop.get_sample()
old_sample = self.get_loaded_sample()
new_sample = drop.get_sample()
if old_sample != new_sample:
if old_sample is not None:
old_sample._set_loaded(False, True)
if new_sample is not None:
new_sample._set_loaded(True, True)
return True
except Exception:
return False
def _do_unload(self, sample_slot=None):
"""
Descript. :
"""
self._reset_loaded_sample()
self._on_state_changed("Ready")
def _do_reset(self):
"""
Descript. :
"""
self._reset(False)
self._wait_device_ready()
def _do_scan(self, component, recursive):
"""
Descript. :
"""
if not isinstance(component, PlateManipulator):
raise Exception("Not supported")
self._initializeData()
if self.get_token() is None:
raise Exception("No plate barcode defined")
self._load_data(self.get_token())
def sync_with_crims(self):
"""
Descript. :
Get Crims information
"""
self.processing_plan = self._load_data(self.plate_barcode)
return self.processing_plan
def _do_select(self, component):
"""
Descript. :
"""
if isinstance(component, Xtal):
self._select_sample(
component.get_cell().get_row_index(),
component.get_cell().get_col() - 1,
component.get_drop().get_well_no() - 1,
)
self._set_selected_sample(component)
component.get_container()._set_selected(True)
component.get_container().get_container()._set_selected(True)
elif isinstance(component, Crims.CrimsXtal):
col = component.Column - 1
row = ord(component.Row.upper()) - ord("A")
cell = self.get_component_by_address(
Cell._get_cell_address(component.Row, component.Column)
)
drop = self.get_component_by_address(
Drop._get_drop_address(cell, component.Shelf)
)
drop._set_selected(True)
drop.get_container()._set_selected(True)
elif isinstance(component, Drop):
self._select_sample(
component.get_cell().get_row_index(),
component.get_cell().get_col() - 1,
component.get_well_no() - 1,
)
component._set_selected(True)
component.get_container().get_container()._set_selected(True)
elif isinstance(component, Cell):
self._select_sample(component.get_row_index(), component.get_col() - 1, 0)
component._set_selected(True)
elif isinstance(component, list):
row = component[0]
col = component[1]
cell = self.get_component_by_address(Cell._get_cell_address(row, col))
cell._set_selected(True)
else:
raise Exception("Invalid selection")
self._reset_loaded_sample()
self._wait_device_ready()
def _load_data(self, plate_barcode):
processing_plan = Crims.get_processing_plan(
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
)
logging.getLogger("user_level_log").error(msg)
else:
msg = (
"Information about plate with barcode %s found in CRIMS" % plate_barcode
)
logging.getLogger("user_level_log").info(msg)
self._set_info(True, processing_plan.plate.barcode, True)
for x in processing_plan.plate.xtal_list:
cell = self.get_component_by_address(
Cell._get_cell_address(x.row, x.column)
)
cell._set_info(True, "", True)
drop = self.get_component_by_address(
Drop._get_drop_address(cell, x.shelf)
)
drop._set_info(True, "", True)
xtal = Xtal(drop, drop.get_number_of_components())
xtal._set_info(True, x.pin_id, True)
xtal._set_image_url(x.image_url)
xtal._set_image_x(x.offset_x)
xtal._set_image_y(x.offset_y)
xtal._set_login(x.login)
xtal._set_name(x.sample)
xtal._set_info_url(x.summary_url)
drop._add_component(xtal)
return processing_plan
def _do_update_info(self):
"""
Descript. :
"""
self._update_state()
def update_info(self):
self._do_update_info()
self._trigger_loaded_sample_changed_event(self.get_loaded_sample())
self._reset_dirty()
def _read_state(self):
return self.diffr.global_state.get_value()
def _update_state(self):
"""Update the state"""
state = self.diffr._get_swstate
self._on_state_changed(state)
return state
def _update_loaded_sample(self):
"""Updates plate location"""
if self.plate_location is not None:
new_sample = self.get_loaded_sample()
if new_sample is not None:
# self._update_sample_barcode(new_sample)
loaded = True
has_been_loaded = True
new_sample._set_loaded(loaded, has_been_loaded)
self._set_loaded_sample(new_sample)
def get_loaded_sample(self):
sample = None
for _s in self.get_sample_list():
if _s.get_address() == self.hw_get_loaded_sample_location():
sample = _s
return sample
def get_sample(self, plate_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
cell = self.get_component_by_address("%s%d" % (chr(65 + row), col + 1))
if cell:
drop = cell.get_component_by_address(
"%s%d:%d" % (chr(65 + row), col + 1, drop_index)
)
return drop.get_sample()
def get_sample_list(self):
"""
Descript. : This is ugly
"""
sample_list = []
for basket in self.get_components():
if isinstance(basket, Basket):
for cell in basket.get_components():
if isinstance(cell, Cell):
for drop in cell.get_components():
sample_list.append(drop.get_sample())
return sample_list
def is_mounted_sample(self, sample_location):
row = sample_location[0] - 1
col = (sample_location[1] - 1) / self.num_drops
drop = sample_location[1] - self.num_drops * col
pos_y = float(drop) / (self.num_drops + 1)
sample = self.get_sample((row, col, drop, pos_y))
return sample.loaded
@property
def _ready(self):
return self._update_state() == "Ready"
def _wait_ready(self, timeout: float | None = None):
"""Wait timeout seconds until status is ready.
Args:
timeout: optional - timeout [s],
if timeout = 0: return at once and do not wait,
if timeout is None: wait forever (default).
"""
if timeout == 0:
return
with Timeout(timeout, RuntimeError("Timeout waiting for status ready")):
while not self._ready:
sleep(0.5)
def get_plate_info(self):
"""
Descript. : returns dict with plate info
"""
plate_info_dict = {}
plate_info_dict["plate_label"] = self.plate_label or "Demo plate label"
plate_info_dict["plate_barcode"] = self.plate_barcode or ""
plate_info_dict["num_cols"] = self.num_cols
plate_info_dict["num_rows"] = self.num_rows
plate_info_dict["num_drops"] = self.num_drops
return plate_info_dict
def move_to_crystal_position(self, crystal_uuid):
"""
Descript. : Move Diff to crystal position
Get crystal_uuid from processing plan for loaded sample/drop
"""
ret = None
if crystal_uuid in ["undefined", None]:
loaded_sample = self.diffr.drop_location.get_value()
row = int(loaded_sample[0])
col = int(loaded_sample[1])
drop = int(loaded_sample[2])
if self.processing_plan:
for x in self.processing_plan.plate.xtal_list:
if (
row == ord(x.row) - 65
and col == x.column - 1
and drop == x.shelf - 1
):
crystal_uuid = x.crystal_uuid
else:
msg = "No processing_plan OR Crystal Found in this well."
logging.getLogger("user_level_log").info(msg)
if hasattr(self.diffr, "move_to_xtal_position") and crystal_uuid:
try:
ret = self.diffr.move_to_xtal_position(self.plate_barcode, crystal_uuid)
except Exception as ex:
raise RuntimeError("Could not move to crystal position") from ex
else:
msg = "move_to_crystal_position command or crystal UUID not found"
logging.getLogger("user_level_log").info(msg)
return ret