# encoding: utf-8
"""Abstract beamline interface message classes
License:
This file is part of MXCuBE.
MXCuBE is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
MXCuBE is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with MXCuBE. If not, see <https://www.gnu.org/licenses/>.
"""
from __future__ import (
absolute_import,
division,
print_function,
unicode_literals,
)
import json
import uuid
from collections import (
OrderedDict,
namedtuple,
)
from mxcubecore.model import crystal_symmetry
__copyright__ = """ Copyright © 2016 - 2019 by Global Phasing Ltd. """
__license__ = "LGPLv3+"
__author__ = "Rasmus H Fogh"
# Enumerations
MESSAGE_INTENTS = {"DOCUMENT", "COMMAND", "EVENT"}
INDEXING_FORMATS = ("IDXREF",)
ABSORPTION_EDGES = ("K", "LI", "LII", "LIII", "MI", "MII", "MIII", "MIV", "MV")
CHEMICAL_ELEMENTS = OrderedDict(
(
("H", "hydrogen"),
("HE", "helium"),
("LI", "lithium"),
("BE", "beryllium"),
("B", "boron"),
("C", "carbon"),
("N", "nitrogen"),
("O", "oxygen"),
("F", "fluorine"),
("NE", "neon"),
("NA", "sodium"),
("MG", "magnesium"),
("AL", "aluminium"),
("SI", "silicon"),
("P", "phosphorus"),
("S", "sulfur"),
("CL", "chlorine"),
("AR", "argon"),
("K", "potassium"),
("CA", "calcium"),
("SC", "scandium"),
("TI", "titanium"),
("V", "vanadium"),
("CR", "chromium"),
("MN", "manganese"),
("FE", "iron"),
("CO", "cobalt"),
("NI", "nickel"),
("CU", "copper"),
("ZN", "zinc"),
("GA", "gallium"),
("GE", "germanium"),
("AS", "arsenic"),
("SE", "selenium"),
("BR", "bromine"),
("KR", "krypton"),
("RB", "rubidium"),
("SR", "strontium"),
("Y", "yttrium"),
("ZR", "zirconium"),
("NB", "niobium"),
("MO", "molybdenum"),
("TC", "technetium"),
("RU", "ruthenium"),
("RH", "rhodium"),
("PD", "palladium"),
("AG", "silver"),
("CD", "cadmium"),
("IN", "indium"),
("SN", "tin"),
("SB", "antimony"),
("TE", "tellurium"),
("I", "iodine"),
("XE", "xenon"),
("CS", "caesium"),
("BA", "barium"),
("LA", "lanthanum"),
("CE", "cerium"),
("PR", "praseodymium"),
("ND", "neodymium"),
("PM", "promethium"),
("SM", "samarium"),
("EU", "europium"),
("GD", "gadolinium"),
("TB", "terbium"),
("DY", "dysprosium"),
("HO", "holmium"),
("ER", "erbium"),
("TM", "thulium"),
("YB", "ytterbium"),
("LU", "lutetium"),
("HF", "hafnium"),
("TA", "tantalum"),
("W", "tungsten"),
("RE", "rhenium"),
("OS", "osmium"),
("IR", "iridium"),
("PT", "platinum"),
("AU", "gold"),
("HG", "mercury"),
("TL", "thallium"),
("PB", "lead"),
("BI", "bismuth"),
("PO", "polonium"),
("AT", "astatine"),
("RN", "radon"),
("FR", "francium"),
("RA", "radium"),
("AC", "actinium"),
("TH", "thorium"),
("PA", "protactinium"),
("U", "uranium"),
("NP", "neptunium"),
("PU", "plutonium"),
("AM", "americium"),
("CM", "curium"),
("BK", "berkelium"),
("CF", "californium"),
("ES", "einsteinium"),
("FM", "fermium"),
("MD", "mendelevium"),
("NO", "nobelium"),
("LR", "lawrencium"),
("RF", "rutherfordium"),
("DB", "dubnium"),
("SG", "seaborgium"),
("BH", "bohrium"),
("HS", "hassium"),
("MT", "meitnerium"),
("DS", "darmstadtium"),
("RG", "roentgenium"),
("CN", "copernicium"),
("UUT", "ununtrium"),
("FL", "flerovium"),
("UUP", "ununpentium"),
("LV", "livermorium"),
)
)
ParsedMessage = namedtuple(
"ParsedMessage", ("message_type", "payload", "enactment_id", "correlation_id")
)
# Abstract classes
[docs]class MessageData(object):
"""Topmost superclass for all message data objects
Later to add e.g. stringification"""
[docs]class Payload(MessageData):
"""Payload - top level message object"""
INTENT = None
def __init__(self):
# This class is abstract
intent = self.__class__.INTENT
if intent not in MESSAGE_INTENTS:
if intent is None:
raise RuntimeError("Attempt to instantiate abstract class Payload")
raise RuntimeError(
"Programming error - "
"Payload subclass %s intent %s must be one of: %s"
% (self.__class__.__name__, intent, sorted(MESSAGE_INTENTS))
)
@property
def intent(self):
"""Message intent - class-level property"""
return self.__class__.INTENT
[docs]class IdentifiedElement(MessageData):
"""Object with persistent uuid"""
def __init__(self, id_=None):
# This class is abstract
if self.__class__.__name__ == "IdentifiedElement":
raise RuntimeError(
"Attempt to instantiate abstract class IdentifiedElement"
)
self._id = None
self.__set_id(id_)
@property
def id_(self):
"""Unique identifier (UUID) for IdentifiedElement.
Defaults to new, time-based uuid"""
return self._id
def __set_id(self, value):
"""Setter for uuid - accessible only within this class"""
if value is None:
self._id = uuid.uuid1()
elif isinstance(value, uuid.UUID):
self._id = value
else:
raise TypeError("UUID input must be of type uuid.UUID")
# Sync with Java 4/5/2017
# Acknowledge???
# Images???
# OrientationMatrix. NB this is a JAva stub. Not needed for now
# ObtainPriorInformation
# PrepareForCentring
# ReadyForCentring
# Intent is now DOCUMENT, COMMAND, EVENT # NBNB TODO
# (data, command, info ca.) I could skip them?
# Simple payloads
[docs]class RequestConfiguration(Payload):
"""Configuration request message"""
def __init__(self, workflowVersion: str | None, abiVersion: str | None):
super().__init__()
self._workflowVersion = workflowVersion
self._abiVersion = abiVersion
INTENT = "COMMAND"
@property
def workflowVersion(self):
return self._workflowVersion
@property
def abiVersion(self):
return self._abiVersion
[docs]class PrepareForCentring(Payload):
"""Prior information request"""
INTENT = "COMMAND"
[docs]class ReadyForCentring(Payload):
"""Prior information request"""
INTENT = "DOCUMENT"
[docs]class SubprocessStopped(Payload):
"""Subprocess Stopped request message"""
INTENT = "EVENT"
[docs]class ConfigurationData(Payload):
"""Configuration Data message"""
INTENT = "DOCUMENT"
# NB coded as mandatory, even if not explicitly non-null
# (but raises MalformedUrlException) in Java.
def __init__(self, location):
super().__init__()
self._location = location
@property
def location(self):
"""Url for directory containing configuration data.
Generally an absolute file path."""
return self._location
[docs]class SubprocessStarted(Payload):
"""Subprocess Started message"""
INTENT = "EVENT"
def __init__(self, name):
super().__init__()
self._name = name
@property
def name(self):
"""name of subprocess"""
return self._name
[docs]class ChooseLattice(Payload):
"""Choose lattice instruction"""
INTENT = "COMMAND"
def __init__(
self,
indexingSolutions,
indexingFormat="IDXREF",
indexingHeader=None,
priorCrystalClasses=(),
priorSpaceGroup=None,
priorSpaceGroupString=None,
userProvidedCell=None,
):
"""
Args:
indexingSolutions (list(IndexingSolution):
indexingFormat (str):
indexingHeader str:
priorCrystalClasses sequence(str):
priorSpaceGroup int:
priorSpaceGroupString str:
userProvidedCell UnitCell:
"""
super().__init__()
self._indexingSolutions = indexingSolutions
self._indexingFormat = indexingFormat
self._indexingHeader = indexingHeader
self._priorCrystalClasses = priorCrystalClasses or ()
self._priorSpaceGroup = priorSpaceGroup
self._priorSpaceGroupString = priorSpaceGroupString
self._userProvidedCell = userProvidedCell
@property
def priorCrystalClasses(self):
"""tuple of crystal class names"""
return self._priorCrystalClasses
@property
def priorSpaceGroup(self):
"""space group number"""
return self._priorSpaceGroup
@property
def priorSpaceGroupString(self):
"""space group name"""
return self._priorSpaceGroupString
@property
def indexingSolutions(self):
"""List of IndexingSolution"""
return self._indexingSolutions
@property
def indexingFormat(self):
"""Indexing format"""
return self._indexingFormat
@property
def indexingHeader(self):
"""Indexing header"""
return self._indexingHeader
@property
def userProvidedCell(self):
"""User provided unit cell"""
return self._userProvidedCell
[docs]class SelectedLattice(MessageData):
"""Lattice selected message"""
INTENT = "DOCUMENT"
def __init__(
self,
data_model,
solution,
):
self._solution = solution
self._strategyDetectorSetting = data_model.detector_setting
self._strategyWavelength = data_model.wavelengths[0]
self._strategyControl = json.dumps(data_model.strategy_options, sort_keys=True)
self._userCrystalClasses = data_model.crystal_classes
sginfo = crystal_symmetry.SPACEGROUP_MAP.get(data_model.space_group)
self._userSpaceGroup = sginfo.number if sginfo else None
if data_model.reference_reflection_files:
self._referenceReflectionFiles = list(data_model.reference_reflection_files)
else:
self._referenceReflectionFiles = []
@property
def solution(self):
"""Proposed solution"""
return self._solution
@property
def strategyDetectorSetting(self):
"""Detector setting to use for strategy calculation and acquisition"""
return self._strategyDetectorSetting
@property
def strategyWavelength(self):
"""Wavelength to use for strategy calculation and acquisition"""
return self._strategyWavelength
@property
def strategyControl(self):
"""JSON string of command line options (*without* prefix)
to use for stratcal wrapper call"""
return self._strategyControl
@property
def userCrystalClasses(self):
"""Crystal classes given by user"""
return self._userCrystalClasses
@property
def userSpaceGroup(self):
"""Space group (int) given by user"""
return self._userSpaceGroup
@property
def referenceReflectionFiles(self):
return list(self._referenceReflectionFiles)
[docs]class IndexingSolution(MessageData):
"""Indexing solution data"""
def __init__(
self,
bravaisLattice,
cell,
isConsistent=None,
latticeCharacter=None,
qualityOfFit=None,
):
"""
Args:
bravaisLattice (string): One of the 14 Bravais lattices ('aP' etc.)
cell (UnitCell):
isConsistent (bool): Is solution consistent with know symmetry?
latticeCharacter (int): Integer 1-44
qualityOfFit (float):
"""
self._bravaisLattice = bravaisLattice
self._cell = cell
self._isConsistent = isConsistent
self._latticeCharacter = latticeCharacter
self._qualityOfFit = qualityOfFit
@property
def bravaisLattice(self):
"""One of the 14 Bravais lattices ('aP' etc.)"""
return self._bravaisLattice
@property
def cell(self):
"""Unit ce;;"""
return self._cell
@property
def isConsistent(self):
"""Is solution consistent with know symmetry?"""
return self._isConsistent
@property
def latticeCharacter(self):
"""Integer 1-44"""
return self._latticeCharacter
@property
def qualityOfFit(self):
""""""
return self._qualityOfFit
[docs]class CollectionDone(MessageData):
"""Collection Done message"""
INTENT = "EVENT"
def __init__(
self,
proposalId,
status,
procWithLatticeParams=False,
imageRoot=None,
scanIdMap=None,
centrings=None,
):
self._proposalId = proposalId
self._imageRoot = imageRoot
self._status = status
self._procWithLatticeParams = procWithLatticeParams
self._scanIdMap = scanIdMap
self._centrings = centrings
@property
def proposalId(self):
"""uuid of collection proposal that has been executed."""
return self._proposalId
@property
def imageRoot(self):
"""Url for directory containing images.
Generally an absolute file path."""
return self._imageRoot
@property
def status(self):
"""Integer status code for collection result"""
return self._status
@property
def procWithLatticeParams(self):
"""Boolean, whether lattice parameters should be used for processing"""
return self._procWithLatticeParams
@property
def scanIdMap(self):
"""Dict[str,str] scan.id_:GoniostatTranslation.id_"""
return self._scanIdMap
@property
def centrings(self):
"""set(GoniostatTranslation)
New GoniostatTranslations acquired during acquisition"""
return self._centrings
# Complex payloads
[docs]class WorkflowDone(Payload):
"""End-of-workflow message"""
INTENT = "EVENT"
def __init__(self, issues=None):
super().__init__()
if self.__class__.__name__ == "WorkflowDone":
raise RuntimeError("Attempt to instantiate abstract class WorkflowDone")
if issues:
ll0 = list(x for x in issues if not isinstance(x, Issue))
if ll0:
raise ValueError("issues parameter contains non-issues: %s" % ll0)
else:
self._issues = tuple(issues)
[docs]class WorkflowCompleted(WorkflowDone):
pass
[docs]class WorkflowAborted(WorkflowDone):
pass
[docs]class WorkflowFailed(WorkflowDone):
def __init__(self, reason=None, issues=None):
super(WorkflowFailed, self).__init__(issues=issues)
self._reason = reason
@property
def reason(self):
return self._reason
[docs]class BeamlineAbort(Payload):
"""Abort workflow from beamline"""
INTENT = "COMMAND"
# Simple data objects
[docs]class AnomalousScatterer(MessageData):
def __init__(self, element, edge):
if element in CHEMICAL_ELEMENTS:
self._element = element
else:
raise ValueError("Chemical element code %s not recognised" % element)
if edge in ABSORPTION_EDGES:
self._edge = edge
else:
raise ValueError("Absorption edge code %s not recognised" % edge)
@property
def element(self):
return self._element
@property
def edge(self):
return self._edge
[docs]class UnitCell(MessageData):
"""Unit cell data type"""
def __init__(self, a, b, c, alpha, beta, gamma):
self._lengths = (a, b, c)
self._angles = (alpha, beta, gamma)
@property
def lengths(self):
return self._lengths
@property
def angles(self):
return self._angles
@property
def a(self):
return self._lengths[0]
@property
def b(self):
return self._lengths[1]
@property
def c(self):
return self._lengths[2]
@property
def alpha(self):
return self._angles[0]
@property
def beta(self):
return self._angles[1]
@property
def gamma(self):
return self._angles[2]
[docs]class Issue(IdentifiedElement):
"""Issue (status information returned with WorkflowDone messages)"""
def __init__(self, component, message, code=None, id_=None):
IdentifiedElement.__init__(self, id_)
self._component = component
self._message = message
self._code = code
@property
def component(self):
"""Part of unique identifier of issue definition"""
return self._component
@property
def code(self):
"""Part of unique identifier of issue definition"""
return self._code
@property
def message(self):
"""Text providing specific details about this issue"""
return self._message
[docs]class PhasingWavelength(IdentifiedElement):
"""Phasing Wavelength"""
def __init__(self, wavelength, role=None, id_=None):
IdentifiedElement.__init__(self, id_)
self._role = role
self._wavelength = wavelength
@property
def role(self):
"""Wavelength role"""
return self._role
@property
def wavelength(self):
"""Wavelength setting for beam"""
return self._wavelength
@wavelength.setter
def wavelength(self, value):
if self._wavelength:
raise TypeError("PhasingWavelength values cannot be re-set if non-zero")
self._wavelength = value
[docs]class BeamSetting(IdentifiedElement):
"""Beam setting"""
def __init__(self, wavelength, id_=None):
IdentifiedElement.__init__(self, id_)
self._wavelength = wavelength
@property
def wavelength(self):
"""Wavelength setting for beam"""
return self._wavelength
[docs]class ScanExposure(IdentifiedElement):
"""Scan Exposure"""
def __init__(self, time, transmission, id_=None):
IdentifiedElement.__init__(self, id_)
self._time = time
self._transmission = transmission
@property
def transmission(self):
"""Scan exposure transmission"""
return self._transmission
@property
def time(self):
"""Scan exposure transmission"""
return self._time
[docs]class ScanWidth(IdentifiedElement):
"""Scan Width"""
def __init__(self, imageWidth, numImages, id_=None):
IdentifiedElement.__init__(self, id_)
self._imageWidth = imageWidth
self._numImages = numImages
@property
def imageWidth(self):
"""Scan image width"""
return self._imageWidth
@property
def numImages(self):
"""Number of images"""
return self._numImages
[docs]class PositionerSetting(IdentifiedElement):
"""Positioner Setting object.
Has a uuid and a settings dictionary of axisName:value
"""
def __init__(self, id_=None, **axisSettings):
super().__init__(id_=id_)
if self.__class__.__name__ == "PositionerSetting":
# This class is abstract
raise RuntimeError(
"Programming error -"
" attempt to instantiate abstract class PositionerSetting"
)
if None in axisSettings.values():
raise ValueError(
"axisSettings contain value None: %s" % sorted(axisSettings.items())
)
self._axisSettings = axisSettings.copy()
@property
def axisSettings(self):
"""axisName:value settings dictionary. NB the returned value is a copy;
modifying it does *not* modify the object internals."""
return self._axisSettings.copy()
[docs]class DetectorSetting(PositionerSetting):
"""Detector position setting"""
[docs]class BcsDetectorSetting(DetectorSetting):
"""Detector position setting with additional (beamline-side) resolution and orgxy"""
def __init__(self, resolution, id_=None, orgxy=(), **axisSettings):
super(BcsDetectorSetting, self).__init__(id_=id_, **axisSettings)
self._resolution = resolution
self._orgxy = tuple(orgxy)
@property
def resolution(self):
"""Resolution (in A) matching detector distance"""
return self._resolution
@property
def orgxy(self):
"""Tuple, empty or of two floats; beam centre on detector"""
return self._orgxy
[docs]class BeamstopSetting(PositionerSetting):
"""Beamstop position setting"""
[docs]class GoniostatRotation(PositionerSetting):
"""Goniostat Rotation setting"""
def __init__(self, id_=None, **axisSettings):
PositionerSetting.__init__(self, id_=id_, **axisSettings)
self._translation = None
@property
def translation(self):
"""GoniostatTranslation corresponding to self
NB This link can be set only by GoniostatTranslation.__init__"""
return self._translation
[docs] def get_motor_settings(self):
"""Get dictionary of rotation and translation motor setting"""
result = dict(self.axisSettings)
translation = self.translation
if translation is not None:
result.update(translation.axisSettings)
return result
[docs]class GoniostatSweepSetting(GoniostatRotation):
"""Goniostat Sweep setting"""
def __init__(self, scanAxis, id_=None, **axisSettings):
GoniostatRotation.__init__(self, id_=id_, **axisSettings)
self._scanAxis = scanAxis
@property
def scanAxis(self):
"""Scanning axis"""
return self._scanAxis
[docs]class GoniostatTranslation(PositionerSetting):
"""Goniostat Translation setting
NB the reverse GoniostatRotation.translation link is set from here.
For this reason the constructor parameters are different from the
object attributes. rotation is taken to be newRotation, except that:
if ( rotation is not None and
(requestedRotationId is None or requestedRotationId == rotation.id_):
self.requestedRotationId = rotation.id_
self.newRotation = None"""
def __init__(
self, rotation=None, requestedRotationId=None, id_=None, **axisSettings
):
PositionerSetting.__init__(self, id_=id_, **axisSettings)
if rotation is None:
if requestedRotationId is None:
raise ValueError("rotation and requestedRotationId cannot both be None")
else:
self._newRotation = None
self._requestedRotationId = requestedRotationId
else:
if requestedRotationId is None or requestedRotationId == rotation.id_:
self._newRotation = None
self._requestedRotationId = rotation.id_
else:
self._newRotation = rotation
self._requestedRotationId = requestedRotationId
# NBNB this deliberately interferes with the internals of
# GoniostatRotation
rotation._translation = self
@property
def newRotation(self):
return self._newRotation
@property
def requestedRotationId(self):
return self._requestedRotationId
# Complex data objects
[docs]class UserProvidedInfo(MessageData):
"""User-provided information"""
def __init__(self, data_model):
self._scatterers = ()
crystal_classes = data_model.crystal_classes
if crystal_classes:
self._crystalClasses = tuple(crystal_classes)
else:
self._crystalClasses = ()
space_group = data_model.space_group
sginfo = crystal_symmetry.SPACEGROUP_MAP.get(data_model.space_group)
self._spaceGroup = sginfo.number if sginfo else None
self._spaceGroupString = space_group or None
cell_parameters = data_model.cell_parameters
if cell_parameters:
self._cell = UnitCell(*cell_parameters)
else:
self._cell = None
detector_setting = data_model.detector_setting
if detector_setting:
self._expectedResolution = detector_setting.resolution
else:
self._expectedResolution = data_model.aimed_resolution
self._isAnisotropic = None
@property
def scatterers(self):
return self._scatterers
@property
def crystalClasses(self):
return self._crystalClasses
@property
def spaceGroup(self):
return self._spaceGroup
@property
def spaceGroupString(self):
return self._spaceGroupString
@property
def cell(self):
return self._cell
@property
def expectedResolution(self):
return self._expectedResolution
@property
def isAnisotropic(self):
return self._isAnisotropic
[docs]class Sweep(IdentifiedElement):
"""Geometric strategy Sweep"""
def __init__(
self,
goniostatSweepSetting,
detectorSetting,
beamSetting,
start,
width,
beamstopSetting=None,
sweepGroup=None,
id_=None,
):
super(Sweep, self).__init__(id_=id_)
self._scans = set()
self._goniostatSweepSetting = goniostatSweepSetting
self._detectorSetting = detectorSetting
self._beamSetting = beamSetting
self._beamstopSetting = beamstopSetting
self._start = start
self._width = width
self._sweepGroup = sweepGroup
@property
def goniostatSweepSetting(self):
return self._goniostatSweepSetting
@property
def detectorSetting(self):
return self._detectorSetting
@property
def beamSetting(self):
return self._beamSetting
@property
def beamstopSetting(self):
return self._beamstopSetting
@property
def start(self):
return self._start
@property
def width(self):
return self._width
@property
def sweepGroup(self):
return self._sweepGroup
@property
def scans(self):
"""Scans that belong to sweeps.
NB this is a two-way link.
It is populated when the Scan objects are created"""
return frozenset(self._scans)
def _add_scan(self, scan):
"""Implementation method. *Only* to be called from Scan.__init__"""
self._scans.add(scan)
[docs] def get_initial_settings(self):
"""Get dict of rotation and translation motor settings for start of sweep"""
result = self.goniostatSweepSetting.get_motor_settings()
result[self.goniostatSweepSetting.scanAxis] = self.start
return result
[docs]class Scan(IdentifiedElement):
"""Collection strategy Scan"""
def __init__(
self, width, exposure, imageStartNum, start, sweep, filenameParams, id_=None
):
super().__init__(id_=id_)
self._filenameParams = dict(filenameParams)
self._width = width
self._exposure = exposure
self._imageStartNum = imageStartNum
self._start = start
sweep._add_scan(self)
self._sweep = sweep
@property
def width(self):
return self._width
@property
def exposure(self):
return self._exposure
@property
def imageStartNum(self):
return self._imageStartNum
@property
def start(self):
return self._start
@property
def sweep(self):
return self._sweep
@property
def filenameParams(self):
return dict(self._filenameParams)
[docs]class GeometricStrategy(IdentifiedElement, Payload):
"""Geometric strategy"""
INTENT = "COMMAND"
def __init__(
self,
isUserModifiable,
defaultDetectorSetting,
defaultBeamSetting,
allowedWidths=(),
sweepOffset=None,
sweepRepeat=None,
defaultWidthIdx=None,
sweeps=(),
reflectingRangeEsd=None,
id_=None,
):
super().__init__(id_=id_)
# self._isInterleaved = isInterleaved
self._isUserModifiable = isUserModifiable
self._defaultDetectorSetting = defaultDetectorSetting
self._defaultBeamSetting = defaultBeamSetting
self._defaultWidthIdx = defaultWidthIdx
self._sweeps = frozenset(sweeps)
self._reflectingRangeEsd = reflectingRangeEsd
if len(set(allowedWidths)) != len(allowedWidths):
raise ValueError(
"allowedWidths contains duplicate value: %s" % allowedWidths
)
else:
self._allowedWidths = tuple(allowedWidths)
self._sweepOffset = sweepOffset
self._sweepRepeat = sweepRepeat
@property
def sweepRepeat(self):
return self._sweepRepeat
@property
def sweepOffset(self):
return self._sweepOffset
@property
def isUserModifiable(self):
return self._isUserModifiable
@property
def defaultWidthIdx(self):
return self._defaultWidthIdx
@property
def defaultDetectorSetting(self):
return self._defaultDetectorSetting
@property
def defaultBeamSetting(self):
return self._defaultBeamSetting
@property
def allowedWidths(self):
return self._allowedWidths
@property
def sweeps(self):
return self._sweeps
@property
def reflectingRangeEsd(self):
return self._reflectingRangeEsd
[docs] def get_ordered_sweeps(self):
"""Get sweeps in acquisition order.
Acquisition order is determined by the sweepGroup -
to get results deterministic we use a secondary sort on
angles, in alphabetical name order as a backup,
(in pracite, 'kappa', 'kappa_phi', 'phi')
which should match what the workflow does internally.
Anyway, it is the sweep"""
ll0 = []
for sweep in self._sweeps:
dd0 = sweep.get_initial_settings()
ll0.append((sweep.sweepGroup, tuple(dd0[x] for x in sorted(dd0)), sweep))
return list(tt0[2] for tt0 in sorted(ll0))
[docs]class CollectionProposal(IdentifiedElement, Payload):
"""Collection proposal"""
INTENT = "COMMAND"
def __init__(self, relativeImageDir, strategy, scans, id_=None):
super().__init__(id_=id_)
self._relativeImageDir = relativeImageDir
self._strategy = strategy
self._scans = tuple(scans)
@property
def relativeImageDir(self):
return self._relativeImageDir
@property
def strategy(self):
return self._strategy
@property
def scans(self):
return self._scans
[docs]class RequestCentring(Payload):
"""Request for centring"""
INTENT = "COMMAND"
def __init__(self, currentSettingNo, totalRotations, goniostatRotation):
super(RequestCentring, self).__init__()
self._currentSettingNo = currentSettingNo
self._totalRotations = totalRotations
self._goniostatRotation = goniostatRotation
@property
def currentSettingNo(self):
return self._currentSettingNo
@property
def totalRotations(self):
return self._totalRotations
@property
def goniostatRotation(self):
return self._goniostatRotation
[docs]class CentringDone(Payload):
"""Centring-done message"""
INTENT = "DOCUMENT"
def __init__(self, status, timestamp, goniostatTranslation):
super().__init__()
self._status = status
self._timestamp = timestamp
self._goniostatTranslation = goniostatTranslation
@property
def status(self):
return self._status
@property
def timestamp(self):
"""Time in seconds since the epoch (Jan 1, 1970),
as returned by time.time()"""
return self._timestamp
@property
def goniostatTranslation(self):
return self._goniostatTranslation
[docs]class SampleCentred(Payload):
INTENT = "DOCUMENT"
def __init__(self, data_model):
super().__init__()
self._imageWidth = data_model.image_width
self._transmission = 0.01 * data_model.transmission
self._exposure = data_model.exposure_time
self._wedgeWidth = round(data_model.wedge_width / data_model.image_width)
self._interleaveOrder = data_model.interleave_order
self._beamstopSetting = data_model.beamstop_setting
self._goniostatTranslations = frozenset(data_model.goniostat_translations)
self._repetition_count = data_model.repetition_count
self._detectorSetting = None
if data_model.characterisation_done:
self._wavelengths = tuple(data_model.wavelengths)
else:
# This trick assumes that characterisation and diffractcal
# use one, the first, wavelength and default interleave order
# Which is true. Not the ideal place to put this code,
# but it works.
self._wavelengths = tuple((data_model.wavelengths[0],))
# if data_model.wftype != "diffractcal":
# self._detectorSetting = data_model.detector_setting
if data_model.wftype != "diffractcal":
self._detectorSetting = data_model.detector_setting
@property
def imageWidth(self):
return self._imageWidth
@property
def transmission(self):
return self._transmission
@property
def exposure(self):
return self._exposure
@property
def interleaveOrder(self):
return self._interleaveOrder
@property
def wedgeWidth(self):
# Wedge width NB in *number of images*
return self._wedgeWidth
@property
def beamstopSetting(self):
return self._beamstopSetting
@property
def detectorSetting(self):
return self._detectorSetting
@property
def goniostatTranslations(self):
return self._goniostatTranslations
@property
def wavelengths(self):
return self._wavelengths
@property
def repetition_count(self):
return self._repetition_count