calcat_interface
Interfaces to calibration constant data.
AGIPD_CalibrationData
¶
Bases: SplitConditionCalibrationData
Calibration data for the AGIPD detector.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
class AGIPD_CalibrationData(SplitConditionCalibrationData):
"""Calibration data for the AGIPD detector."""
dark_calibrations = {
"Offset",
"Noise",
"ThresholdsDark",
"BadPixelsDark",
"BadPixelsPC",
"SlopesPC",
"SlopesCS",
"BadPixelsCS",
}
illuminated_calibrations = {
"BadPixelsFF",
"SlopesFF",
}
dark_parameters = [
"Sensor Bias Voltage",
"Pixels X",
"Pixels Y",
"Memory cells",
"Acquisition rate",
"Gain setting",
"Gain mode",
"Integration time",
]
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,
gain_setting=None,
gain_mode=None,
module_naming="da",
caldb_root=None,
integration_time=12,
source_energy=9.2,
pixels_x=512,
pixels_y=128,
):
super().__init__(
detector_name,
modules,
client,
event_at,
module_naming,
caldb_root,
)
self.sensor_bias_voltage = sensor_bias_voltage
self.memory_cells = memory_cells
self.pixels_x = pixels_x
self.pixels_y = pixels_y
self.acquisition_rate = acquisition_rate
self.gain_setting = gain_setting
self.gain_mode = gain_mode
self.integration_time = integration_time
self.source_energy = source_energy
def _build_condition(self, parameters):
cond = super()._build_condition(parameters)
# Fix-up some database quirks.
if cond.get("Gain mode", None):
cond["Gain mode"] = 1
else:
cond.pop("Gain mode", None)
if cond.get("Integration time", None) == 12:
del cond["Integration time"]
return cond
CCVMetadata
¶
Bases: dict
Dictionary for CCV metadata.
Identical to a regular dict, but with a custom pandas-based string representation to be easier to read.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
class CCVMetadata(dict):
"""Dictionary for CCV metadata.
Identical to a regular dict, but with a custom pandas-based
string representation to be easier to read.
"""
def __str__(self):
"""Pretty-print CCV metadata using pandas."""
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()
}
return str(pd.DataFrame.from_dict(res, orient="index"))
__str__()
¶
Pretty-print CCV metadata using pandas.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
CalCatApi
¶
Internal calibration_client wrapper.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
class CalCatApi(metaclass=ClientWrapper):
"""Internal calibration_client wrapper."""
get_detector_keys = [
"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",
]
def __init__(self, client):
self.client = client
@classmethod
def format_time(cls, dt):
"""Parse different ways to specify time to CalCat."""
if isinstance(dt, datetime):
return dt.astimezone(timezone.utc).isoformat()
elif isinstance(dt, date):
return cls.format_time(datetime.combine(dt, time()))
return dt
def format_cond(self, condition):
"""Encode operating condition to CalCat API format.
Args:
condition (dict): Mapping of parameter DB name to value
Returns:
(dict) Operating condition for use in CalCat API.
"""
return {
"parameters_conditions_attributes": [
{"parameter_name": 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)
if not resp_detector["success"]:
raise CalCatError(resp_detector)
return {k: resp_detector["data"][k] for k in self.get_detector_keys}
@lru_cache()
def physical_detector_units(
self,
detector_id,
pdu_snapshot_at,
module_naming="da"
):
"""Physical detector unit metadata."""
resp_pdus = PhysicalDetectorUnit.get_all_by_detector(
self.client, detector_id, self.format_time(pdu_snapshot_at)
)
if not resp_pdus["success"]:
raise CalCatError(resp_pdus)
# Create dict based on requested keys: karabo_da, module number,
# or QxMx naming convention.
if module_naming == "da":
return {
pdu["karabo_da"]: {k: pdu[k] for k in self.get_pdu_keys}
for pdu in resp_pdus["data"]
}
elif module_naming == "modno":
return {
int(pdu["karabo_da"][-2:]): {
k: pdu[k] for k in self.get_pdu_keys
}
for pdu in resp_pdus["data"]
}
elif module_naming == "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"{module_naming} is unknown!. Expected da, modno, or qm")
@lru_cache()
def calibration_id(self, calibration_name):
"""ID for a calibration in CalCat."""
resp_calibration = Calibration.get_by_name(
self.client, calibration_name
)
if not resp_calibration["success"]:
raise CalCatError(resp_calibration)
return resp_calibration["data"]["id"]
@lru_cache()
def calibration_name(self, calibration_id):
"""Name for a calibration in CalCat."""
resp_calibration = Calibration.get_by_id(
self.client, calibration_id
)
if not resp_calibration["success"]:
raise CalCatError(resp_calibration)
return resp_calibration["data"]["name"]
@lru_cache()
def parameter_id(self, param_name):
"""ID for an operating condition parameter in CalCat."""
resp_parameter = Parameter.get_by_name(self.client, param_name)
if not resp_parameter["success"]:
raise CalCatError(resp_parameter)
return resp_parameter["data"]["id"]
def _closest_ccv_by_time_by_condition(
self,
detector_name: str,
calibration_ids: Sequence[int],
condition: dict,
karabo_da: Optional[str] = None,
event_at=None,
pdu_snapshot_at=None,
):
resp = CalibrationConstantVersion.get_closest_by_time_by_detector_conditions(
self.client,
detector_name,
calibration_ids,
self.format_cond(condition),
karabo_da=karabo_da or "",
event_at=self.format_time(event_at),
pdu_snapshot_at=self.format_time(pdu_snapshot_at),
)
if not resp["success"]:
if resp["status_code"] == 200:
# calibration_client turns empty response into an error
return []
raise CalCatError(resp)
return resp["data"]
def closest_ccv_by_time_by_condition(
self,
detector_name,
calibrations,
condition,
modules=None,
event_at=None,
pdu_snapshot_at=None,
metadata=None,
module_naming="da",
):
"""Query bulk CCV metadata from CalCat.
This method uses the /get_closest_version_by_detector API
to query matching CCVs for PDUs connected to a detector instance
in one go. In particular, it automatically includes the PDU as
an operating condition parameter to allow for a single global
condition rather than PDU-specific ones.
Args:
detector_name (str): Detector instance name.
calibrations (Iterable of str): Calibrations to query
metadata for.
condition (dict): Mapping of parameter name to value.
modules (Collection of int or None): List of module numbers
or None for all (default).
event_at (datetime, date, str or None): Time at which the
CCVs should have been valid or None for now (default).
pdu_snapshot_at (datetime, date, str or None): Time of database
state to look at or None for now (default).
metadata (dict or None): Mapping to fill for results or
None for a new dictionary (default).
module_naming (str or None): Expected module name convention to be
used as metadata dict keys. Expected values are:
`da`: data aggregator name is used. Default.
`modno`: module index is used. Index is chosen based on last 2
integers in karabo_da.
`qm`: QxMx naming convention is used. Virtual names for
AGIPD, DSSC, and LPD.
Returns:
(dict) Nested mapping of module number to calibrations to
CCV metadata. Identical to passed metadata argument if
passed.
"""
event_at = self.format_time(event_at)
pdu_snapshot_at = self.format_time(pdu_snapshot_at)
if metadata is None:
metadata = CCVMetadata()
if not calibrations:
# Make sure there are at least empty dictionaries for each
# module.
for mod in modules.values():
metadata.setdefault(mod, dict())
return metadata
# Map calibration ID to calibration name.
cal_id_map = {
self.calibration_id(calibration): calibration
for calibration in calibrations
}
calibration_ids = list(cal_id_map.keys())
# Map aggregator to the selected module name.
da_to_modname = {
data['karabo_da']: mod_name for mod_name, data in
self.physical_detector_units(
self.detector(detector_name)['id'],
pdu_snapshot_at,
module_naming=module_naming
).items()
if not modules or mod_name in modules
}
# 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(da_to_modname)) if len(da_to_modname) == 1 else '',
resp_data = self._closest_ccv_by_time_by_condition(
detector_name,
calibration_ids,
condition,
karabo_da=karabo_da,
event_at=event_at,
pdu_snapshot_at=pdu_snapshot_at,
)
for ccv in resp_data:
try:
mod = da_to_modname[ccv['physical_detector_unit']['karabo_da']]
except KeyError:
# Not included in our modules
continue
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"],
)
return metadata
calibration_id(calibration_name)
cached
¶
ID for a calibration in CalCat.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
calibration_name(calibration_id)
cached
¶
Name for a calibration in CalCat.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
closest_ccv_by_time_by_condition(detector_name, calibrations, condition, modules=None, event_at=None, pdu_snapshot_at=None, metadata=None, module_naming='da')
¶
Query bulk CCV metadata from CalCat.
This method uses the /get_closest_version_by_detector API to query matching CCVs for PDUs connected to a detector instance in one go. In particular, it automatically includes the PDU as an operating condition parameter to allow for a single global condition rather than PDU-specific ones.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
detector_name |
str
|
Detector instance name. |
required |
calibrations |
Iterable of str
|
Calibrations to query metadata for. |
required |
condition |
dict
|
Mapping of parameter name to value. |
required |
modules |
Collection of int or None
|
List of module numbers or None for all (default). |
None
|
event_at |
datetime, date, str or None
|
Time at which the CCVs should have been valid or None for now (default). |
None
|
pdu_snapshot_at |
datetime, date, str or None
|
Time of database state to look at or None for now (default). |
None
|
metadata |
dict or None
|
Mapping to fill for results or None for a new dictionary (default). |
None
|
module_naming |
str or None
|
Expected module name convention to be
used as metadata dict keys. Expected values are:
|
'da'
|
Returns:
| Type | Description |
|---|---|
(dict) Nested mapping of module number to calibrations to CCV metadata. Identical to passed metadata argument if passed. |
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
def closest_ccv_by_time_by_condition(
self,
detector_name,
calibrations,
condition,
modules=None,
event_at=None,
pdu_snapshot_at=None,
metadata=None,
module_naming="da",
):
"""Query bulk CCV metadata from CalCat.
This method uses the /get_closest_version_by_detector API
to query matching CCVs for PDUs connected to a detector instance
in one go. In particular, it automatically includes the PDU as
an operating condition parameter to allow for a single global
condition rather than PDU-specific ones.
Args:
detector_name (str): Detector instance name.
calibrations (Iterable of str): Calibrations to query
metadata for.
condition (dict): Mapping of parameter name to value.
modules (Collection of int or None): List of module numbers
or None for all (default).
event_at (datetime, date, str or None): Time at which the
CCVs should have been valid or None for now (default).
pdu_snapshot_at (datetime, date, str or None): Time of database
state to look at or None for now (default).
metadata (dict or None): Mapping to fill for results or
None for a new dictionary (default).
module_naming (str or None): Expected module name convention to be
used as metadata dict keys. Expected values are:
`da`: data aggregator name is used. Default.
`modno`: module index is used. Index is chosen based on last 2
integers in karabo_da.
`qm`: QxMx naming convention is used. Virtual names for
AGIPD, DSSC, and LPD.
Returns:
(dict) Nested mapping of module number to calibrations to
CCV metadata. Identical to passed metadata argument if
passed.
"""
event_at = self.format_time(event_at)
pdu_snapshot_at = self.format_time(pdu_snapshot_at)
if metadata is None:
metadata = CCVMetadata()
if not calibrations:
# Make sure there are at least empty dictionaries for each
# module.
for mod in modules.values():
metadata.setdefault(mod, dict())
return metadata
# Map calibration ID to calibration name.
cal_id_map = {
self.calibration_id(calibration): calibration
for calibration in calibrations
}
calibration_ids = list(cal_id_map.keys())
# Map aggregator to the selected module name.
da_to_modname = {
data['karabo_da']: mod_name for mod_name, data in
self.physical_detector_units(
self.detector(detector_name)['id'],
pdu_snapshot_at,
module_naming=module_naming
).items()
if not modules or mod_name in modules
}
# 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(da_to_modname)) if len(da_to_modname) == 1 else '',
resp_data = self._closest_ccv_by_time_by_condition(
detector_name,
calibration_ids,
condition,
karabo_da=karabo_da,
event_at=event_at,
pdu_snapshot_at=pdu_snapshot_at,
)
for ccv in resp_data:
try:
mod = da_to_modname[ccv['physical_detector_unit']['karabo_da']]
except KeyError:
# Not included in our modules
continue
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"],
)
return metadata
detector(detector_name)
cached
¶
Detector metadata.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
format_cond(condition)
¶
Encode operating condition to CalCat API format.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
condition |
dict
|
Mapping of parameter DB name to value |
required |
Returns:
| Type | Description |
|---|---|
(dict) Operating condition for use in CalCat API. |
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
def format_cond(self, condition):
"""Encode operating condition to CalCat API format.
Args:
condition (dict): Mapping of parameter DB name to value
Returns:
(dict) Operating condition for use in CalCat API.
"""
return {
"parameters_conditions_attributes": [
{"parameter_name": k, "value": str(v)}
for k, v in condition.items()
]
}
format_time(dt)
classmethod
¶
Parse different ways to specify time to CalCat.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
parameter_id(param_name)
cached
¶
ID for an operating condition parameter in CalCat.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
physical_detector_units(detector_id, pdu_snapshot_at, module_naming='da')
cached
¶
Physical detector unit metadata.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
@lru_cache()
def physical_detector_units(
self,
detector_id,
pdu_snapshot_at,
module_naming="da"
):
"""Physical detector unit metadata."""
resp_pdus = PhysicalDetectorUnit.get_all_by_detector(
self.client, detector_id, self.format_time(pdu_snapshot_at)
)
if not resp_pdus["success"]:
raise CalCatError(resp_pdus)
# Create dict based on requested keys: karabo_da, module number,
# or QxMx naming convention.
if module_naming == "da":
return {
pdu["karabo_da"]: {k: pdu[k] for k in self.get_pdu_keys}
for pdu in resp_pdus["data"]
}
elif module_naming == "modno":
return {
int(pdu["karabo_da"][-2:]): {
k: pdu[k] for k in self.get_pdu_keys
}
for pdu in resp_pdus["data"]
}
elif module_naming == "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"{module_naming} is unknown!. Expected da, modno, or qm")
CalCatError
¶
Bases: Exception
CalCat API error.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
CalibrationData
¶
Calibration constants data for detectors.
European XFEL uses a web app and database to store records about the characterization of detectors and the data necessary to their correction and analysis, collectively called CalCat. The default installation is available at https://in.xfel.eu/calibration.
A detector is identified by a name (e.g. SPB_DET_AGIPD1M-1) and consists of one or more detector modules. The modules are a virtual concept and may be identified by their number (e.g. 3), the Karabo data aggregator in EuXFEL's DAQ system they're connected to (e.g. AGIPD05) or a virtual device name describing their relative location (e.g. Q3M2).
A detector module is mapped to an actual physical detector unit (PDU), which may be changed in case of a physical replacement. When characterization data is inserted into the database, it is attached to the PDU currently mapped to a module and not the virtual module itself.
Characterization data is organized by its type just called calibration (e.g. Offset or SlopesFF) and the operating condition it was taken in, which is a mapping of parameter keys to their values (e.g. Sensor bias voltage or integration time). Any unique combination of calibration (type) and operating condition is a calibration constant (CC). Any individual measurement of a CC is called a calibration constant version (CCV). There may be many CCVs for any given CC.
Note that while an authenticated connection to CalCat is possible from anywhere, the actual calibration data referred to is only available on the European XFEL computing infrastructure. If no explicit credentials are supplied, an anonymous read-only connection is established that is also only available from there.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
class CalibrationData:
"""Calibration constants data for detectors.
European XFEL uses a web app and database to store records about the
characterization of detectors and the data necessary to their
correction and analysis, collectively called CalCat. The default
installation is available at https://in.xfel.eu/calibration.
A detector is identified by a name (e.g. SPB_DET_AGIPD1M-1) and
consists of one or more detector modules. The modules are a virtual
concept and may be identified by their number (e.g. 3), the Karabo
data aggregator in EuXFEL's DAQ system they're connected to
(e.g. AGIPD05) or a virtual device name describing their relative
location (e.g. Q3M2).
A detector module is mapped to an actual physical detector unit
(PDU), which may be changed in case of a physical replacement. When
characterization data is inserted into the database, it is attached
to the PDU currently mapped to a module and not the virtual module
itself.
Characterization data is organized by its type just called
calibration (e.g. Offset or SlopesFF) and the operating condition it
was taken in, which is a mapping of parameter keys to their values
(e.g. Sensor bias voltage or integration time). Any unique
combination of calibration (type) and operating condition is a
calibration constant (CC). Any individual measurement of a CC is
called a calibration constant version (CCV). There may be many CCVs
for any given CC.
Note that while an authenticated connection to CalCat is possible
from anywhere, the actual calibration data referred to is only
available on the European XFEL computing infrastructure. If no
explicit credentials are supplied, an anonymous read-only connection
is established that is also only available from there.
"""
calibrations = set()
default_client = None
_default_caldb_root = ...
def __init__(
self,
detector_name,
modules=None,
client=None,
event_at=None,
module_naming="da",
caldb_root=None,
):
"""Initialize a new CalibrationData object.
If no calibration-client object is passed or has been created
using Calibration.new_client, an anonymous read-only connection
is established automatically.
Args:
detector_name (str): Name of detector in CalCat.
modules (Iterable of int, optional): Module numbers to
query for or None for all available (default).
client (CalibrationClient, optional): Client for CalCat
communication, global one by default.
event_at (datetime, date, str or None): Default time at which the
CCVs should have been valid, now if omitted
module_naming (str or None): Expected module name convention to be
used as metadata dict keys. Expected values are:
`da`: data aggregator name is used. Default.
`modno`: module index is used. Index is chosen based on last 2
integers in karabo_da.
`qm`: QxMx naming convention is used. Virtual names for
AGIPD, DSSC, and LPD.
caldb_root (str or None): Path to the root directory for caldb
files, finds folder for production caldb by default.
**condition_params: Operating condition parameters defined
on an instance level.
"""
self.detector_name = detector_name
self.modules = modules
self.event_at = event_at
self.pdu_snapshot_at = event_at
self.module_naming = module_naming
if caldb_root is None:
self.caldb_root = self._get_default_caldb_root()
else:
self.caldb_root = Path(caldb_root)
if client is None:
client = (
self.__class__.default_client
or self.__class__.new_anonymous_client()
)
self._api = CalCatApi(client)
@staticmethod
def new_anonymous_client():
"""Create an anonymous calibration-client object.
This connection allows read-only access to CalCat using a
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"
)
return CalibrationData.new_client(
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,
):
"""Create a new calibration-client object.
The client object is saved as a class property and is
automatically to any future CalibrationData objects created, if
no other client is passed explicitly.
Arguments:
client_id (str): Client ID.
client_secret (str): Client secret.
user_email (str): LDAP user email.
installation (str, optional): Prefix for CalCat
installation, production system by default.
base_url (str, optional): URL template for CalCat
installation, public European XFEL by default.
Any further keyword arguments are passed on to
CalibrationClient.__init__().
Returns:
(CalibrationClient) CalCat client.
"""
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
# detector-specific implementations.
CalibrationData.default_client = CalibrationClient(
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,
)
return CalibrationData.default_client
@staticmethod
def _get_default_caldb_root():
if CalibrationData._default_caldb_root is ...:
onc_path = Path("/common/cal/caldb_store")
maxwell_path = Path("/gpfs/exfel/d/cal/caldb_store")
if onc_path.is_dir():
CalibrationData._default_caldb_root = onc_path
elif maxwell_path.is_dir():
CalibrationData._default_caldb_root = maxwell_path
else:
CalibrationData._default_caldb_root = None
return CalibrationData._default_caldb_root
@property
def client(self):
return self._api.client
@property
def detector(self):
return self._api.detector(self.detector_name)
@property
def physical_detector_units(self):
return self._api.physical_detector_units(
self.detector["id"], self.pdu_snapshot_at, self.module_naming
)
@property
def mod_to_pdu(self):
"""Get the physical detector units and create a dictionary
mapping each module name to physical name (physical detector unit).
Returns:
DICT: mapping module to physical detector unit name.
"""
return {
mod: pdu_md["physical_name"] for mod, pdu_md in self._api.physical_detector_units( # noqa
self.detector["id"],
self.pdu_snapshot_at,
self.module_naming,
).items()
}
@property
def condition(self):
return self._build_condition(self.parameters)
def replace(self, **new_kwargs):
"""Create a new CalibrationData object with altered values."""
keys = {
"detector_name",
"modules",
"client",
"event_at",
"pdu_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)
return self.__class__(**kwargs)
def metadata(
self,
calibrations=None,
event_at=None,
pdu_snapshot_at=None,
):
"""Query CCV metadata for calibrations, conditions and time.
Args:
calibrations (Iterable of str, optional): Calibrations to
query metadata for, may be None to retrieve all.
event_at (datetime, date, str or None): Time at which the
CCVs should have been valid, now or default value passed at
initialization time if omitted.
pdu_snapshot_at (datetime, date, str or None): Time of database
state to look at, now or default value passed at
initialization time if omitted.
Returns:
(CCVMetadata) CCV metadata result.
"""
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,
pdu_snapshot_at or self.pdu_snapshot_at,
metadata,
module_naming=self.module_naming,
)
return metadata
def ndarray(
self,
module,
calibration,
metadata=None,
):
"""Load CCV data as ndarray.
Args:
module (int): Module number
calibration (str): Calibration constant.
metadata (CCVMetadata, optional): CCV metadata to load
constant data for, may be None to query metadata.
Returns:
(ndarray): CCV data
"""
import numpy as np
if self.caldb_root is None:
raise RuntimeError("calibration database store unavailable")
if self.modules and module not in self.modules:
raise ValueError("module not part of this calibration data")
if metadata is None:
metadata = self.metadata([calibration])
row = metadata[module][calibration]
with h5py.File(self.caldb_root / row['path'], 'r') as f:
return np.asarray(f[row['dataset'] + '/data'])
def _allocate_constant_arrays(self, metadata, const_load_mp, const_data):
for mod, ccv_entry in metadata.items():
const_data[mod] = {}
for cname, mdata in ccv_entry.items():
dataset = mdata["dataset"]
with h5py.File(self.caldb_root / mdata["path"], "r") as cf:
shape = cf[f"{dataset}/data"].shape
dtype = cf[f"{dataset}/data"].dtype
const_data[mod][cname] = const_load_mp.alloc(
shape=shape, dtype=dtype
)
def load_constants_from_metadata(self, metadata):
"""Load the data for all constants in metadata object.
Args:
metadata (CCVMetadata, optional): CCV metadata to load
constant data for, may be None to query metadata.
Returns:
(Dict): A dictionary of constant data.
{module: {calibration: ndarray}}.
"""
def _load_constant_dataset(wid, index, mod):
"""Load constant dataset from the CCVMetadata `metadata` into
a shared allocated array.
Args:
mod (str): module key in `metadata` object
"""
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 = dict()
const_load_mp = psh.ProcessContext(num_workers=24)
self._allocate_constant_arrays(metadata, const_load_mp, const_data)
const_load_mp.map(_load_constant_dataset, list(metadata.keys()))
return const_data
def ndarray_map(
self,
calibrations=None,
metadata=None,
):
"""Load all CCV data in a nested map of ndarrays.
Args:
calibrations (Iterable of str, optional): Calibration constants
or None for all available (default).
metadata (CCVMetadata, optional): CCV metadata to load constant
for or None to query metadata automatically (default).
Returns:
(dict of dict of ndarray): CCV data by module number and
calibration constant name.
{module: {calibration: ndarray}}
"""
if self.caldb_root is None:
raise RuntimeError("calibration database store unavailable")
if metadata is None:
metadata = self.metadata(calibrations)
return self.load_constants_from_metadata(metadata)
def display_markdown_retrieved_constants(
self,
metadata=None,
ccvs_url="https://in.xfel.eu/calibration/calibration_constant_versions/" # noqa
):
"""
Display markdown table with reference links for the
retrieved constants. Tables are split into groups of a
maximum of 4 modules.
Args:
metadata (dict, optional): Metadata for calibration constants.
Defaults to None.
ccvs_url (str, optional): URL for calibration constant versions.
Defaults to
"https://in.xfel.eu/calibration/calibration_constant_versions/".
"""
from IPython.display import Markdown, display
from tabulate import tabulate
if metadata is None:
metadata = self.metadata()
calibrations = set()
# Get all calibrations available in the metadata for all modules.
for c in list(metadata.values()):
calibrations |= c.keys()
cal_groups = [
list(calibrations)[x:x+4] for x in range(0, len(calibrations), 4)]
# Loop over groups of calibrations.
for cal_group in cal_groups:
table = [["Modules"] + cal_group]
# Loop over calibrations and modules to form the next rows.
for mod in metadata:
mod_consts = []
for cname in cal_group:
c_mdata = metadata[mod].get(cname)
# A calibration that is available in given metadata.
if c_mdata is not None:
# Have the creation time a reference
# link to the CCV on CALCAT.
c_time = datetime.fromisoformat(
c_mdata["begin_validity_at"]).strftime(
"%Y-%m-%d %H:%M")
mod_consts.append(
f"[{c_time}]({ccvs_url}/{c_mdata['ccv_id']})")
else:
# Constant is not available for this module.
mod_consts.append("___")
table.append([mod] + mod_consts)
display(
Markdown(
tabulate(
table,
tablefmt="pipe",
headers="firstrow",
)
)
)
def _build_condition(self, parameters):
cond = dict()
for db_name in parameters:
value = getattr(self, self._simplify_parameter_name(db_name), None)
if value is not None:
cond[db_name] = value
return cond
@classmethod
def _from_multimod_detector_data(
cls,
component_cls,
data,
detector,
modules,
client,
):
if isinstance(detector, component_cls):
detector_name = detector.detector_name
elif detector is None:
detector_name = component_cls._find_detector_name(data)
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"
)
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:
modules = sorted(source_to_modno.values())
creation_date = cls._determine_data_creation_date(data)
# Create new CalibrationData object.
caldata = cls(
detector_name,
modules,
client,
creation_date,
creation_date,
)
caldata.memory_cells = component_cls._get_memory_cell_count(
detector_sources[0]
)
caldata.pixels_x = component_cls.module_shape[1]
caldata.pixels_y = component_cls.module_shape[0]
return caldata, detector_sources
@staticmethod
def _simplify_parameter_name(name):
"""Convert parameter names to valid Python symbols."""
return name.lower().replace(" ", "_")
@staticmethod
def _determine_data_creation_date(data):
"""Determine data creation date."""
assert data.files, "data contains no files"
try:
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"
)
return datetime.fromtimestamp(
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"
)
return creation_date
mod_to_pdu
property
¶
Get the physical detector units and create a dictionary mapping each module name to physical name (physical detector unit).
Returns:
| Name | Type | Description |
|---|---|---|
DICT | mapping module to physical detector unit name. |
__init__(detector_name, modules=None, client=None, event_at=None, module_naming='da', caldb_root=None)
¶
Initialize a new CalibrationData object.
If no calibration-client object is passed or has been created using Calibration.new_client, an anonymous read-only connection is established automatically.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
detector_name |
str
|
Name of detector in CalCat. |
required |
modules |
Iterable of int
|
Module numbers to query for or None for all available (default). |
None
|
client |
CalibrationClient
|
Client for CalCat communication, global one by default. |
None
|
event_at |
datetime, date, str or None
|
Default time at which the CCVs should have been valid, now if omitted |
None
|
module_naming |
str or None
|
Expected module name convention to be
used as metadata dict keys. Expected values are:
|
'da'
|
caldb_root |
str or None
|
Path to the root directory for caldb files, finds folder for production caldb by default. |
None
|
**condition_params |
Operating condition parameters defined on an instance level. |
required |
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
def __init__(
self,
detector_name,
modules=None,
client=None,
event_at=None,
module_naming="da",
caldb_root=None,
):
"""Initialize a new CalibrationData object.
If no calibration-client object is passed or has been created
using Calibration.new_client, an anonymous read-only connection
is established automatically.
Args:
detector_name (str): Name of detector in CalCat.
modules (Iterable of int, optional): Module numbers to
query for or None for all available (default).
client (CalibrationClient, optional): Client for CalCat
communication, global one by default.
event_at (datetime, date, str or None): Default time at which the
CCVs should have been valid, now if omitted
module_naming (str or None): Expected module name convention to be
used as metadata dict keys. Expected values are:
`da`: data aggregator name is used. Default.
`modno`: module index is used. Index is chosen based on last 2
integers in karabo_da.
`qm`: QxMx naming convention is used. Virtual names for
AGIPD, DSSC, and LPD.
caldb_root (str or None): Path to the root directory for caldb
files, finds folder for production caldb by default.
**condition_params: Operating condition parameters defined
on an instance level.
"""
self.detector_name = detector_name
self.modules = modules
self.event_at = event_at
self.pdu_snapshot_at = event_at
self.module_naming = module_naming
if caldb_root is None:
self.caldb_root = self._get_default_caldb_root()
else:
self.caldb_root = Path(caldb_root)
if client is None:
client = (
self.__class__.default_client
or self.__class__.new_anonymous_client()
)
self._api = CalCatApi(client)
display_markdown_retrieved_constants(metadata=None, ccvs_url='https://in.xfel.eu/calibration/calibration_constant_versions/')
¶
Display markdown table with reference links for the retrieved constants. Tables are split into groups of a maximum of 4 modules.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
metadata |
dict
|
Metadata for calibration constants. Defaults to None. |
None
|
ccvs_url |
str
|
URL for calibration constant versions. Defaults to "https://in.xfel.eu/calibration/calibration_constant_versions/". |
'https://in.xfel.eu/calibration/calibration_constant_versions/'
|
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
def display_markdown_retrieved_constants(
self,
metadata=None,
ccvs_url="https://in.xfel.eu/calibration/calibration_constant_versions/" # noqa
):
"""
Display markdown table with reference links for the
retrieved constants. Tables are split into groups of a
maximum of 4 modules.
Args:
metadata (dict, optional): Metadata for calibration constants.
Defaults to None.
ccvs_url (str, optional): URL for calibration constant versions.
Defaults to
"https://in.xfel.eu/calibration/calibration_constant_versions/".
"""
from IPython.display import Markdown, display
from tabulate import tabulate
if metadata is None:
metadata = self.metadata()
calibrations = set()
# Get all calibrations available in the metadata for all modules.
for c in list(metadata.values()):
calibrations |= c.keys()
cal_groups = [
list(calibrations)[x:x+4] for x in range(0, len(calibrations), 4)]
# Loop over groups of calibrations.
for cal_group in cal_groups:
table = [["Modules"] + cal_group]
# Loop over calibrations and modules to form the next rows.
for mod in metadata:
mod_consts = []
for cname in cal_group:
c_mdata = metadata[mod].get(cname)
# A calibration that is available in given metadata.
if c_mdata is not None:
# Have the creation time a reference
# link to the CCV on CALCAT.
c_time = datetime.fromisoformat(
c_mdata["begin_validity_at"]).strftime(
"%Y-%m-%d %H:%M")
mod_consts.append(
f"[{c_time}]({ccvs_url}/{c_mdata['ccv_id']})")
else:
# Constant is not available for this module.
mod_consts.append("___")
table.append([mod] + mod_consts)
display(
Markdown(
tabulate(
table,
tablefmt="pipe",
headers="firstrow",
)
)
)
load_constants_from_metadata(metadata)
¶
Load the data for all constants in metadata object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
metadata |
CCVMetadata
|
CCV metadata to load constant data for, may be None to query metadata. |
required |
Returns:
| Type | Description |
|---|---|
Dict
|
A dictionary of constant data. {module: {calibration: ndarray}}. |
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
def load_constants_from_metadata(self, metadata):
"""Load the data for all constants in metadata object.
Args:
metadata (CCVMetadata, optional): CCV metadata to load
constant data for, may be None to query metadata.
Returns:
(Dict): A dictionary of constant data.
{module: {calibration: ndarray}}.
"""
def _load_constant_dataset(wid, index, mod):
"""Load constant dataset from the CCVMetadata `metadata` into
a shared allocated array.
Args:
mod (str): module key in `metadata` object
"""
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 = dict()
const_load_mp = psh.ProcessContext(num_workers=24)
self._allocate_constant_arrays(metadata, const_load_mp, const_data)
const_load_mp.map(_load_constant_dataset, list(metadata.keys()))
return const_data
metadata(calibrations=None, event_at=None, pdu_snapshot_at=None)
¶
Query CCV metadata for calibrations, conditions and time.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
calibrations |
Iterable of str
|
Calibrations to query metadata for, may be None to retrieve all. |
None
|
event_at |
datetime, date, str or None
|
Time at which the CCVs should have been valid, now or default value passed at initialization time if omitted. |
None
|
pdu_snapshot_at |
datetime, date, str or None
|
Time of database state to look at, now or default value passed at initialization time if omitted. |
None
|
Returns:
| Type | Description |
|---|---|
(CCVMetadata) CCV metadata result. |
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
def metadata(
self,
calibrations=None,
event_at=None,
pdu_snapshot_at=None,
):
"""Query CCV metadata for calibrations, conditions and time.
Args:
calibrations (Iterable of str, optional): Calibrations to
query metadata for, may be None to retrieve all.
event_at (datetime, date, str or None): Time at which the
CCVs should have been valid, now or default value passed at
initialization time if omitted.
pdu_snapshot_at (datetime, date, str or None): Time of database
state to look at, now or default value passed at
initialization time if omitted.
Returns:
(CCVMetadata) CCV metadata result.
"""
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,
pdu_snapshot_at or self.pdu_snapshot_at,
metadata,
module_naming=self.module_naming,
)
return metadata
ndarray(module, calibration, metadata=None)
¶
Load CCV data as ndarray.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
module |
int
|
Module number |
required |
calibration |
str
|
Calibration constant. |
required |
metadata |
CCVMetadata
|
CCV metadata to load constant data for, may be None to query metadata. |
None
|
Returns:
| Type | Description |
|---|---|
ndarray
|
CCV data |
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
def ndarray(
self,
module,
calibration,
metadata=None,
):
"""Load CCV data as ndarray.
Args:
module (int): Module number
calibration (str): Calibration constant.
metadata (CCVMetadata, optional): CCV metadata to load
constant data for, may be None to query metadata.
Returns:
(ndarray): CCV data
"""
import numpy as np
if self.caldb_root is None:
raise RuntimeError("calibration database store unavailable")
if self.modules and module not in self.modules:
raise ValueError("module not part of this calibration data")
if metadata is None:
metadata = self.metadata([calibration])
row = metadata[module][calibration]
with h5py.File(self.caldb_root / row['path'], 'r') as f:
return np.asarray(f[row['dataset'] + '/data'])
ndarray_map(calibrations=None, metadata=None)
¶
Load all CCV data in a nested map of ndarrays.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
calibrations |
Iterable of str
|
Calibration constants or None for all available (default). |
None
|
metadata |
CCVMetadata
|
CCV metadata to load constant for or None to query metadata automatically (default). |
None
|
Returns:
| Type | Description |
|---|---|
dict of dict of ndarray
|
CCV data by module number and calibration constant name. {module: {calibration: ndarray}} |
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
def ndarray_map(
self,
calibrations=None,
metadata=None,
):
"""Load all CCV data in a nested map of ndarrays.
Args:
calibrations (Iterable of str, optional): Calibration constants
or None for all available (default).
metadata (CCVMetadata, optional): CCV metadata to load constant
for or None to query metadata automatically (default).
Returns:
(dict of dict of ndarray): CCV data by module number and
calibration constant name.
{module: {calibration: ndarray}}
"""
if self.caldb_root is None:
raise RuntimeError("calibration database store unavailable")
if metadata is None:
metadata = self.metadata(calibrations)
return self.load_constants_from_metadata(metadata)
new_anonymous_client()
staticmethod
¶
Create an anonymous calibration-client object.
This connection allows read-only access to CalCat using a facility-provided OAuth reverse proxy. This is only accessible on the European XFEL computing infrastructure.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
@staticmethod
def new_anonymous_client():
"""Create an anonymous calibration-client object.
This connection allows read-only access to CalCat using a
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"
)
return CalibrationData.new_client(
None,
None,
None,
use_oauth2=False,
base_url="http://exflcalproxy:8080/",
)
new_client(client_id, client_secret, user_email, installation='', base_url='https://in.xfel.eu/{}calibration', **kwargs)
staticmethod
¶
Create a new calibration-client object.
The client object is saved as a class property and is automatically to any future CalibrationData objects created, if no other client is passed explicitly.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
client_id |
str
|
Client ID. |
required |
client_secret |
str
|
Client secret. |
required |
user_email |
str
|
LDAP user email. |
required |
installation |
str
|
Prefix for CalCat installation, production system by default. |
''
|
base_url |
str
|
URL template for CalCat installation, public European XFEL by default. |
'https://in.xfel.eu/{}calibration'
|
Returns:
| Type | Description |
|---|---|
(CalibrationClient) CalCat client. |
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
@staticmethod
def new_client(
client_id,
client_secret,
user_email,
installation="",
base_url="https://in.xfel.eu/{}calibration",
**kwargs,
):
"""Create a new calibration-client object.
The client object is saved as a class property and is
automatically to any future CalibrationData objects created, if
no other client is passed explicitly.
Arguments:
client_id (str): Client ID.
client_secret (str): Client secret.
user_email (str): LDAP user email.
installation (str, optional): Prefix for CalCat
installation, production system by default.
base_url (str, optional): URL template for CalCat
installation, public European XFEL by default.
Any further keyword arguments are passed on to
CalibrationClient.__init__().
Returns:
(CalibrationClient) CalCat client.
"""
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
# detector-specific implementations.
CalibrationData.default_client = CalibrationClient(
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,
)
return CalibrationData.default_client
replace(**new_kwargs)
¶
Create a new CalibrationData object with altered values.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
def replace(self, **new_kwargs):
"""Create a new CalibrationData object with altered values."""
keys = {
"detector_name",
"modules",
"client",
"event_at",
"pdu_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)
return self.__class__(**kwargs)
ClientWrapper
¶
Bases: type
Metaclass to wrap each calibration_client exactly once.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
DSSC_CalibrationData
¶
Bases: CalibrationData
Calibration data for the DSSC detetor.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
class DSSC_CalibrationData(CalibrationData):
"""Calibration data for the DSSC detetor."""
calibrations = {
"Offset",
"Noise",
}
parameters = [
"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,
module_naming="da",
caldb_root=None,
):
super().__init__(
detector_name,
modules,
client,
event_at,
module_naming,
caldb_root,
)
self.sensor_bias_voltage = sensor_bias_voltage
self.memory_cells = memory_cells
self.pixels_x = pixels_x
self.pixels_y = pixels_y
self.pulse_id_checksum = pulse_id_checksum
self.acquisition_rate = acquisition_rate
self.target_gain = target_gain
self.encoded_gain = encoded_gain
EPIX100_CalibrationData
¶
Bases: SplitConditionCalibrationData
Calibration data for the ePix100 detector.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
class EPIX100_CalibrationData(SplitConditionCalibrationData):
"""Calibration data for the ePix100 detector."""
dark_calibrations = {
"OffsetEPix100",
"NoiseEPix100",
"BadPixelsDarkEPix100",
}
illuminated_calibrations = {
"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"]
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,
modules=None,
client=None,
event_at=None,
module_naming="da",
caldb_root=None,
):
# Ignore modules for this detector.
super().__init__(
detector_name,
modules,
client,
event_at,
module_naming,
caldb_root,
)
self.sensor_bias_voltage = sensor_bias_voltage
self.integration_time = integration_time
self.memory_cells = 1 # Ignore memory_cells for this detector
self.pixels_x = pixels_x
self.pixels_y = pixels_y
self.in_vacuum = in_vacuum
self.sensor_temperature = sensor_temperature
self.source_energy = source_energy
GOTTHARD2_CalibrationData
¶
Bases: CalibrationData
Calibration data for the Gotthard II detector.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
class GOTTHARD2_CalibrationData(CalibrationData):
"""Calibration data for the Gotthard II detector."""
calibrations = {
"LUTGotthard2",
"OffsetGotthard2",
"NoiseGotthard2",
"BadPixelsDarkGotthard2",
"RelativeGainGotthard2",
"BadPixelsFFGotthard2",
}
parameters = [
"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,
modules=None,
client=None,
event_at=None,
module_naming="da",
caldb_root=None,
):
# Ignore modules for this detector.
super().__init__(
detector_name,
modules,
client,
event_at,
module_naming,
caldb_root,
)
self.sensor_bias_voltage = sensor_bias_voltage
self.exposure_time = exposure_time
self.exposure_period = exposure_period
self.acquisition_rate = acquisition_rate
self.single_photon = single_photon
JUNGFRAU_CalibrationData
¶
Bases: CalibrationData
Calibration data for the JUNGFRAU detector.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
class JUNGFRAU_CalibrationData(CalibrationData):
"""Calibration data for the JUNGFRAU detector."""
calibrations = {
"Offset10Hz",
"Noise10Hz",
"BadPixelsDark10Hz",
"RelativeGain10Hz",
"BadPixelsFF10Hz",
}
parameters = [
"Sensor Bias Voltage",
"Memory Cells",
"Pixels X",
"Pixels Y",
"Integration Time",
"Sensor temperature",
"Gain Setting",
"Gain mode",
]
def __init__(
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,
module_naming="da",
caldb_root=None,
):
super().__init__(
detector_name,
modules,
client,
event_at,
module_naming,
caldb_root,
)
self.sensor_bias_voltage = sensor_bias_voltage
self.memory_cells = memory_cells
self.pixels_x = pixels_x
self.pixels_y = pixels_y
self.integration_time = integration_time
self.sensor_temperature = sensor_temperature
self.gain_setting = gain_setting
self.gain_mode = gain_mode
def _build_condition(self, parameters):
cond = super()._build_condition(parameters)
# Fix-up some database quirks.
if int(cond.get("Gain mode", -1)) == 0:
del cond["Gain mode"]
return cond
LPD_CalibrationData
¶
Bases: SplitConditionCalibrationData
Calibration data for the LPD detector.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
class LPD_CalibrationData(SplitConditionCalibrationData):
"""Calibration data for the LPD detector."""
dark_calibrations = {
"Offset",
"Noise",
"BadPixelsDark",
}
illuminated_calibrations = {
"RelativeGain",
"GainAmpMap",
"FFMap",
"BadPixelsFF",
}
dark_parameters = [
"Sensor Bias Voltage",
"Memory cells",
"Pixels X",
"Pixels Y",
"Feedback capacitor",
"Memory cell order",
]
illuminated_parameters = [
"Sensor Bias Voltage",
"Memory cells",
"Pixels X",
"Pixels Y",
"Feedback capacitor",
"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,
memory_cell_order=None,
category=1,
modules=None,
client=None,
event_at=None,
module_naming="da",
caldb_root=None,
):
super().__init__(
detector_name,
modules,
client,
event_at,
module_naming,
caldb_root,
)
self.sensor_bias_voltage = sensor_bias_voltage
self.memory_cells = memory_cells
self.pixels_x = pixels_x
self.pixels_y = pixels_y
self.feedback_capacitor = feedback_capacitor
self.memory_cell_order = memory_cell_order
self.source_energy = source_energy
self.category = category
PNCCD_CalibrationData
¶
Bases: SplitConditionCalibrationData
Calibration data for the pnCCD detector.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
class PNCCD_CalibrationData(SplitConditionCalibrationData):
"""Calibration data for the pnCCD detector."""
dark_calibrations = {
"OffsetCCD",
"BadPixelsDarkCCD",
"NoiseCCD",
}
illuminated_calibrations = {
"RelativeGainCCD",
"CTECCD",
}
dark_parameters = [
"Sensor Bias Voltage",
"Memory cells",
"Pixels X",
"Pixels Y",
"Integration Time",
"Sensor Temperature",
"Gain Setting",
]
illuminated_parameters = dark_parameters + ["Source energy"]
def __init__(
self,
detector_name,
sensor_bias_voltage,
integration_time,
sensor_temperature,
gain_setting,
source_energy=9.2,
pixels_x=1024,
pixels_y=1024,
modules=None,
client=None,
event_at=None,
module_naming="da",
caldb_root=None,
):
# Ignore modules for this detector.
super().__init__(
detector_name,
modules,
client,
event_at,
module_naming,
caldb_root,
)
self.sensor_bias_voltage = sensor_bias_voltage
self.memory_cells = 1 # Ignore memory_cells for this detector
self.pixels_x = pixels_x
self.pixels_y = pixels_y
self.integration_time = integration_time
self.sensor_temperature = sensor_temperature
self.gain_setting = gain_setting
self.source_energy = source_energy
SplitConditionCalibrationData
¶
Bases: CalibrationData
Calibration data with dark and illuminated conditions.
Some detectors of this kind distinguish between two different operating conditions depending on whether photons illuminate the detector or not, correspondingly called the illuminated and dark conditions. Typically the illuminated condition is a superset of the dark condition.
Not all implementations for semiconductor detectors inherit from this type, but only those that make this distinction such as AGIPD and LPD.
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
class SplitConditionCalibrationData(CalibrationData):
"""Calibration data with dark and illuminated conditions.
Some detectors of this kind distinguish between two different
operating conditions depending on whether photons illuminate the
detector or not, correspondingly called the illuminated and dark
conditions. Typically the illuminated condition is a superset of the
dark condition.
Not all implementations for semiconductor detectors inherit from
this type, but only those that make this distinction such as AGIPD
and LPD.
"""
dark_calibrations = set()
illuminated_calibrations = set()
dark_parameters = list()
illuminated_parameters = list()
@property
def calibrations(self):
"""Compatibility with CalibrationData."""
return self.dark_calibrations | self.illuminated_calibrations
@property
def parameters(self):
"""Compatibility with CalibrationData."""
# Removes likely duplicates while preserving order.
return list(
dict.fromkeys(self.dark_parameters + self.illuminated_parameters)
)
@property
def condition(self):
"""Compatibility with CalibrationData."""
cond = dict()
cond.update(self.dark_condition)
cond.update(self.illuminated_condition)
return cond
@property
def dark_condition(self):
return self._build_condition(self.dark_parameters)
@property
def illuminated_condition(self):
return self._build_condition(self.illuminated_parameters)
def metadata(
self,
calibrations=None,
event_at=None,
pdu_snapshot_at=None,
):
"""Query CCV metadata for calibrations, conditions and time.
Args:
calibrations (Iterable of str, optional): Calibrations to
query metadata for, may be None to retrieve all.
event_at (datetime, date, str or None): Time at which the
CCVs should have been valid, now or default value passed at
initialization time if omitted.
pdu_snapshot_at (datetime, date, str or None): Time of database
state to look at, now or default value passed at
initialization time if omitted.
Returns:
(CCVMetadata) CCV metadata result.
"""
if calibrations is None:
calibrations = (
self.dark_calibrations | self.illuminated_calibrations
)
metadata = CCVMetadata()
# Calibrations are sorted to ensure using exactly the same query
# for multiple configuration. e.g. This is essential for calparrot.
dark_calibrations = sorted(
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,
pdu_snapshot_at or self.pdu_snapshot_at,
metadata,
module_naming=self.module_naming,
)
illum_calibrations = sorted(
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,
pdu_snapshot_at or self.pdu_snapshot_at,
metadata,
module_naming=self.module_naming,
)
return metadata
calibrations
property
¶
Compatibility with CalibrationData.
condition
property
¶
Compatibility with CalibrationData.
parameters
property
¶
Compatibility with CalibrationData.
metadata(calibrations=None, event_at=None, pdu_snapshot_at=None)
¶
Query CCV metadata for calibrations, conditions and time.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
calibrations |
Iterable of str
|
Calibrations to query metadata for, may be None to retrieve all. |
None
|
event_at |
datetime, date, str or None
|
Time at which the CCVs should have been valid, now or default value passed at initialization time if omitted. |
None
|
pdu_snapshot_at |
datetime, date, str or None
|
Time of database state to look at, now or default value passed at initialization time if omitted. |
None
|
Returns:
| Type | Description |
|---|---|
(CCVMetadata) CCV metadata result. |
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/calcat_interface.py
def metadata(
self,
calibrations=None,
event_at=None,
pdu_snapshot_at=None,
):
"""Query CCV metadata for calibrations, conditions and time.
Args:
calibrations (Iterable of str, optional): Calibrations to
query metadata for, may be None to retrieve all.
event_at (datetime, date, str or None): Time at which the
CCVs should have been valid, now or default value passed at
initialization time if omitted.
pdu_snapshot_at (datetime, date, str or None): Time of database
state to look at, now or default value passed at
initialization time if omitted.
Returns:
(CCVMetadata) CCV metadata result.
"""
if calibrations is None:
calibrations = (
self.dark_calibrations | self.illuminated_calibrations
)
metadata = CCVMetadata()
# Calibrations are sorted to ensure using exactly the same query
# for multiple configuration. e.g. This is essential for calparrot.
dark_calibrations = sorted(
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,
pdu_snapshot_at or self.pdu_snapshot_at,
metadata,
module_naming=self.module_naming,
)
illum_calibrations = sorted(
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,
pdu_snapshot_at or self.pdu_snapshot_at,
metadata,
module_naming=self.module_naming,
)
return metadata