Skip to content

Commit

Permalink
Cleaned up for linting
Browse files Browse the repository at this point in the history
  • Loading branch information
@rhfogh
rhfogh committed Oct 11, 2024
1 parent f235ff1 commit 58416cc
Showing 1 changed file with 69 additions and 72 deletions.
141 changes: 69 additions & 72 deletions mxcubecore/HardwareObjects/Gphl/GphlWorkflow.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,6 @@
from mxcubecore.queue_entry import QueueAbortedException

from mxcubecore.HardwareObjects.Gphl import GphlMessages
from mxcubecore.HardwareObjects.Gphl.Transcal2MiniKappa import make_home_data
from mxcubecore import HardwareRepository as HWR


Expand Down Expand Up @@ -132,28 +131,28 @@ class GphlWorkflowStates(enum.Enum):
)

all_point_group_tags = []
for tag in (
for atag in (
"Triclinic",
"Monoclinic",
"Orthorhombic",
"Tetragonal",
"Hexagonal",
"Cubic",
):
all_point_group_tags += lattice2point_group_tags[tag]
all_point_group_tags += lattice2point_group_tags[atag]

# Allowed altervative lattices for a given lattice
alternative_lattices = {}
for ll0 in (
for list0 in (
["aP", "Triclinic"],
["mP", "mC", "mI", "Monoclinic"],
["oP", "oC", "oF", "oI", "Orthorhombic"],
["tP", "tI", "Tetragonal"],
["hP", "hR", "Hexagonal"],
["cP", "cF", "cI", "Cubic"],
):
for tag in ll0:
alternative_lattices[tag] = ll0
for atag in list0:
alternative_lattices[atag] = list0


class GphlWorkflow(HardwareObjectYaml):
Expand Down Expand Up @@ -321,6 +320,7 @@ def query_pre_strategy_params(self, choose_lattice=None):
data_model = self._queue_entry.get_data_model()
strategy_settings = data_model.strategy_settings
space_group = data_model.space_group or ""
header = soldict = select_row = None
if choose_lattice:
header, soldict, select_row = self.parse_indexing_solution(choose_lattice)
lattice = list(soldict.values())[select_row].bravaisLattice
Expand Down Expand Up @@ -1592,7 +1592,6 @@ def setup_data_collection(self, payload, correlation_id):
sweepSettings.reverse()

# Handle centring of first orientation
pos_dict = HWR.beamline.diffractometer.get_positions()
sweepSetting = sweepSettings[0]

# Get current position
Expand Down Expand Up @@ -2093,7 +2092,7 @@ def collect_data(self, payload, correlation_id):

try:
queue_manager.execute_entry(data_collection_entry)
except:
except Exception:
HWR.beamline.queue_manager.emit("queue_execution_failed", (None,))
self._data_collection_group = None

Expand All @@ -2103,7 +2102,7 @@ def collect_data(self, payload, correlation_id):
else:
status = 0

failedScanIds = set(scan.id_ for scan in self._key_to_scan.values())
# failedScanIds = set(scan.id_ for scan in self._key_to_scan.values())

return GphlMessages.CollectionDone(
status=status,
Expand Down Expand Up @@ -2352,69 +2351,69 @@ def process_centring_request(self, payload, correlation_id):
logging.getLogger("HWR").warning(
"No predefined positions for zoom motor."
)
elif True:
else:
logging.getLogger("user_level_log").info(
"Sample re-centering now active - Zoom in before continuing."
)

else:
# TODO The UI popup does not work in mxcubeweb
# NB Temporarily inactivated pending a fix

# Ask user to zoom
info_text = """Automatic sample re-centering is now active
Switch to maximum zoom before continuing"""

schema = {
"title": "GΦL Translational calibration",
"type": "object",
"properties": {},
}
fields = schema["properties"]
fields["_info"] = {
"type": "textdisplay",
"default": info_text,
"readOnly": True,
}
ui_schema = {
"ui:order": ["_info"],
"ui:widget": "vertical_box",
"ui:options": {
"return_signal": self.PARAMETER_RETURN_SIGNAL,
# "update_signal": self.PARAMETER_UPDATE_SIGNAL,
# "update_on_change": "selected",
},
}
self._return_parameters = gevent.event.AsyncResult()
try:
dispatcher.connect(
self.receive_ok_cancel,
self.PARAMETER_RETURN_SIGNAL,
dispatcher.Any,
)
responses = dispatcher.send(
self.PARAMETERS_NEEDED,
self,
schema,
ui_schema,
)
if not responses:
self._return_parameters.set_exception(
RuntimeError(
"Signal %s is not connected" % self.PARAMETERS_NEEDED
)
)

result = self._return_parameters.get()
if result is StopIteration:
return StopIteration
finally:
dispatcher.disconnect(
self.receive_ok_cancel,
self.PARAMETER_RETURN_SIGNAL,
dispatcher.Any,
)
self._return_parameters = None
# else:
# # TODO The UI popup does not work in mxcubeweb
# # NB Temporarily inactivated pending a fix
#
# # Ask user to zoom
# info_text = """Automatic sample re-centering is now active
# Switch to maximum zoom before continuing"""
#
# schema = {
# "title": "GΦL Translational calibration",
# "type": "object",
# "properties": {},
# }
# fields = schema["properties"]
# fields["_info"] = {
# "type": "textdisplay",
# "default": info_text,
# "readOnly": True,
# }
# ui_schema = {
# "ui:order": ["_info"],
# "ui:widget": "vertical_box",
# "ui:options": {
# "return_signal": self.PARAMETER_RETURN_SIGNAL,
# # "update_signal": self.PARAMETER_UPDATE_SIGNAL,
# # "update_on_change": "selected",
# },
# }
# self._return_parameters = gevent.event.AsyncResult()
# try:
# dispatcher.connect(
# self.receive_ok_cancel,
# self.PARAMETER_RETURN_SIGNAL,
# dispatcher.Any,
# )
# responses = dispatcher.send(
# self.PARAMETERS_NEEDED,
# self,
# schema,
# ui_schema,
# )
# if not responses:
# self._return_parameters.set_exception(
# RuntimeError(
# "Signal %s is not connected" % self.PARAMETERS_NEEDED
# )
# )
#
# result = self._return_parameters.get()
# if result is StopIteration:
# return StopIteration
# finally:
# dispatcher.disconnect(
# self.receive_ok_cancel,
# self.PARAMETER_RETURN_SIGNAL,
# dispatcher.Any,
# )
# self._return_parameters = None

settings = goniostatRotation.axisSettings.copy()
if goniostatTranslation is not None:
Expand Down Expand Up @@ -2513,7 +2512,7 @@ def execute_sample_centring(
queue_manager = self._queue_entry.get_queue_controller()
try:
queue_manager.execute_entry(centring_entry)
except:
except Exception:
HWR.beamline.queue_manager.emit("queue_execution_failed", (None,))

centring_result = centring_entry.get_data_model().get_centring_result()
Expand Down Expand Up @@ -2764,7 +2763,7 @@ def get_emulation_sample_dir(self, sample_name=None):
Returns:
"""
sample_dir = None
sample_dir = ""
if sample_name is None:
sample_name = (
self._queue_entry.get_data_model().get_sample_node().get_name()
Expand All @@ -2791,8 +2790,6 @@ def get_emulation_crystal_data(self, sample_name=None):
Returns:
Optional[str]
"""
crystal_data = None
hklfile = None
sample_dir = self.get_emulation_sample_dir(sample_name=sample_name)
if os.path.isdir(sample_dir):
crystal_file = os.path.join(sample_dir, "crystal.nml")
Expand Down

0 comments on commit 58416cc

Please sign in to comment.