diff --git a/src/cal_tools/calcat_interface.py b/src/cal_tools/calcat_interface.py
index 4b1e2288af408af050c57a51037206014950ad1a..d70e3e91cd46afa179cff495389291e5b6531f59 100644
--- a/src/cal_tools/calcat_interface.py
+++ b/src/cal_tools/calcat_interface.py
@@ -1,6 +1,4 @@
-
"""Interfaces to calibration constant data."""
-
import re
import socket
from datetime import date, datetime, time, timezone
@@ -20,16 +18,18 @@ from calibration_client.modules import (
PhysicalDetectorUnit,
)
+from cal_tools.tools import module_index_to_qm
+
__all__ = [
- 'CalCatError',
- 'CalibrationData',
- 'AGIPD_CalibrationData',
- 'LPD_CalibrationData',
- 'DSSC_CalibrationData',
- 'JUNGFRAU_CalibrationData',
- 'PNCCD_CalibrationData',
- 'EPIX100_CalibrationData',
- 'GOTTHARD2_CalibrationData',
+ "CalCatError",
+ "CalibrationData",
+ "AGIPD_CalibrationData",
+ "LPD_CalibrationData",
+ "DSSC_CalibrationData",
+ "JUNGFRAU_CalibrationData",
+ "PNCCD_CalibrationData",
+ "EPIX100_CalibrationData",
+ "GOTTHARD2_CalibrationData",
]
@@ -45,18 +45,22 @@ class CCVMetadata(dict):
import pandas as pd
- res = {pdu_idx: {calibration: ccv_data['ccv_name']
- for calibration, ccv_data in pdu_data.items()}
- for pdu_idx, pdu_data in self.items()}
+ res = {
+ pdu_idx: {
+ calibration: ccv_data["ccv_name"]
+ for calibration, ccv_data in pdu_data.items()
+ }
+ for pdu_idx, pdu_data in self.items()
+ }
- return str(pd.DataFrame.from_dict(res, orient='index'))
+ return str(pd.DataFrame.from_dict(res, orient="index"))
class CalCatError(Exception):
"""CalCat API error."""
def __init__(self, response):
- super().__init__(response['info'])
+ super().__init__(response["info"])
class ClientWrapper(type):
@@ -77,11 +81,22 @@ class CalCatApi(metaclass=ClientWrapper):
"""Internal calibration_client wrapper."""
get_detector_keys = [
- 'id', 'name', 'identifier', 'karabo_name',
- 'karabo_id_control', 'description']
+ "id",
+ "name",
+ "identifier",
+ "karabo_name",
+ "karabo_id_control",
+ "description",
+ ]
get_pdu_keys = [
- 'id', 'physical_name', 'karabo_da', 'virtual_device_name',
- 'detector_type_id', 'detector_id', 'description']
+ "id",
+ "physical_name",
+ "karabo_da",
+ "virtual_device_name",
+ "detector_type_id",
+ "detector_id",
+ "description",
+ ]
def __init__(self, client):
self.client = client
@@ -91,7 +106,7 @@ class CalCatApi(metaclass=ClientWrapper):
"""Parse different ways to specify time to CalCat."""
if isinstance(dt, datetime):
- return dt.astimezone(timezone.utc).strftime('%Y%m%dT%H%M%S%Z')
+ return dt.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%S%Z")
elif isinstance(dt, date):
return cls.format_time(datetime.combine(dt, time()))
@@ -108,48 +123,71 @@ class CalCatApi(metaclass=ClientWrapper):
(dict) Operating condition for use in CalCat API.
"""
- return {'parameters_conditions_attributes': [
- {'parameter_id': self.parameter_id(k), 'value': str(v)}
- for k, v in condition.items()]}
+ return {
+ "parameters_conditions_attributes": [
+ {"parameter_id": self.parameter_id(k), "value": str(v)}
+ for k, v in condition.items()
+ ]
+ }
@lru_cache()
def detector(self, detector_name):
"""Detector metadata."""
- resp_detector = Detector.get_by_identifier(
- self.client, detector_name)
+ resp_detector = Detector.get_by_identifier(self.client, detector_name)
- if not resp_detector['success']:
+ if not resp_detector["success"]:
raise CalCatError(resp_detector)
- return {k: resp_detector['data'][k] for k in self.get_detector_keys}
+ return {k: resp_detector["data"][k] for k in self.get_detector_keys}
@lru_cache()
- def physical_detector_units(self, detector_id, snapshot_at):
+ def physical_detector_units(self, detector_id, snapshot_at, da_to="da"):
"""Physical detector unit metadata."""
resp_pdus = PhysicalDetectorUnit.get_all_by_detector(
- self.client, detector_id, self.format_time(snapshot_at))
+ self.client, detector_id, self.format_time(snapshot_at)
+ )
- if not resp_pdus['success']:
+ if not resp_pdus["success"]:
raise CalCatError(resp_pdus)
- return {
- pdu['karabo_da']: {
- k: pdu[k] for k in self.get_pdu_keys}
- for pdu in resp_pdus['data']
- }
+
+ # Create dict based on requested keys: karabo_da, module number,
+ # or QxMx naming convention.
+ if da_to == "da":
+ return {
+ pdu["karabo_da"]: {k: pdu[k] for k in self.get_pdu_keys}
+ for pdu in resp_pdus["data"]
+ }
+ elif da_to == "modno":
+ return {
+ int(pdu["karabo_da"][-2:]): {
+ k: pdu[k] for k in self.get_pdu_keys
+ }
+ for pdu in resp_pdus["data"]
+ }
+ elif da_to == "qm":
+ return {
+ module_index_to_qm(int(pdu["karabo_da"][-2:])): {
+ k: pdu[k] for k in self.get_pdu_keys
+ }
+ for pdu in resp_pdus["data"]
+ }
+ else:
+ raise ValueError(f"{da_to} is unknown!")
@lru_cache()
def calibration_id(self, calibration_name):
"""ID for a calibration in CalCat."""
resp_calibration = Calibration.get_by_name(
- self.client, calibration_name)
+ self.client, calibration_name
+ )
- if not resp_calibration['success']:
+ if not resp_calibration["success"]:
raise CalCatError(resp_calibration)
- return resp_calibration['data']['id']
+ return resp_calibration["data"]["id"]
@lru_cache()
def parameter_id(self, param_name):
@@ -157,14 +195,21 @@ class CalCatApi(metaclass=ClientWrapper):
resp_parameter = Parameter.get_by_name(self.client, param_name)
- if not resp_parameter['success']:
+ if not resp_parameter["success"]:
raise CalCatError(resp_parameter)
- return resp_parameter['data']['id']
+ return resp_parameter["data"]["id"]
def closest_ccv_by_time_by_condition(
- self, detector_name, calibrations, condition,
- modules=None, event_at=None, snapshot_at=None, metadata=None,
+ self,
+ detector_name,
+ calibrations,
+ condition,
+ modules=None,
+ event_at=None,
+ snapshot_at=None,
+ metadata=None,
+ da_to="da",
):
"""Query bulk CCV metadata from CalCat.
@@ -196,13 +241,6 @@ class CalCatApi(metaclass=ClientWrapper):
event_at = self.format_time(event_at)
snapshot_at = self.format_time(snapshot_at)
- # Map aggregator to module number.
- # da_to_modno = {
- # data['karabo_da']: modno for modno, data in
- # self.physical_detector_units(
- # self.detector(detector_name)['id'], snapshot_at).items()
- # if not modules or modno in modules}
-
if metadata is None:
metadata = CCVMetadata()
@@ -214,49 +252,68 @@ class CalCatApi(metaclass=ClientWrapper):
return metadata
# Map calibration ID to calibration name.
- cal_id_map = {self.calibration_id(calibration): calibration
- for calibration in calibrations}
+ cal_id_map = {
+ self.calibration_id(calibration): calibration
+ for calibration in calibrations
+ }
calibration_ids = list(cal_id_map.keys())
# The API call supports a single module or all modules, as the
# performance increase is only minor in between. Hence, all
# modules are queried if more than one is selected and filtered
# afterwards, if necessary.
- karabo_da = next(iter(modules)) if modules is not None and len(modules) == 1 else '', # noqa
+ karabo_da = (
+ next(iter(modules))
+ if modules is not None and len(modules) == 1
+ else "",
+ ) # noqa
resp_versions = CalibrationConstantVersion.get_closest_by_time_by_detector_conditions( # noqa
- self.client, detector_name, calibration_ids,
+ self.client,
+ detector_name,
+ calibration_ids,
self.format_cond(condition),
karabo_da=karabo_da,
- event_at=event_at, snapshot_at=snapshot_at)
+ event_at=event_at,
+ snapshot_at=snapshot_at,
+ )
- if not resp_versions['success']:
+ if not resp_versions["success"]:
raise CalCatError(resp_versions)
- for ccv in resp_versions['data']:
+ for ccv in resp_versions["data"]:
try:
- mod = ccv['physical_detector_unit']['karabo_da']
+ if da_to == "da":
+ mod = ccv["physical_detector_unit"]["karabo_da"]
+ elif da_to == "qm": # Can be used for AGIPD, LPD, and DSSC.
+ mod = module_index_to_qm(
+ int(ccv["physical_detector_unit"]["karabo_da"][-2:])
+ )
+ elif da_to == "modno":
+ mod = int(ccv["physical_detector_unit"]["karabo_da"][-2:])
+ else:
+ raise ValueError(f"{da_to} is unknown!")
except KeyError:
# Not included in our modules
continue
- cc = ccv['calibration_constant']
+ cc = ccv["calibration_constant"]
metadata.setdefault(mod, dict())[
- cal_id_map[cc['calibration_id']]] = dict(
- cc_id=cc['id'],
- cc_name=cc['name'],
- condition_id=cc['condition_id'],
- ccv_id=ccv['id'],
- ccv_name=ccv['name'],
- path=Path(ccv['path_to_file']) / ccv['file_name'],
- dataset=ccv['data_set_name'],
- begin_validity_at=ccv['begin_validity_at'],
- end_validity_at=ccv['end_validity_at'],
- raw_data_location=ccv['raw_data_location'],
- start_idx=ccv['start_idx'],
- end_idx=ccv['end_idx'],
- physical_name=ccv[
- 'physical_detector_unit']['physical_name'],
- )
+ cal_id_map[cc["calibration_id"]]
+ ] = dict(
+ cc_id=cc["id"],
+ cc_name=cc["name"],
+ condition_id=cc["condition_id"],
+ ccv_id=ccv["id"],
+ ccv_name=ccv["name"],
+ path=Path(ccv["path_to_file"]) / ccv["file_name"],
+ dataset=ccv["data_set_name"],
+ begin_validity_at=ccv["begin_validity_at"],
+ end_validity_at=ccv["end_validity_at"],
+ raw_data_location=ccv["raw_data_location"],
+ start_idx=ccv["start_idx"],
+ end_idx=ccv["end_idx"],
+ physical_name=ccv["physical_detector_unit"]["physical_name"],
+ )
return metadata
@@ -300,8 +357,15 @@ class CalibrationData:
calibrations = set()
default_client = None
- def __init__(self, detector_name, modules=None, client=None, event_at=None,
- snapshot_at=None):
+ def __init__(
+ self,
+ detector_name,
+ modules=None,
+ client=None,
+ event_at=None,
+ snapshot_at=None,
+ da_to="da",
+ ):
"""Initialize a new CalibrationData object.
If no calibration-client object is passed or has been created
@@ -326,10 +390,13 @@ class CalibrationData:
self.modules = modules
self.event_at = event_at
self.snapshot_at = snapshot_at
+ self.da_to = da_to
if client is None:
- client = self.__class__.default_client or \
- self.__class__.new_anonymous_client()
+ client = (
+ self.__class__.default_client
+ or self.__class__.new_anonymous_client()
+ )
self._api = CalCatApi(client)
@@ -338,21 +405,31 @@ class CalibrationData:
"""Create an anonymous calibration-client object.
This connection allows read-only access to CalCat using a
- facility-proveded OAuth reverse proxy. This is only accessible
+ facility-provided OAuth reverse proxy. This is only accessible
on the European XFEL computing infrastructure.
"""
- print('Access to CalCat via the XFEL OAuth proxy is currently '
- 'considered in testing, please report any issues to '
- 'da-support@xfel.eu')
+ print(
+ "Access to CalCat via the XFEL OAuth proxy is currently "
+ "considered in testing, please report any issues to "
+ "da-support@xfel.eu"
+ )
return CalibrationData.new_client(
- None, None, None, use_oauth2=False,
- base_url='http://exflcalproxy:8080/')
+ None,
+ None,
+ None,
+ use_oauth2=False,
+ base_url="http://exflcalproxy:8080/",
+ )
@staticmethod
def new_client(
- client_id, client_secret, user_email, installation='',
- base_url='https://in.xfel.eu/{}calibration', **kwargs,
+ client_id,
+ client_secret,
+ user_email,
+ installation="",
+ base_url="https://in.xfel.eu/{}calibration",
+ **kwargs,
):
"""Create a new calibration-client object.
@@ -375,7 +452,7 @@ class CalibrationData:
(CalibrationClient) CalCat client.
"""
- base_url = base_url.format(f'{installation}_' if installation else '')
+ base_url = base_url.format(f"{installation}_" if installation else "")
# Note this is not a classmethod and we're modifying
# CalibrationData directly to use the same object across all
@@ -384,12 +461,12 @@ class CalibrationData:
client_id=client_id,
client_secret=client_secret,
user_email=user_email,
- base_api_url=f'{base_url}/api/',
- token_url=f'{base_url}/oauth/token',
- refresh_url=f'{base_url}/oauth/token',
- auth_url=f'{base_url}/oauth/authorize',
- scope='',
- **kwargs
+ base_api_url=f"{base_url}/api/",
+ token_url=f"{base_url}/oauth/token",
+ refresh_url=f"{base_url}/oauth/token",
+ auth_url=f"{base_url}/oauth/authorize",
+ scope="",
+ **kwargs,
)
return CalibrationData.default_client
@@ -402,14 +479,15 @@ class CalibrationData:
None if not available.
"""
- if not hasattr(CalibrationData, '_caldb_root'):
- if getenv('SASE'):
+ if not hasattr(CalibrationData, "_caldb_root"):
+ if getenv("SASE"):
# ONC
- CalibrationData._caldb_root = Path('/common/cal/caldb_store')
- elif re.match(r'^max-(.+)\.desy\.de$', socket.getfqdn()):
+ CalibrationData._caldb_root = Path("/common/cal/caldb_store")
+ elif re.match(r"^max-(.+)\.desy\.de$", socket.getfqdn()):
# Maxwell
CalibrationData._caldb_root = Path(
- '/gpfs/exfel/d/cal/caldb_store')
+ "/gpfs/exfel/d/cal/caldb_store"
+ )
else:
# Probably unavailable
CalibrationData._caldb_root = None
@@ -427,7 +505,8 @@ class CalibrationData:
@property
def physical_detector_units(self):
return self._api.physical_detector_units(
- self.detector['id'], self.snapshot_at)
+ self.detector["id"], self.snapshot_at, self.da_to
+ )
@property
def condition(self):
@@ -437,10 +516,12 @@ class CalibrationData:
"""Create a new CalibrationData object with altered values."""
keys = {
- 'detector_name', 'modules', 'client', 'event_at', 'snapshot_at'
- } | {
- self._simplify_parameter_name(name)for name in self.parameters
- }
+ "detector_name",
+ "modules",
+ "client",
+ "event_at",
+ "snapshot_at",
+ } | {self._simplify_parameter_name(name) for name in self.parameters}
kwargs = {key: getattr(self, key) for key in keys}
kwargs.update(new_kwargs)
@@ -448,7 +529,10 @@ class CalibrationData:
return self.__class__(**kwargs)
def metadata(
- self, calibrations=None, event_at=None, snapshot_at=None,
+ self,
+ calibrations=None,
+ event_at=None,
+ snapshot_at=None,
):
"""Query CCV metadata for calibrations, conditions and time.
@@ -468,14 +552,22 @@ class CalibrationData:
metadata = CCVMetadata()
self._api.closest_ccv_by_time_by_condition(
- self.detector_name, calibrations or self.calibrations,
- self.condition, self.modules,
- event_at or self.event_at, snapshot_at or self.snapshot_at,
- metadata)
+ self.detector_name,
+ calibrations or self.calibrations,
+ self.condition,
+ self.modules,
+ event_at or self.event_at,
+ snapshot_at or self.snapshot_at,
+ metadata,
+ da_to=self.da_to,
+ )
return metadata
def ndarray(
- self, module, calibration, metadata,
+ self,
+ module,
+ calibration,
+ metadata,
):
"""Load CCV data as ndarray.
@@ -489,10 +581,10 @@ class CalibrationData:
(ndarray): CCV data
"""
if self.caldb_root is None:
- raise RuntimeError('calibration database store unavailable')
+ raise RuntimeError("calibration database store unavailable")
if self.modules and module not in self.modules:
- raise ValueError('module not part of this calibration data')
+ raise ValueError("module not part of this calibration data")
if metadata is None:
metadata = self.metadata([calibration])
@@ -510,15 +602,16 @@ class CalibrationData:
dtype = cf[f"{dataset}/data"].dtype
const_data[mod][cname] = const_load_mp.alloc(
- shape=shape, dtype=dtype)
+ shape=shape, dtype=dtype
+ )
def load_constants_data(self, metadata):
-
def load_constant_dataset(wid, index, mod):
for cname, mdata in metadata[mod].items():
with h5py.File(self.caldb_root / mdata["path"], "r") as cf:
cf[f"{mdata['dataset']}/data"].read_direct(
- const_data[mod][cname])
+ const_data[mod][cname]
+ )
const_data = dict()
const_load_mp = psh.ProcessContext(num_workers=24)
@@ -528,7 +621,9 @@ class CalibrationData:
return const_data
def ndarray_map(
- self, calibrations=None, metadata=None,
+ self,
+ calibrations=None,
+ metadata=None,
):
"""Load all CCV data in a nested map of ndarrays.
@@ -543,7 +638,7 @@ class CalibrationData:
{module: {calibration: ndarray}}
"""
if self.caldb_root is None:
- raise RuntimeError('calibration database store unavailable')
+ raise RuntimeError("calibration database store unavailable")
if metadata is None:
metadata = self.metadata(calibrations)
@@ -551,7 +646,9 @@ class CalibrationData:
return self.load_constants_data(metadata)
def data_map(
- self, calibrations=None, metadata=None,
+ self,
+ calibrations=None,
+ metadata=None,
):
"""Load all CCV data in a nested map of ndarrays.
@@ -566,7 +663,7 @@ class CalibrationData:
{module: {calibration: ndarray}}
"""
if self.caldb_root is None:
- raise RuntimeError('calibration database store unavailable')
+ raise RuntimeError("calibration database store unavailable")
if metadata is None:
metadata = self.metadata(calibrations)
@@ -586,8 +683,12 @@ class CalibrationData:
@classmethod
def _from_multimod_detector_data(
- cls, component_cls, data, detector,
- modules, client,
+ cls,
+ component_cls,
+ data,
+ detector,
+ modules,
+ client,
):
if isinstance(detector, component_cls):
detector_name = detector.detector_name
@@ -596,11 +697,14 @@ class CalibrationData:
elif isinstance(detector, str):
detector_name = detector
else:
- raise ValueError(f'detector may be an object of type '
- f'{type(cls)}, a string or None')
+ raise ValueError(
+ f"detector may be an object of type "
+ f"{type(cls)}, a string or None"
+ )
- source_to_modno = dict(component_cls._source_matches(
- data, detector_name))
+ source_to_modno = dict(
+ component_cls._source_matches(data, detector_name)
+ )
detector_sources = [data[source] for source in source_to_modno.keys()]
if modules is None:
@@ -610,12 +714,16 @@ class CalibrationData:
# Create new CalibrationData object.
caldata = cls(
- detector_name, modules, client,
- creation_date, creation_date,
- )
+ detector_name,
+ modules,
+ client,
+ creation_date,
+ creation_date,
+ )
caldata.memory_cells = component_cls._get_memory_cell_count(
- detector_sources[0])
+ detector_sources[0]
+ )
caldata.pixels_x = component_cls.module_shape[1]
caldata.pixels_y = component_cls.module_shape[0]
@@ -625,28 +733,35 @@ class CalibrationData:
def _simplify_parameter_name(name):
"""Convert parameter names to valid Python symbols."""
- return name.lower().replace(' ', '_')
+ return name.lower().replace(" ", "_")
@staticmethod
def _determine_data_creation_date(data):
"""Determine data creation date."""
- assert data.files, 'data contains no files'
+ assert data.files, "data contains no files"
try:
- creation_date = data.files[0].metadata()['creationDate']
+ creation_date = data.files[0].metadata()["creationDate"]
except KeyError:
from warnings import warn
- warn('Last file modification time used as creation date for old '
- 'DAQ file format may be unreliable')
+
+ warn(
+ "Last file modification time used as creation date for old "
+ "DAQ file format may be unreliable"
+ )
return datetime.fromtimestamp(
- Path(data.files[0].filename).lstat().st_mtime)
+ Path(data.files[0].filename).lstat().st_mtime
+ )
else:
if not data.is_single_run:
from warnings import warn
- warn('Sample file used to determine creation date for multi '
- 'run data')
+
+ warn(
+ "Sample file used to determine creation date for multi "
+ "run data"
+ )
return creation_date
@@ -681,8 +796,9 @@ class SplitConditionCalibrationData(CalibrationData):
"""Compatibility with CalibrationData."""
# Removes likely duplicates while preserving order.
- return list(dict.fromkeys(
- self.dark_parameters + self.illuminated_parameters))
+ return list(
+ dict.fromkeys(self.dark_parameters + self.illuminated_parameters)
+ )
@property
def condition(self):
@@ -703,7 +819,10 @@ class SplitConditionCalibrationData(CalibrationData):
return self._build_condition(self.illuminated_parameters)
def metadata(
- self, calibrations=None, event_at=None, snapshot_at=None,
+ self,
+ calibrations=None,
+ event_at=None,
+ snapshot_at=None,
):
"""Query CCV metadata for calibrations, conditions and time.
@@ -723,25 +842,36 @@ class SplitConditionCalibrationData(CalibrationData):
if calibrations is None:
calibrations = (
- self.dark_calibrations | self.illuminated_calibrations)
+ self.dark_calibrations | self.illuminated_calibrations
+ )
metadata = CCVMetadata()
dark_calibrations = self.dark_calibrations & set(calibrations)
if dark_calibrations:
self._api.closest_ccv_by_time_by_condition(
- self.detector_name, dark_calibrations,
- self.dark_condition, self.modules,
- event_at or self.event_at, snapshot_at or self.snapshot_at,
- metadata)
+ self.detector_name,
+ dark_calibrations,
+ self.dark_condition,
+ self.modules,
+ event_at or self.event_at,
+ snapshot_at or self.snapshot_at,
+ metadata,
+ da_to=self.da_to,
+ )
illum_calibrations = self.illuminated_calibrations & set(calibrations)
if illum_calibrations:
self._api.closest_ccv_by_time_by_condition(
- self.detector_name, illum_calibrations,
- self.illuminated_condition, self.modules,
- event_at or self.event_at, snapshot_at or self.snapshot_at,
- metadata)
+ self.detector_name,
+ illum_calibrations,
+ self.illuminated_condition,
+ self.modules,
+ event_at or self.event_at,
+ snapshot_at or self.snapshot_at,
+ metadata,
+ da_to=self.da_to,
+ )
return metadata
@@ -749,41 +879,55 @@ class AGIPD_CalibrationData(SplitConditionCalibrationData):
"""Calibration data for the AGIPD detector."""
dark_calibrations = {
- 'Offset',
- 'Noise',
- 'ThresholdsDark',
- 'BadPixelsDark',
- 'BadPixelsPC',
- 'SlopesPC',
+ "Offset",
+ "Noise",
+ "ThresholdsDark",
+ "BadPixelsDark",
+ "BadPixelsPC",
+ "SlopesPC",
}
illuminated_calibrations = {
- 'BadPixelsFF',
- 'SlopesFF',
+ "BadPixelsFF",
+ "SlopesFF",
}
dark_parameters = [
- 'Sensor Bias Voltage',
- 'Pixels X',
- 'Pixels Y',
- 'Memory cells',
- 'Acquisition rate',
- 'Gain setting',
- 'Gain mode',
- 'Integration time',
+ "Sensor Bias Voltage",
+ "Pixels X",
+ "Pixels Y",
+ "Memory cells",
+ "Acquisition rate",
+ "Gain setting",
+ "Gain mode",
+ "Integration time",
]
- illuminated_parameters = dark_parameters + ['Source energy']
+ illuminated_parameters = dark_parameters + ["Source energy"]
def __init__(
- self, detector_name, sensor_bias_voltage,
- memory_cells, acquisition_rate,
- modules=None, client=None,
- event_at=None, snapshot_at=None,
- gain_setting=None, gain_mode=None,
- integration_time=12, source_energy=9.2,
- pixels_x=512, pixels_y=128,
+ self,
+ detector_name,
+ sensor_bias_voltage,
+ memory_cells,
+ acquisition_rate,
+ modules=None,
+ client=None,
+ event_at=None,
+ snapshot_at=None,
+ gain_setting=None,
+ gain_mode=None,
+ da_to="da",
+ integration_time=12,
+ source_energy=9.2,
+ pixels_x=512,
+ pixels_y=128,
):
super().__init__(
- detector_name, modules, client, event_at, snapshot_at,
+ detector_name,
+ modules,
+ client,
+ event_at,
+ snapshot_at,
+ da_to,
)
self.sensor_bias_voltage = sensor_bias_voltage
@@ -800,11 +944,11 @@ class AGIPD_CalibrationData(SplitConditionCalibrationData):
cond = super()._build_condition(parameters)
# Fix-up some database quirks.
- if int(cond.get('Gain mode', -1)) == 0:
- del cond['Gain mode']
+ if int(cond.get("Gain mode", -1)) == 0:
+ del cond["Gain mode"]
- if int(cond.get('Integration time', -1)) == 12:
- del cond['Integration time']
+ if int(cond.get("Integration time", -1)) == 12:
+ del cond["Integration time"]
return cond
@@ -813,36 +957,49 @@ class LPD_CalibrationData(SplitConditionCalibrationData):
"""Calibration data for the LPD detector."""
dark_calibrations = {
- 'Offset',
- 'Noise',
- 'BadPixelsDark',
+ "Offset",
+ "Noise",
+ "BadPixelsDark",
}
illuminated_calibrations = {
- 'RelativeGain',
- 'GainAmpMap',
- 'FFMap',
- 'BadPixelsFF',
+ "RelativeGain",
+ "GainAmpMap",
+ "FFMap",
+ "BadPixelsFF",
}
dark_parameters = [
- 'Sensor Bias Voltage',
- 'Memory cells',
- 'Pixels X',
- 'Pixels Y',
- 'Feedback capacitor',
+ "Sensor Bias Voltage",
+ "Memory cells",
+ "Pixels X",
+ "Pixels Y",
+ "Feedback capacitor",
]
- illuminated_parameters = dark_parameters + ['Source Energy', 'category']
+ illuminated_parameters = dark_parameters + ["Source Energy", "category"]
def __init__(
- self, detector_name, sensor_bias_voltage,
- memory_cells, feedback_capacitor=5.0,
- pixels_x=256, pixels_y=256,
- source_energy=9.2, category=1,
- modules=None, client=None,
- event_at=None, snapshot_at=None,
+ self,
+ detector_name,
+ sensor_bias_voltage,
+ memory_cells,
+ feedback_capacitor=5.0,
+ pixels_x=256,
+ pixels_y=256,
+ source_energy=9.2,
+ category=1,
+ modules=None,
+ client=None,
+ event_at=None,
+ snapshot_at=None,
+ da_to="da",
):
super().__init__(
- detector_name, modules, client, event_at, snapshot_at,
+ detector_name,
+ modules,
+ client,
+ event_at,
+ snapshot_at,
+ da_to,
)
self.sensor_bias_voltage = sensor_bias_voltage
@@ -858,31 +1015,44 @@ class DSSC_CalibrationData(CalibrationData):
"""Calibration data for the DSSC detetor."""
calibrations = {
- 'Offset',
- 'Noise',
+ "Offset",
+ "Noise",
}
parameters = [
- 'Sensor Bias Voltage',
- 'Memory cells',
- 'Pixels X',
- 'Pixels Y',
- 'Pulse id checksum',
- 'Acquisition rate',
- 'Target gain',
- 'Encoded gain',
+ "Sensor Bias Voltage",
+ "Memory cells",
+ "Pixels X",
+ "Pixels Y",
+ "Pulse id checksum",
+ "Acquisition rate",
+ "Target gain",
+ "Encoded gain",
]
def __init__(
- self, detector_name,
- sensor_bias_voltage, memory_cells,
- pulse_id_checksum=None, acquisition_rate=None,
- target_gain=None, encoded_gain=None,
- pixels_x=512, pixels_y=128,
- modules=None, client=None,
- event_at=None, snapshot_at=None,
+ self,
+ detector_name,
+ sensor_bias_voltage,
+ memory_cells,
+ pulse_id_checksum=None,
+ acquisition_rate=None,
+ target_gain=None,
+ encoded_gain=None,
+ pixels_x=512,
+ pixels_y=128,
+ modules=None,
+ client=None,
+ event_at=None,
+ snapshot_at=None,
+ da_to="da",
):
super().__init__(
- detector_name, modules, client, event_at, snapshot_at,
+ detector_name,
+ modules,
+ client,
+ event_at,
+ snapshot_at,
+ da_to,
)
self.sensor_bias_voltage = sensor_bias_voltage
@@ -899,33 +1069,46 @@ class JUNGFRAU_CalibrationData(CalibrationData):
"""Calibration data for the JUNGFRAU detector."""
calibrations = {
- 'Offset10Hz',
- 'Noise10Hz',
- 'BadPixelsDark10Hz',
- 'RelativeGain10Hz',
- 'BadPixelsFF10Hz',
+ "Offset10Hz",
+ "Noise10Hz",
+ "BadPixelsDark10Hz",
+ "RelativeGain10Hz",
+ "BadPixelsFF10Hz",
}
parameters = [
- 'Sensor Bias Voltage',
- 'Memory Cells',
- 'Pixels X',
- 'Pixels Y',
- 'Integration Time',
- 'Sensor temperature',
- 'Gain Setting',
+ "Sensor Bias Voltage",
+ "Memory Cells",
+ "Pixels X",
+ "Pixels Y",
+ "Integration Time",
+ "Sensor temperature",
+ "Gain Setting",
]
def __init__(
- self, detector_name, sensor_bias_voltage,
- memory_cells, integration_time,
+ self,
+ detector_name,
+ sensor_bias_voltage,
+ memory_cells,
+ integration_time,
gain_setting,
- gain_mode=None, sensor_temperature=291,
- pixels_x=1024, pixels_y=512,
- modules=None, client=None,
- event_at=None, snapshot_at=None,
+ gain_mode=None,
+ sensor_temperature=291,
+ pixels_x=1024,
+ pixels_y=512,
+ modules=None,
+ client=None,
+ event_at=None,
+ snapshot_at=None,
+ da_to="da",
):
super().__init__(
- detector_name, modules, client, event_at, snapshot_at,
+ detector_name,
+ modules,
+ client,
+ event_at,
+ snapshot_at,
+ da_to,
)
self.sensor_bias_voltage = sensor_bias_voltage
@@ -941,40 +1124,52 @@ class JUNGFRAU_CalibrationData(CalibrationData):
cond = super()._build_condition(parameters)
# Fix-up some database quirks.
- if int(cond.get('Gain mode', -1)) == 0:
- del cond['Gain mode']
+ if int(cond.get("Gain mode", -1)) == 0:
+ del cond["Gain mode"]
return cond
class PNCCD_CalibrationData(CalibrationData):
calibrations = {
- 'OffsetCCD',
- 'BadPixelsDarkCCD',
- 'NoiseCCD',
- 'RelativeGainCCD',
- 'CTECCD',
+ "OffsetCCD",
+ "BadPixelsDarkCCD",
+ "NoiseCCD",
+ "RelativeGainCCD",
+ "CTECCD",
}
parameters = [
- 'Sensor Bias Voltage',
- 'Memory cells',
- 'Pixels X',
- 'Pixels Y',
- 'Integration Time',
- 'Sensor Temperature',
- 'Gain Setting',
+ "Sensor Bias Voltage",
+ "Memory cells",
+ "Pixels X",
+ "Pixels Y",
+ "Integration Time",
+ "Sensor Temperature",
+ "Gain Setting",
]
def __init__(
- self, detector_name, sensor_bias_voltage,
- integration_time, sensor_temperature,
- gain_setting, pixels_x=1024,
- pixels_y=1024, client=None,
- event_at=None, snapshot_at=None,
+ self,
+ detector_name,
+ sensor_bias_voltage,
+ integration_time,
+ sensor_temperature,
+ gain_setting,
+ pixels_x=1024,
+ pixels_y=1024,
+ client=None,
+ event_at=None,
+ snapshot_at=None,
+ da_to="da",
):
# Ignore modules for this detector.
super().__init__(
- detector_name, None, client, event_at, snapshot_at,
+ detector_name,
+ None,
+ client,
+ event_at,
+ snapshot_at,
+ da_to,
)
self.sensor_bias_voltage = sensor_bias_voltage
@@ -988,36 +1183,48 @@ class PNCCD_CalibrationData(CalibrationData):
class EPIX100_CalibrationData(SplitConditionCalibrationData):
dark_calibrations = {
- 'OffsetEPix100',
- 'NoiseEPix100',
- 'BadPixelsDarkEPix100',
- }
+ "OffsetEPix100",
+ "NoiseEPix100",
+ "BadPixelsDarkEPix100",
+ }
illuminated_calibrations = {
- 'RelativeGainEPix100',
+ "RelativeGainEPix100",
# 'BadPixelsFFEPix100',
}
dark_parameters = [
- 'Sensor Bias Voltage',
- 'Memory cells',
- 'Pixels X',
- 'Pixels Y',
- 'Integration time',
- 'Sensor temperature',
- 'In vacuum',
- ]
- illuminated_parameters = dark_parameters + ['Source energy']
+ "Sensor Bias Voltage",
+ "Memory cells",
+ "Pixels X",
+ "Pixels Y",
+ "Integration time",
+ "Sensor temperature",
+ "In vacuum",
+ ]
+ illuminated_parameters = dark_parameters + ["Source energy"]
def __init__(
- self, detector_name,
- sensor_bias_voltage, integration_time,
- in_vacuum=0, sensor_temperature=288,
- pixels_x=708, pixels_y=768,
- source_energy=9.2, client=None,
- event_at=None, snapshot_at=None,
+ self,
+ detector_name,
+ sensor_bias_voltage,
+ integration_time,
+ in_vacuum=0,
+ sensor_temperature=288,
+ pixels_x=708,
+ pixels_y=768,
+ source_energy=9.2,
+ client=None,
+ event_at=None,
+ snapshot_at=None,
+ da_to="da",
):
# Ignore modules for this detector.
super().__init__(
- detector_name, None, client, event_at, snapshot_at,
+ detector_name,
+ None,
+ client,
+ event_at,
+ snapshot_at,
+ da_to,
)
self.sensor_bias_voltage = sensor_bias_voltage
@@ -1032,31 +1239,41 @@ class EPIX100_CalibrationData(SplitConditionCalibrationData):
class GOTTHARD2_CalibrationData(CalibrationData):
calibrations = {
- 'LUTGotthard2'
- 'OffsetGotthard2',
- 'NoiseGotthard2',
- 'BadPixelsDarkGotthard2',
- 'RelativeGainGotthard2',
- 'BadPixelsFFGotthard2',
- }
+ "LUTGotthard2" "OffsetGotthard2",
+ "NoiseGotthard2",
+ "BadPixelsDarkGotthard2",
+ "RelativeGainGotthard2",
+ "BadPixelsFFGotthard2",
+ }
parameters = [
- 'Sensor Bias Voltage',
- 'Exposure time',
- 'Exposure period',
- 'Acquisition rate',
- 'Single photon',
- ]
+ "Sensor Bias Voltage",
+ "Exposure time",
+ "Exposure period",
+ "Acquisition rate",
+ "Single photon",
+ ]
def __init__(
- self, detector_name,
- sensor_bias_voltage, exposure_time,
- exposure_period, acquisition_rate,
- single_photon, client=None,
- event_at=None, snapshot_at=None,
+ self,
+ detector_name,
+ sensor_bias_voltage,
+ exposure_time,
+ exposure_period,
+ acquisition_rate,
+ single_photon,
+ client=None,
+ event_at=None,
+ snapshot_at=None,
+ da_to="da",
):
# Ignore modules for this detector.
super().__init__(
- detector_name, None, client, event_at, snapshot_at,
+ detector_name,
+ None,
+ client,
+ event_at,
+ snapshot_at,
+ da_to,
)
self.sensor_bias_voltage = sensor_bias_voltage