168 lines
12 KiB
Python
168 lines
12 KiB
Python
# coding: utf-8
|
|
|
|
"""
|
|
Jungfraujoch
|
|
|
|
API to control Jungfraujoch developed by the Paul Scherrer Institute (Switzerland). Jungfraujoch is a data acquisition and analysis system for pixel array detectors, primarly PSI JUNGFRAU. Jungfraujoch uses FPGA boards to acquire data at high data rates.
|
|
|
|
The version of the OpenAPI document: 1.0.0-rc.23
|
|
Contact: filip.leonarski@psi.ch
|
|
Generated by OpenAPI Generator (https://openapi-generator.tech)
|
|
|
|
Do not edit the class manually.
|
|
""" # noqa: E501
|
|
|
|
|
|
from __future__ import annotations
|
|
import pprint
|
|
import re # noqa: F401
|
|
import json
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictFloat, StrictInt, StrictStr, field_validator
|
|
from typing import Any, ClassVar, Dict, List, Optional, Union
|
|
from typing_extensions import Annotated
|
|
from jfjoch_client.models.dataset_settings_unit_cell import DatasetSettingsUnitCell
|
|
from jfjoch_client.models.rotation_axis import RotationAxis
|
|
from typing import Optional, Set
|
|
from typing_extensions import Self
|
|
|
|
class DatasetSettings(BaseModel):
|
|
"""
|
|
DatasetSettings
|
|
""" # noqa: E501
|
|
images_per_trigger: Optional[Annotated[int, Field(strict=True, ge=1)]] = Field(default=1, description="For standard synchrotron data collection - this is number of images collected per one TTL trigger For XFEL (pulsed source) - this number is ignored and set to 1 For storage cell mode - this number is ignored and set to number of storage cells ")
|
|
ntrigger: Optional[Annotated[int, Field(strict=True, ge=1)]] = Field(default=1, description="Number of TTL trigger that the detector is expected to receive during data collection ")
|
|
image_time_us: Optional[Annotated[int, Field(strict=True, ge=0)]] = Field(default=None, description="Image time. If not provided (or zero value) the frame time is assumed as default. Image time must be multiple of frame time; max value is 256 * frame_time. In XFEL mode: summation happens for frames collected with multiple triggers. Ignored for storage cells and if raw data are saved. ")
|
|
beam_x_pxl: Union[StrictFloat, StrictInt] = Field(description="/entry/detector/beam_center_x in NXmx Beam center in X direction [pixels] ")
|
|
beam_y_pxl: Union[StrictFloat, StrictInt] = Field(description="/entry/detector/beam_center_y in NXmx Beam center in X direction [pixels] ")
|
|
detector_distance_mm: Union[Annotated[float, Field(strict=True, ge=0)], Annotated[int, Field(strict=True, ge=0)]] = Field(description="/entry/detector/distance in NXmx Detector distance [mm]")
|
|
incident_energy_ke_v: Union[Annotated[float, Field(le=500.0, strict=True, ge=0.001)], Annotated[int, Field(le=500, strict=True, ge=1)]] = Field(description="Used to calculate /entry/beam/incident_wavelength in NXmx Incident particle (photon, electron) energy in keV ", alias="incident_energy_keV")
|
|
file_prefix: Optional[StrictStr] = Field(default='', description="Prefix for filenames. If left empty, no file will be saved.")
|
|
images_per_file: Optional[Annotated[int, Field(strict=True, ge=0)]] = Field(default=1000, description="Number of files in a single HDF5 data file (0 = write all images to a single data file).")
|
|
space_group_number: Optional[Annotated[int, Field(le=194, strict=True, ge=0)]] = 0
|
|
sample_name: Optional[StrictStr] = Field(default='', description="/entry/sample/name in NXmx Sample name ")
|
|
compression: Optional[StrictStr] = 'bslz4'
|
|
total_flux: Optional[Union[StrictFloat, StrictInt]] = Field(default=None, description="/entry/beam/total_flux in NXmx Flux incident on beam plane in photons per second. In other words this is the flux integrated over area. [photons/s] ")
|
|
transmission: Optional[Union[Annotated[float, Field(le=1.0, strict=True, ge=0.0)], Annotated[int, Field(le=1, strict=True, ge=0)]]] = Field(default=None, description="/entry/instrument/attenuator/attenuator_transmission Transmission of attenuator (filter) [no units] ")
|
|
goniometer: Optional[RotationAxis] = None
|
|
header_appendix: Optional[Any] = Field(default=None, description="Header appendix, added as user_data/user to start message (can be any valid JSON)")
|
|
image_appendix: Optional[Any] = Field(default=None, description="Image appendix, added as user_data to image message (can be any valid JSON)")
|
|
data_reduction_factor_serialmx: Optional[Union[Annotated[float, Field(le=1.0, strict=True, ge=0.0)], Annotated[int, Field(le=1, strict=True, ge=0)]]] = Field(default=1.0, description="Rate at which non-indexed images are accepted to be forwarded to writer. Value of 1.0 (default) means that all images are written. Values below zero mean that non-indexed images will be accepted with a given probability. ")
|
|
pixel_value_low_threshold: Optional[Annotated[int, Field(strict=True, ge=0)]] = Field(default=None, description="Set all counts lower than the value to zero. When the value is set, negative numbers other than error pixel value are always set to zero. Setting to zero is equivalent to turning the option off. ")
|
|
run_number: Optional[Annotated[int, Field(strict=True, ge=0)]] = Field(default=None, description="Number of run within an experimental session. Transferred over CBOR stream as \"series ID\", though not saved in HDF5 file. It is highly recommended to keep this number unique for each data collection during experimental series. If not provided, the number will be automatically incremented. ")
|
|
run_name: Optional[StrictStr] = Field(default=None, description="Unique ID of run. Transferred over CBOR stream as \"unique series ID\", though not saved in HDF5 file. It is highly recommended to keep this name unique for each data collection during experimental series. If not provided, the name will be automatically generated as number + colon + file_prefix. ")
|
|
experiment_group: Optional[StrictStr] = Field(default=None, description="Name of group owning the data (e.g. p-group or proposal number). Transferred over CBOR stream, though not saved in HDF5 file. ")
|
|
poisson_compression: Optional[Annotated[int, Field(le=16, strict=True, ge=0)]] = Field(default=None, description="Enable lossy compression of pixel values that preserves Poisson statistics. Requires to provide a numerical factor SQ. Pixel value P will be transformed to round(sqrt(P) * SQ), with rounding to the closest integer. Compression is turned off if the value is missing or it is set to zero. ")
|
|
write_nxmx_hdf5_master: Optional[StrictBool] = Field(default=True, description="Write NXmx formatted HDF5 master file. Recommended to use for macromolecular crystallography experiments and to turn off for other experiments. ")
|
|
save_calibration: Optional[StrictBool] = Field(default=None, description="Forward image calibration (at the moment pedestal and pedestal RMS for JUNGFRAU) using the ZeroMQ stream to writer. If parameter is not provided calibration will be saved only if more than 4 images are recorded. ")
|
|
unit_cell: Optional[DatasetSettingsUnitCell] = None
|
|
__properties: ClassVar[List[str]] = ["images_per_trigger", "ntrigger", "image_time_us", "beam_x_pxl", "beam_y_pxl", "detector_distance_mm", "incident_energy_keV", "file_prefix", "images_per_file", "space_group_number", "sample_name", "compression", "total_flux", "transmission", "goniometer", "header_appendix", "image_appendix", "data_reduction_factor_serialmx", "pixel_value_low_threshold", "run_number", "run_name", "experiment_group", "poisson_compression", "write_nxmx_hdf5_master", "save_calibration", "unit_cell"]
|
|
|
|
@field_validator('compression')
|
|
def compression_validate_enum(cls, value):
|
|
"""Validates the enum"""
|
|
if value is None:
|
|
return value
|
|
|
|
if value not in set(['bslz4', 'bszstd', 'bszstd_rle', 'none']):
|
|
raise ValueError("must be one of enum values ('bslz4', 'bszstd', 'bszstd_rle', 'none')")
|
|
return value
|
|
|
|
model_config = ConfigDict(
|
|
populate_by_name=True,
|
|
validate_assignment=True,
|
|
protected_namespaces=(),
|
|
)
|
|
|
|
|
|
def to_str(self) -> str:
|
|
"""Returns the string representation of the model using alias"""
|
|
return pprint.pformat(self.model_dump(by_alias=True))
|
|
|
|
def to_json(self) -> str:
|
|
"""Returns the JSON representation of the model using alias"""
|
|
# TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead
|
|
return json.dumps(self.to_dict())
|
|
|
|
@classmethod
|
|
def from_json(cls, json_str: str) -> Optional[Self]:
|
|
"""Create an instance of DatasetSettings from a JSON string"""
|
|
return cls.from_dict(json.loads(json_str))
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Return the dictionary representation of the model using alias.
|
|
|
|
This has the following differences from calling pydantic's
|
|
`self.model_dump(by_alias=True)`:
|
|
|
|
* `None` is only added to the output dict for nullable fields that
|
|
were set at model initialization. Other fields with value `None`
|
|
are ignored.
|
|
"""
|
|
excluded_fields: Set[str] = set([
|
|
])
|
|
|
|
_dict = self.model_dump(
|
|
by_alias=True,
|
|
exclude=excluded_fields,
|
|
exclude_none=True,
|
|
)
|
|
# override the default output from pydantic by calling `to_dict()` of goniometer
|
|
if self.goniometer:
|
|
_dict['goniometer'] = self.goniometer.to_dict()
|
|
# override the default output from pydantic by calling `to_dict()` of unit_cell
|
|
if self.unit_cell:
|
|
_dict['unit_cell'] = self.unit_cell.to_dict()
|
|
# set to None if header_appendix (nullable) is None
|
|
# and model_fields_set contains the field
|
|
if self.header_appendix is None and "header_appendix" in self.model_fields_set:
|
|
_dict['header_appendix'] = None
|
|
|
|
# set to None if image_appendix (nullable) is None
|
|
# and model_fields_set contains the field
|
|
if self.image_appendix is None and "image_appendix" in self.model_fields_set:
|
|
_dict['image_appendix'] = None
|
|
|
|
return _dict
|
|
|
|
@classmethod
|
|
def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
|
|
"""Create an instance of DatasetSettings from a dict"""
|
|
if obj is None:
|
|
return None
|
|
|
|
if not isinstance(obj, dict):
|
|
return cls.model_validate(obj)
|
|
|
|
_obj = cls.model_validate({
|
|
"images_per_trigger": obj.get("images_per_trigger") if obj.get("images_per_trigger") is not None else 1,
|
|
"ntrigger": obj.get("ntrigger") if obj.get("ntrigger") is not None else 1,
|
|
"image_time_us": obj.get("image_time_us"),
|
|
"beam_x_pxl": obj.get("beam_x_pxl"),
|
|
"beam_y_pxl": obj.get("beam_y_pxl"),
|
|
"detector_distance_mm": obj.get("detector_distance_mm"),
|
|
"incident_energy_keV": obj.get("incident_energy_keV"),
|
|
"file_prefix": obj.get("file_prefix") if obj.get("file_prefix") is not None else '',
|
|
"images_per_file": obj.get("images_per_file") if obj.get("images_per_file") is not None else 1000,
|
|
"space_group_number": obj.get("space_group_number") if obj.get("space_group_number") is not None else 0,
|
|
"sample_name": obj.get("sample_name") if obj.get("sample_name") is not None else '',
|
|
"compression": obj.get("compression") if obj.get("compression") is not None else 'bslz4',
|
|
"total_flux": obj.get("total_flux"),
|
|
"transmission": obj.get("transmission"),
|
|
"goniometer": RotationAxis.from_dict(obj["goniometer"]) if obj.get("goniometer") is not None else None,
|
|
"header_appendix": obj.get("header_appendix"),
|
|
"image_appendix": obj.get("image_appendix"),
|
|
"data_reduction_factor_serialmx": obj.get("data_reduction_factor_serialmx") if obj.get("data_reduction_factor_serialmx") is not None else 1.0,
|
|
"pixel_value_low_threshold": obj.get("pixel_value_low_threshold"),
|
|
"run_number": obj.get("run_number"),
|
|
"run_name": obj.get("run_name"),
|
|
"experiment_group": obj.get("experiment_group"),
|
|
"poisson_compression": obj.get("poisson_compression"),
|
|
"write_nxmx_hdf5_master": obj.get("write_nxmx_hdf5_master") if obj.get("write_nxmx_hdf5_master") is not None else True,
|
|
"save_calibration": obj.get("save_calibration"),
|
|
"unit_cell": DatasetSettingsUnitCell.from_dict(obj["unit_cell"]) if obj.get("unit_cell") is not None else None
|
|
})
|
|
return _obj
|
|
|
|
|