Source code for xrd_tools.meta
import inspect
import json
import logging
from dataclasses import dataclass, field
from . import device_factory, utils
from .config import ENCODING, JSON_INDENT
from .device import DEVICE_TYPES, HTChamber, LabDiffractometer
logger = logging.getLogger(__name__)
[docs]@dataclass
class Meta:
"""Class to store and handle metadata of a XRD measurement.
Parameters
----------
measurement_id :
The ID of the measurement.
operator :
The operator of the measurement.
description :
A description of the measurement.
sample :
The ID of the sample that was measured.
compound :
The formula of the compound that was measured.
temperature :
The temperature at which the measurement took place.
pressure :
The pressure at which the measurement took place.
atmosphere :
The atmosphere in which the measurement took place.
ht_mode :
Whether the measurement was taken in high temperature mode.
devices :
A list of dictionaries containing information about the devices used in the measurement.
Each dictionary holds the arguments required to initiate the corresponding Device object.
xrd_datetime :
The date and time at which the measurement was taken.
processing_state :
The processing state of the measurement, expression must be defined in PROCESSING_STATES.
comment :
A comment about the measurement.
"""
measurement_id: str
ht_mode: bool = None # ht_mode must be listed prior to devices!
operator: str = None
description: str = None
sample: str = None
compound: str = None
temperature: float = None
pressure: float = None
atmosphere: str = None
devices: list[dict[str, any]] = field(default_factory=lambda: [])
xrd_datetime: str = None
processing_state: str = None
comment: str = None
def __post_init__(self):
self._register_device_types()
self._validate_devices()
def _register_device_types(self) -> None:
"""Register device types."""
device_factory.register(DEVICE_TYPES["lab_xrd"], LabDiffractometer)
device_factory.register(DEVICE_TYPES["ht_chamber"], HTChamber)
def _get_devices(self) -> list[device_factory.Device]:
"""Returns list with objects of devices used in the measurement."""
return [device_factory.create(item)[0] for item in self.devices]
def _validate_devices(self) -> None:
"""Validates number and types of devices assigned to a measurement.
If ht_mode is False, checks that number of devices is at most 1 and that
the device is a LabDiffractometer instance. If either condition is not met,
raises ValueError.
If ht_mode is True, checks that number of devices is at most 2 and that
each device is either a LabDiffractometer or a HTChamber instance, and not
the same instance as the previous device. If either condition is not met,
raises ValueError.
Raises:
ValueError: If number or types of devices are invalid.
"""
devices = self._get_devices()
# Checks for measurement in high temperature mode
if self.ht_mode:
if len(devices) > 2:
raise ValueError(
(
f"Invalid amount of devices! Maximum two devices are allowed to be "
f"assigned to a {self.mode} measurement."
)
)
if len(devices) == 2:
for i, device in enumerate(devices):
if not isinstance(device, (LabDiffractometer, HTChamber)):
raise ValueError("Invalid device type.")
if i > 0 and id(device) == id(devices[i - 1]):
raise ValueError(
f"Devices must be of type {DEVICE_TYPES['lab_xrd']!r} and "
f"{DEVICE_TYPES['ht_chamber']!r} in a {self.mode} measurement."
)
else:
# Checks for measurement in ambient mode
if len(devices) > 1:
raise ValueError(
(
f"Invalid amount of devices! Exactly one device is allowed to be "
f"assigned to a {self.mode} measurement."
)
)
if len(devices) == 1 and not isinstance(devices[0], LabDiffractometer):
raise ValueError(
(
f"Invalid device type! Only one device of type {DEVICE_TYPES['lab_xrd']!r} "
f"is allowed to be assigned to a {self.mode} measurement."
)
)
@property
def xrd_device(self) -> LabDiffractometer:
"""Returns diffractometer device object."""
for dev in self._get_devices():
if dev.device_type == DEVICE_TYPES["lab_xrd"]:
return dev
logger.debug(f"No {DEVICE_TYPES['lab_xrd']} device registered.")
return None
@property
def mode(self) -> str:
"""Returns string with measurement mode ('XRD' or 'HT-XRD')."""
if not self.ht_mode:
mode_str = "XRD"
else:
mode_str = "HT-XRD"
return mode_str
[docs] @classmethod
def from_json(cls, file_path: str = None, json_str: str = None):
"""
Alternative constructor to initiate a Meta object from a JSON string.
:param file_path: path to the JSON file to read from
:param json_str: JSON string containing the Meta attributes
Either the `file_path` or `json_str` parameter must be provided.
If both are provided, the `json_str` parameter takes precedence.
:return: Meta object with attributes initialized from the JSON string
"""
if file_path is not None:
utils.ensure_file_exists(file_path)
with open(file_path, "r", encoding=ENCODING) as fobj:
json_str = fobj.read()
return cls(**json.loads(json_str))
[docs] def to_json(
self, file_path: str, indent: int = JSON_INDENT, encoding: str = ENCODING
) -> None:
"""Writes Meta class dict as JSON string to file."""
kwargs = self.__dict__.copy()
kwargs = {k: v for k, v in kwargs.items() if v is not None}
utils.write_to_json(
file_path=file_path, data=kwargs, indent=indent, encoding=encoding
)
@staticmethod
def _validate_key(key: str) -> None:
"""Check if provided metadata key is known.
Raises:
KeyError: If a key not corresponding to an class argument is provided.
"""
keys = list(inspect.signature(Meta).parameters)
if key not in keys:
raise KeyError(f"Unknown metadata key: {key!r}")
[docs] def update_value(self, key: str, value: any, file_path: str = None) -> None:
"""Assign value to key of metadata dictionary.
Args:
file_path (str): Write metadata to JSON file if a path is provided.
"""
self._validate_key(key)
old_value = self.__dict__[key]
if value == old_value:
logger.debug(f"{key!r} did not change ({value}).")
return None
self.__dict__[key] = value
if old_value is None:
logger.info(f"Set {key!r}: {value!r}")
elif value is None:
logger.info(f"Reset {key!r} (old value: {old_value!r})")
else:
logger.info(f"Updated {key!r}: {old_value!r} -> {value!r}")
if file_path is not None:
self.to_json(file_path)