From 742a4413fdf515d01255b03bfdd5ed2906e701ec Mon Sep 17 00:00:00 2001 From: pecomyint Date: Mon, 24 Aug 2026 16:53:20 -0500 Subject: [PATCH] Refactor: centralize safe RSM configuration and geometry --- .../hpc/analysis/hpc_rsm_consumer.py | 113 +++--- .../hpc_spontaneous_analysis_consumer.py | 4 +- .../hpc/meta/hpc_metadata_consumer.py | 4 +- src/dashpva/database/interface.py | 15 + src/dashpva/database/managers/profile.py | 60 +++ src/dashpva/settings.py | 39 +- src/dashpva/utils/__init__.py | 6 + src/dashpva/utils/config/__init__.py | 18 +- src/dashpva/utils/config/resolver.py | 175 +++++++++ src/dashpva/utils/config/revision.py | 20 + src/dashpva/utils/config/source.py | 288 +++++++++++++-- src/dashpva/utils/rsm_converter.py | 105 ++++-- src/dashpva/utils/rsm_geometry.py | 347 ++++++++++++++++++ .../viewer/area_det/area_det_viewer.py | 119 +++--- src/dashpva/workflow/workflow.py | 2 +- tests/unit/test_config_cas.py | 143 ++++++++ tests/unit/test_config_resolution.py | 153 ++++++++ tests/unit/test_pr1_geometry_contract.py | 231 ++++++++++++ 18 files changed, 1680 insertions(+), 162 deletions(-) create mode 100644 src/dashpva/utils/config/resolver.py create mode 100644 src/dashpva/utils/config/revision.py create mode 100644 src/dashpva/utils/rsm_geometry.py create mode 100644 tests/unit/test_config_cas.py create mode 100644 tests/unit/test_config_resolution.py create mode 100644 tests/unit/test_pr1_geometry_contract.py diff --git a/src/dashpva/consumers/hpc/analysis/hpc_rsm_consumer.py b/src/dashpva/consumers/hpc/analysis/hpc_rsm_consumer.py index a3c321a2..a046e73c 100644 --- a/src/dashpva/consumers/hpc/analysis/hpc_rsm_consumer.py +++ b/src/dashpva/consumers/hpc/analysis/hpc_rsm_consumer.py @@ -9,13 +9,46 @@ import lz4.block import numpy as np import pvaccess as pva -import xrayutilities as xu from pvaccess import PvObject from pvapy.hpc.adImageProcessor import AdImageProcessor from pvapy.utility.floatWithUnits import FloatWithUnits from pvapy.utility.timeUtility import TimeUtility from dashpva.utils.log_manager import LogMixin +from dashpva.utils.rsm_geometry import ( + DetectorModel, + RotationAxis, + RSMGeometry, + build_hxrd, + calculate_q, +) + +_RSM_SECTION_FIELDS = { + 'PRIMARY_BEAM_DIRECTION': ('AXIS_NUMBER_1', 'AXIS_NUMBER_2', 'AXIS_NUMBER_3'), + 'INPLANE_REFERENCE_DIRECITON': ('AXIS_NUMBER_1', 'AXIS_NUMBER_2', 'AXIS_NUMBER_3'), + 'SAMPLE_SURFACE_NORMAL_DIRECITON': ('AXIS_NUMBER_1', 'AXIS_NUMBER_2', 'AXIS_NUMBER_3'), + 'SPEC': ('ENERGY_VALUE', 'UB_MATRIX_VALUE'), + 'DETECTOR_SETUP': ( + 'CENTER_CHANNEL_PIXEL', + 'DISTANCE', + 'PIXEL_DIRECTION_1', + 'PIXEL_DIRECTION_2', + 'SIZE', + ), +} + + +def _required_rsm_channels(hkl_config): + channels = set() + for section_name, section in hkl_config.items(): + if not isinstance(section, dict): + continue + if section_name.startswith(('SAMPLE_CIRCLE', 'DETECTOR_CIRCLE')): + fields = ('DIRECTION_AXIS', 'POSITION') + else: + fields = _RSM_SECTION_FIELDS.get(section_name, ()) + channels.update(section[field] for field in fields if section.get(field)) + return channels class HpcRsmProcessor(AdImageProcessor, LogMixin): @@ -123,7 +156,6 @@ def __init__(self, configDict={}): self.hkl_pv_channels = set() self.hkl_attributes = {} self.old_attrbutes : dict = None - self.q_conv = None self.qx = None self.qy = None self.qz = None @@ -141,9 +173,9 @@ def configure(self, configDict): """Configure processor settings and initialize HKL parameters from DB config.""" self.logger.debug(f'Configuration update: {configDict}') - from dashpva.utils.config.source import ConfigSource + from dashpva.utils.config import ConfigSource, resolve_profile_config locator = configDict.get('profile_id') or configDict.get('path') or None - config = ConfigSource(locator).load() + config = resolve_profile_config(ConfigSource(locator).load()) if not config: raise RuntimeError( @@ -153,12 +185,7 @@ def configure(self, configDict): self.config = config self.hkl_config = self.config.get('HKL') or {} - self.hkl_pv_channels = set() - for section in self.hkl_config.values(): - if isinstance(section, dict): - for channel in section.values(): - if channel: - self.hkl_pv_channels.add(channel) + self.hkl_pv_channels = _required_rsm_channels(self.hkl_config) def parse_hkl_ndattributes(self, pva_object): """ @@ -193,17 +220,11 @@ def get_sample_and_detector_circles(self, hkl_attr: dict): # loop sorting pv channels for section, pv_dict in self.hkl_config.items(): if section.startswith('SAMPLE_CIRCLE'): - for pv_name in pv_dict.values(): - if pv_name.endswith('DirectionAxis'): - sample_circle_directions.append(hkl_attr[pv_name]) - elif pv_name.endswith('Position'): - sample_circle_positions.append(hkl_attr[pv_name]) + sample_circle_directions.append(hkl_attr[pv_dict['DIRECTION_AXIS']]) + sample_circle_positions.append(hkl_attr[pv_dict['POSITION']]) elif section.startswith('DETECTOR_CIRCLE'): - for pv_name in pv_dict.values(): - if pv_name.endswith('DirectionAxis'): - det_circle_directions.append(hkl_attr[pv_name]) - elif pv_name.endswith('Position'): - det_circle_positions.append(hkl_attr[pv_name]) + det_circle_directions.append(hkl_attr[pv_dict['DIRECTION_AXIS']]) + det_circle_positions.append(hkl_attr[pv_dict['POSITION']]) return sample_circle_directions, sample_circle_positions, det_circle_directions, det_circle_positions @@ -249,18 +270,6 @@ def create_rsm(self, hkl_attr: dict, shape: tuple): ub_matrix = np.reshape(ub_matrix, (3,3)) energy = self.get_energy(hkl_attr) * 1000 - # Initialize QConversion - q_conv = xu.experiment.QConversion( - sample_circle_directions, - det_circle_directions, - primary_beam_directions - ) - # Initialize HXRD - hxrd = xu.HXRD(inplane_beam_direction, - sample_surface_normal_direction, - en=energy, - qconv=q_conv) - # Set up detector parameters — look up by the PV name in the active # HKL config so any prefix (xidb:, 6idb:, none) works without edits. ds_cfg = self.hkl_config.get('DETECTOR_SETUP', {}) or {} @@ -274,18 +283,38 @@ def create_rsm(self, hkl_attr: dict, shape: tuple): pixel_width2 = size_xy[1] / nch2 distance = hkl_attr[ds_cfg['DISTANCE']] - hxrd.Ang2Q.init_area( - pixel_dir1, pixel_dir2, - cch1=cch1, cch2=cch2, - Nch1=nch1, Nch2=nch2, - pwidth1=pixel_width1, - pwidth2=pixel_width2, + detector = DetectorModel( + pixel_direction_1=pixel_dir1, + pixel_direction_2=pixel_dir2, + center_channel=(cch1, cch2), + shape=(nch1, nch2), + pixel_width=(pixel_width1, pixel_width2), distance=distance, - roi=roi + roi=tuple(roi), + ) + canonical = self.config.get('IOC_RSM_PARAMETER', {}) or {} + model = RSMGeometry( + sample_axes=tuple( + RotationAxis('sample', direction) + for direction in sample_circle_directions + ), + detector_axes=tuple( + RotationAxis('detector', direction) + for direction in det_circle_directions + ), + primary_beam_direction=primary_beam_directions, + inplane_reference_direction=inplane_beam_direction, + sample_surface_normal_direction=sample_surface_normal_direction, + energy_eV=energy, + ub_matrix=ub_matrix, + detector=detector, + sample_orientation=canonical.get('SAMPLE_ORIENTATION', 'det'), + ) + return calculate_q( + build_hxrd(model), + sample_circle_positions, + det_circle_positions, ) - - angles = [*sample_circle_positions, *det_circle_positions] - return hxrd.Ang2Q.area(*angles, UB=ub_matrix) except Exception as e: try: if hasattr(self, 'logger'): diff --git a/src/dashpva/consumers/hpc/analysis/hpc_spontaneous_analysis_consumer.py b/src/dashpva/consumers/hpc/analysis/hpc_spontaneous_analysis_consumer.py index 575faed9..33c58637 100755 --- a/src/dashpva/consumers/hpc/analysis/hpc_spontaneous_analysis_consumer.py +++ b/src/dashpva/consumers/hpc/analysis/hpc_spontaneous_analysis_consumer.py @@ -49,11 +49,11 @@ def configure(self, configDict): self.path = configDict['path'] else: import dashpva.settings as _settings - self.path = _settings.TOML_FILE + self.path = _settings.ensure_path() if self.path is None: raise RuntimeError( "HpcAnalysisProcessor: no 'path' in configDict and " - "settings.TOML_FILE is not set — configure a TOML config first." + "no effective config path is available — configure a profile first." ) with open(self.path, 'r') as f: diff --git a/src/dashpva/consumers/hpc/meta/hpc_metadata_consumer.py b/src/dashpva/consumers/hpc/meta/hpc_metadata_consumer.py index 43fbfbc4..d595e2eb 100755 --- a/src/dashpva/consumers/hpc/meta/hpc_metadata_consumer.py +++ b/src/dashpva/consumers/hpc/meta/hpc_metadata_consumer.py @@ -132,11 +132,11 @@ def configure(self, configDict): self.path = configDict["path"] else: import dashpva.settings as _settings - self.path = _settings.TOML_FILE + self.path = _settings.ensure_path() if self.path is None: raise RuntimeError( "HpcAdMetadataProcessor: no 'path' in configDict and " - "settings.TOML_FILE is not set — configure a TOML config first." + "no effective config path is available — configure a profile first." ) with open(self.path, "r") as config_file: diff --git a/src/dashpva/database/interface.py b/src/dashpva/database/interface.py index 3378d7ef..0c10d4ff 100644 --- a/src/dashpva/database/interface.py +++ b/src/dashpva/database/interface.py @@ -127,6 +127,21 @@ def rename_config_type(self, profile_id: int, old_type: str, new_type: str) -> b # Import / Export TOML + def load_profile_toml_strict(self, profile_id: int) -> Dict[str, Any]: + return self._mgr.load_profile_toml_strict(profile_id) + + def replace_profile_toml_if_revision( + self, + profile_id: int, + toml_data: Dict[str, Any], + expected_revision: str, + ) -> str: + return self._mgr.replace_profile_toml_if_revision( + profile_id, + toml_data, + expected_revision, + ) + def import_toml_to_profile(self, profile_id: int, toml_data: Dict[str, Any]) -> bool: return self._mgr.import_toml_to_profile(profile_id, toml_data) diff --git a/src/dashpva/database/managers/profile.py b/src/dashpva/database/managers/profile.py index bbdcd896..72383fee 100644 --- a/src/dashpva/database/managers/profile.py +++ b/src/dashpva/database/managers/profile.py @@ -4,10 +4,12 @@ from typing import Any, Dict, List, Optional import toml +from sqlalchemy import text from dashpva.database.db import get_session from dashpva.database.managers.base import BaseManager from dashpva.database.models.profile import Profile, ProfileConfig +from dashpva.utils.config.revision import mapping_revision class ProfileManager(BaseManager): @@ -310,6 +312,64 @@ def rename_config_type(self, profile_id: int, old_type: str, new_type: str) -> b # Import / Export TOML # ------------------------------------------------------------------ # + def load_profile_toml_strict(self, profile_id: int) -> Dict[str, Any]: + """Load one profile without converting database errors to an empty dict.""" + session = get_session() + try: + if session.query(Profile.id).filter_by(id=profile_id).first() is None: + raise LookupError(f"profile {profile_id} does not exist") + blob = session.query(ProfileConfig).filter_by( + profile_id=profile_id, + config_type='__toml__', + config_key='__data__', + ).first() + return self.clean(json.loads(blob.config_value)) if blob else {} + finally: + session.close() + + def replace_profile_toml_if_revision( + self, + profile_id: int, + toml_data: Dict[str, Any], + expected_revision: str, + ) -> str: + """Atomically replace a profile blob when its revision still matches.""" + session = get_session() + try: + session.execute(text("BEGIN IMMEDIATE")) + if session.query(Profile.id).filter_by(id=profile_id).first() is None: + session.rollback() + return "error" + + blob = session.query(ProfileConfig).filter_by( + profile_id=profile_id, + config_type='__toml__', + config_key='__data__', + ).first() + current = self.clean(json.loads(blob.config_value)) if blob else {} + if mapping_revision(current) != expected_revision: + session.rollback() + return "conflict" + + encoded = json.dumps(toml_data) + if blob is None: + session.add(ProfileConfig( + profile_id=profile_id, + config_type='__toml__', + config_section=None, + config_key='__data__', + config_value=encoded, + )) + else: + blob.config_value = encoded + session.commit() + return "saved" + except Exception: + session.rollback() + return "error" + finally: + session.close() + def import_toml_to_profile(self, profile_id: int, toml_data: Dict[str, Any]) -> bool: """Store the full TOML dict as a JSON blob for reliable round-trip export.""" session = get_session() diff --git a/src/dashpva/settings.py b/src/dashpva/settings.py index 4e7c720e..7369ffd0 100644 --- a/src/dashpva/settings.py +++ b/src/dashpva/settings.py @@ -27,8 +27,9 @@ - Diagnostics: settings.SOURCE_TYPE -> "toml", "db", or None settings.LOCATOR -> the current locator (str path, "profile:", or int id) - settings.CONFIG -> full configuration dictionary - settings.ensure_path() -> a TOML path (original path or temp file when using DB) + settings.RAW_CONFIG -> exact persisted configuration dictionary + settings.CONFIG -> effective runtime configuration dictionary + settings.ensure_path() -> a TOML path containing the effective configuration """ import os @@ -36,10 +37,14 @@ from typing import Any, Dict, Optional, Union try: + from dashpva.utils.config.resolver import resolve_profile_config from dashpva.utils.config.source import ConfigSource except Exception: ConfigSource = None # type: ignore[assignment] + def resolve_profile_config(raw): + return dict(raw or {}) + # NeXus / HDF5 structure definition (static — not config-driven) HDF5_STRUCTURE = { @@ -239,6 +244,7 @@ CONSUMERS_PATH: Optional[str] = None # Diagnostics +RAW_CONFIG: Dict[str, Any] = {} CONFIG: Dict[str, Any] = {} SOURCE_TYPE: Optional[str] = None LOCATOR: Optional[Union[int, str]] = None @@ -273,16 +279,19 @@ def set_locator(locator: Union[int, str]) -> None: def ensure_path() -> Optional[str]: - """Return a TOML path: original path for TOML sources, temp file for DB sources.""" + """Return a TOML path containing the effective runtime configuration.""" eff = _get_effective_locator() if ConfigSource is None: return None - return ConfigSource(eff).ensure_path() + src = ConfigSource(eff) + raw = src.load() + effective = resolve_profile_config(raw) + return src.ensure_path(effective if effective != raw else None) def reload() -> None: """Re-resolve current LOCATOR and repopulate all exported constants from the configuration source.""" - global CONFIG, SOURCE_TYPE, LOCATOR, TOML_FILE + global RAW_CONFIG, CONFIG, SOURCE_TYPE, LOCATOR, TOML_FILE global DETECTOR_PREFIX, IOC_PREFIX, INPUT_CHANNEL, INPUT_CHANNEL_HKL3D, OUTPUT_FILE_LOCATION, CONSUMER_MODE global CACHING_MODE, CACHE_OPTIONS, ALIGNMENT_MAX_CACHE_SIZE global SCAN_FLAG_PV, FILE_PATH_PV, FILE_NAME_PV @@ -295,12 +304,14 @@ def reload() -> None: LOCATOR = eff src = ConfigSource(eff) if ConfigSource else None - cfg = src.load() if src else {} + raw_cfg = src.load() if src else {} + cfg = resolve_profile_config(raw_cfg) + RAW_CONFIG = raw_cfg CONFIG = cfg SOURCE_TYPE = src.source_type if (src and eff is not None) else None try: - TOML_FILE = ensure_path() + TOML_FILE = src.ensure_path() if src else None except Exception: TOML_FILE = None @@ -527,6 +538,7 @@ def __init__( self.source_type: Optional[str] = None self.locator: Optional[Union[int, str]] = None self._source: Optional[Any] = None # only set for custom source objects + self.RAW_CONFIG: Dict[str, Any] = {} self.CONFIG: Dict[str, Any] = {} self.PROJECT_ROOT: Path = Path(__file__).resolve().parent.parent.parent @@ -620,6 +632,13 @@ def ensure_path(self) -> Optional[str]: if src is None: return None if hasattr(src, 'ensure_path'): + raw = src.load() + effective = resolve_profile_config(raw) + if effective != raw: + try: + return src.ensure_path(effective) + except TypeError: + pass return src.ensure_path() return None @@ -629,9 +648,11 @@ def reload(self) -> None: self.source_type = getattr(src, 'source_type', None) if src else None cfg: Dict[str, Any] = {} try: - cfg = src.load() if src else {} + raw_cfg = src.load() if src else {} except Exception: - cfg = {} + raw_cfg = {} + cfg = resolve_profile_config(raw_cfg) + self.RAW_CONFIG = raw_cfg self.CONFIG = cfg # Core diff --git a/src/dashpva/utils/__init__.py b/src/dashpva/utils/__init__.py index 451cd122..d9cc0252 100644 --- a/src/dashpva/utils/__init__.py +++ b/src/dashpva/utils/__init__.py @@ -9,6 +9,12 @@ "rotation_cycle": "dashpva.utils.generators", "DashAnalysis": "dashpva.utils.dash_analysis", "RSMConverter": "dashpva.utils.rsm_converter", + "RotationAxis": "dashpva.utils.rsm_geometry", + "DetectorModel": "dashpva.utils.rsm_geometry", + "RSMGeometry": "dashpva.utils.rsm_geometry", + "BuiltRSMGeometry": "dashpva.utils.rsm_geometry", + "build_hxrd": "dashpva.utils.rsm_geometry", + "calculate_q": "dashpva.utils.rsm_geometry", "MaskManager": "dashpva.utils.mask_manager", } diff --git a/src/dashpva/utils/config/__init__.py b/src/dashpva/utils/config/__init__.py index f8668b3f..7b6e6194 100644 --- a/src/dashpva/utils/config/__init__.py +++ b/src/dashpva/utils/config/__init__.py @@ -1,12 +1,14 @@ # Copyright (C) UChicago Argonne, LLC # See LICENSE file for details -""" -Configuration repository and sources for DashPVA settings. +"""Configuration loading, resolution, and persistence for DashPVA.""" -This package provides a consistent abstraction for loading/saving configuration -data from different backends (TOML files and database profiles). -""" +from .resolver import resolve_profile_config +from .source import ConfigSaveResult, ConfigSaveStatus, ConfigSource, ConfigSourceError -from .source import ConfigSource - -__all__ = ["ConfigSource"] +__all__ = [ + "ConfigSaveResult", + "ConfigSaveStatus", + "ConfigSource", + "ConfigSourceError", + "resolve_profile_config", +] diff --git a/src/dashpva/utils/config/resolver.py b/src/dashpva/utils/config/resolver.py new file mode 100644 index 00000000..b6777619 --- /dev/null +++ b/src/dashpva/utils/config/resolver.py @@ -0,0 +1,175 @@ +# Copyright (C) UChicago Argonne, LLC +# See LICENSE file for details +"""Resolve persisted beamline profiles into runtime configuration.""" + +from __future__ import annotations + +import copy +import re +from typing import Any, Mapping + +_AXIS_SECTION = re.compile(r"^(?:SAMPLE|DETECTOR)_CIRCLE_AXIS_\d+$") +_RECORD_NAME = re.compile(r"[A-Za-z_][A-Za-z0-9_.-]*") +_AXIS_MANAGED_FIELDS = { + "ANGLE_UNITS", + "AXIS_NUMBER", + "DIRECTION_AXIS", + "POSITION", + "SPEC_MOTOR_NAME", +} + + +def _normalized_prefix(config: Mapping[str, Any]) -> str: + prefix = config.get("IOC_PREFIX", "") + if not isinstance(prefix, str): + raise ValueError("IOC_PREFIX must be a string") + prefix = prefix.strip() + if prefix and not prefix.endswith(":"): + prefix += ":" + return prefix + + +def _pv(prefix: str, suffix: str) -> str: + return f"{prefix}{suffix}" + + +def _managed_section( + hkl: dict[str, Any], + name: str, + values: Mapping[str, str], + aliases: tuple[str, ...] = (), +) -> None: + section: dict[str, Any] = {} + for candidate in (name, *aliases): + existing = hkl.pop(candidate, None) + if isinstance(existing, dict): + section.update(existing) + section.update(values) + hkl[name] = section + + +def _axis_mapping(prefix: str, record_name: str) -> dict[str, str]: + base = _pv(prefix, record_name) + return { + "AXIS_NUMBER": f"{base}:AxisNumber", + "DIRECTION_AXIS": f"{base}:DirectionAxis", + "POSITION": f"{base}:Position", + "SPEC_MOTOR_NAME": f"{base}:SpecMotorName", + } + + +def _canonical_axes(parameters: Mapping[str, Any], key: str) -> list[Mapping[str, Any]]: + axes = parameters.get(key, []) + if axes is None: + return [] + if not isinstance(axes, list) or not all(isinstance(axis, Mapping) for axis in axes): + raise ValueError(f"IOC_RSM_PARAMETER.{key} must be a list of axis tables") + return axes + + +def resolve_profile_config(raw: Mapping[str, Any] | None) -> dict[str, Any]: + """Return a detached runtime view of *raw*. + + Legacy profiles are copied unchanged. A profile containing + ``IOC_RSM_PARAMETER`` gets an effective ``HKL`` section generated from its + canonical ordered axes and IOC prefix. The persisted mapping is never + modified. + """ + if raw is None: + return {} + if not isinstance(raw, Mapping): + raise TypeError("profile configuration must be a mapping") + + effective = copy.deepcopy(dict(raw)) + parameters = raw.get("IOC_RSM_PARAMETER") + if parameters is None: + return effective + if not isinstance(parameters, Mapping): + raise ValueError("IOC_RSM_PARAMETER must be a table") + + prefix = _normalized_prefix(raw) + effective["IOC_PREFIX"] = prefix + hkl = copy.deepcopy(raw.get("HKL", {})) + if not isinstance(hkl, dict): + raise ValueError("HKL must be a table") + + old_axis_sections = { + name: value + for name, value in hkl.items() + if _AXIS_SECTION.fullmatch(name) and isinstance(value, dict) + } + for name in tuple(hkl): + if _AXIS_SECTION.fullmatch(name): + hkl.pop(name) + + seen_records: set[str] = set() + for role, key in (("SAMPLE", "SAMPLE_AXES"), ("DETECTOR", "DETECTOR_AXES")): + for index, axis in enumerate(_canonical_axes(parameters, key), start=1): + record_name = axis.get("RECORD_NAME") + if not isinstance(record_name, str) or not record_name.strip(): + raise ValueError(f"IOC_RSM_PARAMETER.{key}[{index - 1}] needs RECORD_NAME") + record_name = record_name.strip() + if _RECORD_NAME.fullmatch(record_name) is None: + raise ValueError( + "RECORD_NAME must be an unprefixed record stem containing " + f"letters, digits, '_', '.', or '-': {record_name!r}" + ) + if record_name in seen_records: + raise ValueError(f"duplicate RECORD_NAME across RSM axes: {record_name!r}") + seen_records.add(record_name) + + section_name = f"{role}_CIRCLE_AXIS_{index}" + extension = { + name: value + for name, value in old_axis_sections.get(section_name, {}).items() + if name not in _AXIS_MANAGED_FIELDS + } + extension.update(_axis_mapping(prefix, record_name)) + hkl[section_name] = extension + + spec = { + "ENERGY_VALUE": _pv(prefix, "spec:Energy:Value"), + "UB_MATRIX_VALUE": _pv(prefix, "spec:UB_matrix:Value"), + } + if parameters.get("ENERGY_UNITS") is not None: + spec["ENERGY_UNITS"] = _pv(prefix, "spec:Energy:Units") + _managed_section(hkl, "SPEC", spec) + + for name, record, aliases in ( + ("PRIMARY_BEAM_DIRECTION", "PrimaryBeamDirection", ()), + ( + "INPLANE_REFERENCE_DIRECITON", + "InplaneReferenceDirection", + ("INPLANE_REFERENCE_DIRECTION",), + ), + ( + "SAMPLE_SURFACE_NORMAL_DIRECITON", + "SampleSurfaceNormalDirection", + ("SAMPLE_SURFACE_NORMAL_DIRECTION",), + ), + ): + _managed_section( + hkl, + name, + { + f"AXIS_NUMBER_{index}": _pv(prefix, f"{record}:AxisNumber{index}") + for index in range(1, 4) + }, + aliases, + ) + + _managed_section( + hkl, + "DETECTOR_SETUP", + { + "CENTER_CHANNEL_PIXEL": _pv(prefix, "DetectorSetup:CenterChannelPixel"), + "DISTANCE": _pv(prefix, "DetectorSetup:Distance"), + "PIXEL_DIRECTION_1": _pv(prefix, "DetectorSetup:PixelDirection1"), + "PIXEL_DIRECTION_2": _pv(prefix, "DetectorSetup:PixelDirection2"), + "SIZE": _pv(prefix, "DetectorSetup:Size"), + "UNITS": _pv(prefix, "DetectorSetup:Units"), + }, + ) + + effective["HKL"] = hkl + return effective diff --git a/src/dashpva/utils/config/revision.py b/src/dashpva/utils/config/revision.py new file mode 100644 index 00000000..c0bc0c6e --- /dev/null +++ b/src/dashpva/utils/config/revision.py @@ -0,0 +1,20 @@ +# Copyright (C) UChicago Argonne, LLC +# See LICENSE file for details +"""Stable revisions for configuration compare-and-swap operations.""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any, Mapping + + +def mapping_revision(config: Mapping[str, Any]) -> str: + """Return a deterministic revision for a TOML-shaped mapping.""" + payload = json.dumps( + config, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + return hashlib.sha256(payload).hexdigest() diff --git a/src/dashpva/utils/config/source.py b/src/dashpva/utils/config/source.py index c7611975..ffb5f646 100644 --- a/src/dashpva/utils/config/source.py +++ b/src/dashpva/utils/config/source.py @@ -23,12 +23,122 @@ from __future__ import annotations +import contextlib +import hashlib import os +import stat import tempfile +import threading +import time +from dataclasses import dataclass +from enum import Enum +from pathlib import Path from typing import Any, Dict, Optional, Union import toml +from dashpva.utils.config.revision import mapping_revision + +try: + import fcntl +except ImportError: # pragma: no cover - Windows fallback + fcntl = None + +try: + import msvcrt +except ImportError: # pragma: no cover - POSIX + msvcrt = None + +_LOCAL_LOCK = threading.RLock() + + +class ConfigSourceError(RuntimeError): + """A strict configuration read or write could not be completed.""" + + +class ConfigSaveStatus(str, Enum): + """Outcome of a strict compare-and-swap save.""" + + SAVED = "saved" + CONFLICT = "conflict" + ERROR = "error" + + +@dataclass(frozen=True) +class ConfigSaveResult: + """Result returned by :meth:`ConfigSource.replace_if_revision`.""" + + status: ConfigSaveStatus + revision: Optional[str] = None + error: Optional[str] = None + + @property + def saved(self) -> bool: + return self.status is ConfigSaveStatus.SAVED + + +def _bytes_revision(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +@contextlib.contextmanager +def _config_lock(path: Path, *, exclusive: bool): + lock_path = path.with_name(f".{path.name}.lock") + with _LOCAL_LOCK: + with lock_path.open("a+b") as lock_file: + if fcntl is not None: + mode = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH + fcntl.flock(lock_file.fileno(), mode) + elif msvcrt is not None: # pragma: no cover - Windows + lock_file.seek(0, os.SEEK_END) + if lock_file.tell() == 0: + lock_file.write(b"\0") + lock_file.flush() + mode = msvcrt.LK_LOCK if exclusive else msvcrt.LK_RLCK + while True: + try: + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), mode, 1) + break + except OSError: + time.sleep(0.05) + try: + yield + finally: + if fcntl is not None: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + elif msvcrt is not None: # pragma: no cover - Windows + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1) + + +def _read_toml_bytes(path: Path) -> bytes: + try: + return path.read_bytes() + except FileNotFoundError: + return b"" + + +def _parse_toml(payload: bytes, path: Path) -> Dict[str, Any]: + try: + return toml.loads(payload.decode("utf-8")) if payload else {} + except Exception as exc: + raise ConfigSourceError(f"could not parse configuration {path}: {exc}") from exc + + +def _write_temp_config(config: Dict[str, Any]) -> str: + fd, tmp = tempfile.mkstemp(suffix=".toml", prefix="dashpva_") + try: + with os.fdopen(fd, "w", encoding="utf-8") as stream: + toml.dump(config, stream) + stream.flush() + os.fsync(stream.fileno()) + except Exception: + with contextlib.suppress(OSError): + os.unlink(tmp) + raise + return tmp + # --------------------------------------------------------------------------- # Concrete source backends # --------------------------------------------------------------------------- @@ -43,19 +153,91 @@ def __init__(self, path: str) -> None: def load(self) -> Dict[str, Any]: try: - return toml.load(self.path) + config, _ = self.load_snapshot() + return config except Exception: return {} - def save(self, update: Dict[str, Any]) -> bool: + def load_snapshot(self) -> tuple[Dict[str, Any], str]: + path = Path(self.path) + try: + payload = _read_toml_bytes(path) + return _parse_toml(payload, path), _bytes_revision(payload) + except ConfigSourceError: + raise + except Exception as exc: + raise ConfigSourceError(f"could not read configuration {path}: {exc}") from exc + + def replace_if_revision( + self, + full_config: Dict[str, Any], + revision: str, + ) -> ConfigSaveResult: + path = Path(self.path) + if not isinstance(full_config, dict): + return ConfigSaveResult(ConfigSaveStatus.ERROR, error="configuration must be a dict") + try: + replacement = toml.dumps(full_config).encode("utf-8") + except Exception as exc: + return ConfigSaveResult(ConfigSaveStatus.ERROR, error=f"could not encode TOML: {exc}") + + tmp_path: Optional[Path] = None try: - existing = self.load() + path.parent.mkdir(parents=True, exist_ok=True) + with _config_lock(path, exclusive=True): + current = _read_toml_bytes(path) + if _bytes_revision(current) != revision: + return ConfigSaveResult(ConfigSaveStatus.CONFLICT) + + fd, tmp_name = tempfile.mkstemp( + dir=str(path.parent), + prefix=f".{path.name}.", + suffix=".tmp", + ) + tmp_path = Path(tmp_name) + if path.exists(): + try: + os.fchmod(fd, stat.S_IMODE(path.stat().st_mode)) + except (AttributeError, OSError): + pass + with os.fdopen(fd, "wb") as stream: + stream.write(replacement) + stream.flush() + os.fsync(stream.fileno()) + os.replace(tmp_path, path) + tmp_path = None + try: + directory_fd = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + except OSError: + pass + return ConfigSaveResult( + ConfigSaveStatus.SAVED, + revision=_bytes_revision(replacement), + ) + except Exception as exc: + return ConfigSaveResult(ConfigSaveStatus.ERROR, error=str(exc)) + finally: + if tmp_path is not None: + with contextlib.suppress(OSError): + tmp_path.unlink() + + def save(self, update: Dict[str, Any]) -> bool: + for _ in range(3): + try: + existing, revision = self.load_snapshot() + except ConfigSourceError: + return False existing.update(update or {}) - with open(self.path, "w") as f: - toml.dump(existing, f) - return True - except Exception: - return False + result = self.replace_if_revision(existing, revision) + if result.status is ConfigSaveStatus.SAVED: + return True + if result.status is ConfigSaveStatus.ERROR: + return False + return False class DbProfileConfigSource: @@ -97,16 +279,64 @@ def load(self) -> Dict[str, Any]: except Exception: return {} - def save(self, update: Dict[str, Any]) -> bool: + def load_snapshot(self) -> tuple[Dict[str, Any], str]: + profile_id = self._resolve_profile_id() + if profile_id is None or self.db is None: + raise ConfigSourceError("database profile could not be resolved") + try: + if hasattr(self.db, "load_profile_toml_strict"): + config = self.db.load_profile_toml_strict(profile_id) + else: + profile = self.db.get_profile_by_id(profile_id) + if profile is None: + raise ConfigSourceError(f"database profile {profile_id} does not exist") + config = self.db.export_profile_to_toml(profile_id) or {} + return config, mapping_revision(config) + except ConfigSourceError: + raise + except Exception as exc: + raise ConfigSourceError(f"could not read database profile {profile_id}: {exc}") from exc + + def replace_if_revision( + self, + full_config: Dict[str, Any], + revision: str, + ) -> ConfigSaveResult: profile_id = self._resolve_profile_id() if profile_id is None or self.db is None: - return False + return ConfigSaveResult(ConfigSaveStatus.ERROR, error="database profile could not be resolved") + if not isinstance(full_config, dict): + return ConfigSaveResult(ConfigSaveStatus.ERROR, error="configuration must be a dict") try: - existing = self.db.export_profile_to_toml(profile_id) or {} + status = self.db.replace_profile_toml_if_revision( + profile_id, + full_config, + revision, + ) + except Exception as exc: + return ConfigSaveResult(ConfigSaveStatus.ERROR, error=str(exc)) + if status == ConfigSaveStatus.SAVED.value: + return ConfigSaveResult( + ConfigSaveStatus.SAVED, + revision=mapping_revision(full_config), + ) + if status == ConfigSaveStatus.CONFLICT.value: + return ConfigSaveResult(ConfigSaveStatus.CONFLICT) + return ConfigSaveResult(ConfigSaveStatus.ERROR, error="database replacement failed") + + def save(self, update: Dict[str, Any]) -> bool: + for _ in range(3): + try: + existing, revision = self.load_snapshot() + except ConfigSourceError: + return False existing.update(update or {}) - return self.db.import_toml_to_profile(profile_id, existing) - except Exception: - return False + result = self.replace_if_revision(existing, revision) + if result.status is ConfigSaveStatus.SAVED: + return True + if result.status is ConfigSaveStatus.ERROR: + return False + return False # --------------------------------------------------------------------------- @@ -193,29 +423,45 @@ def load(self) -> Dict[str, Any]: return self._backend.load() return {} + def load_snapshot(self) -> tuple[Dict[str, Any], str]: + """Return raw configuration and an opaque compare-and-swap revision.""" + if self._backend is None: + raise ConfigSourceError("no configuration source is available") + return self._backend.load_snapshot() + + def replace_if_revision( + self, + full_config: Dict[str, Any], + revision: str, + ) -> ConfigSaveResult: + """Replace the raw profile only when *revision* is still current.""" + if self._backend is None: + return ConfigSaveResult(ConfigSaveStatus.ERROR, error="no configuration source is available") + return self._backend.replace_if_revision(full_config, revision) + def save(self, update: Dict[str, Any]) -> bool: """Persist an updated configuration dict back to the current source.""" if self._backend is not None: return self._backend.save(update) return False - def ensure_path(self) -> Optional[str]: + def ensure_path(self, config: Optional[Dict[str, Any]] = None) -> Optional[str]: """Return a file path to a TOML representation of the config. - TOML source: returns the original file path (already on disk). - DB source: exports the config to a temporary TOML file and returns that path. - None source: returns None. """ + if config is not None: + try: + return _write_temp_config(config) + except Exception: + return None if self.source_type == "toml": return self._backend.path if self.source_type == "db": try: - data = self.load() - fd, tmp = tempfile.mkstemp(suffix=".toml", prefix="dashpva_") - os.close(fd) - with open(tmp, "w") as f: - toml.dump(data, f) - return tmp + return _write_temp_config(self.load()) except Exception: return None return None diff --git a/src/dashpva/utils/rsm_converter.py b/src/dashpva/utils/rsm_converter.py index 4995b811..b426408d 100644 --- a/src/dashpva/utils/rsm_converter.py +++ b/src/dashpva/utils/rsm_converter.py @@ -1,13 +1,20 @@ # Copyright (C) UChicago Argonne, LLC # See LICENSE file for details from dataclasses import dataclass -from typing import List, Optional, Tuple +from typing import List, Optional, Sequence, Tuple import h5py import numpy as np -import xrayutilities as xu import dashpva.settings as app_settings +from dashpva.utils.rsm_geometry import ( + BuiltRSMGeometry, + DetectorModel, + RotationAxis, + RSMGeometry, + build_hxrd, + calculate_q, +) """Utilities for converting detector frames into reciprocal space (RSM). This module provides a concise RSMConverter focused on the essential @@ -26,11 +33,21 @@ class FileGeometry: frame the way create_rsm() does (kept as-is there for backward compatibility with existing single-frame callers). """ - hxrd: "xu.HXRD" - ub: np.ndarray - energy_eV: float + built: BuiltRSMGeometry shape: Tuple[int, ...] + @property + def hxrd(self): + return self.built.hxrd + + @property + def ub(self) -> np.ndarray: + return self.built.ub + + @property + def energy_eV(self) -> float: + return self.built.energy_eV + class Data: """Simple container for 3D points and intensities.""" @@ -77,26 +94,13 @@ def load_h5_to_3d(self, filename: str): def create_rsm(self, filename: str, frame: int): """Create reciprocal space mapping for a single frame using xrayutilities.""" - try: - with h5py.File(filename, "r") as f: - shape = f["entry/data/data"].shape - sc_dir, sc_pos, dc_dir, dc_pos = self.get_sample_and_detector_circles(f, frame) - primary, inplane, surface, ub, energy = self.get_physics_params(f) - qconv = xu.experiment.QConversion(sc_dir, dc_dir, primary) - hxrd = xu.HXRD(inplane, surface, en=energy, qconv=qconv) - p_dir1, p_dir2, cch1, cch2, nch1, nch2, pw1, pw2, dist, roi = self.get_detector_setup(f, shape) - hxrd.Ang2Q.init_area( - p_dir1, p_dir2, - cch1=cch1, cch2=cch2, - Nch1=nch1, Nch2=nch2, - pwidth1=pw1, pwidth2=pw2, - distance=dist, - roi=roi, - ) - angles = [*sc_pos, *dc_pos] - return hxrd.Ang2Q.area(*angles, UB=ub) - except Exception: - raise + with h5py.File(filename, "r") as h5_file: + shape = h5_file["entry/data/data"].shape + sc_dir, sc_pos, dc_dir, dc_pos = self.get_sample_and_detector_circles( + h5_file, frame + ) + model = self._build_geometry_model(h5_file, shape, sc_dir, dc_dir) + return calculate_q(build_hxrd(model), sc_pos, dc_pos) def get_q_points(self, filename: str) -> np.ndarray: """Compute Q points for all frames and return flattened (N, 3) array.""" @@ -309,19 +313,49 @@ def build_file_geometry(self, h5_file: h5py.File) -> FileGeometry: FileGeometry across every frame/batch of this file via q_for_frames.""" shape = h5_file["entry/data/data"].shape sc_dir, dc_dir = self.get_circle_directions(h5_file) + model = self._build_geometry_model(h5_file, shape, sc_dir, dc_dir) + return FileGeometry(built=build_hxrd(model), shape=shape) + + def _build_geometry_model( + self, + h5_file: h5py.File, + shape: tuple, + sample_directions: Sequence[str], + detector_directions: Sequence[str], + ) -> RSMGeometry: + """Translate legacy file metadata into the canonical shared model.""" primary, inplane, surface, ub, energy = self.get_physics_params(h5_file) - qconv = xu.experiment.QConversion(sc_dir, dc_dir, primary) - hxrd = xu.HXRD(inplane, surface, en=energy, qconv=qconv) p_dir1, p_dir2, cch1, cch2, nch1, nch2, pw1, pw2, dist, roi = self.get_detector_setup(h5_file, shape) - hxrd.Ang2Q.init_area( - p_dir1, p_dir2, - cch1=cch1, cch2=cch2, - Nch1=nch1, Nch2=nch2, - pwidth1=pw1, pwidth2=pw2, + detector = DetectorModel( + pixel_direction_1=p_dir1, + pixel_direction_2=p_dir2, + center_channel=(cch1, cch2), + shape=(nch1, nch2), + pixel_width=(pw1, pw2), distance=dist, - roi=roi, + roi=tuple(roi), + ) + orientation_path = "entry/data/metadata/HKL/SAMPLE_ORIENTATION" + sample_orientation = ( + self._static_str(h5_file[orientation_path], "sample orientation") + if orientation_path in h5_file + else "det" + ) + return RSMGeometry( + sample_axes=tuple( + RotationAxis("sample", direction) for direction in sample_directions + ), + detector_axes=tuple( + RotationAxis("detector", direction) for direction in detector_directions + ), + primary_beam_direction=primary, + inplane_reference_direction=inplane, + sample_surface_normal_direction=surface, + energy_eV=energy, + ub_matrix=ub, + detector=detector, + sample_orientation=sample_orientation, ) - return FileGeometry(hxrd=hxrd, ub=ub, energy_eV=energy, shape=shape) def q_for_frames(self, geom: FileGeometry, h5_file: h5py.File, start: int, stop: int): """Angles -> Q for frames [start, stop) via a single batched @@ -330,8 +364,7 @@ def q_for_frames(self, geom: FileGeometry, h5_file: h5py.File, start: int, stop: masked pixels before raveling. """ sc_pos, dc_pos = self.get_circle_positions_batch(h5_file, start, stop) - angles = [*sc_pos, *dc_pos] - qx, qy, qz = geom.hxrd.Ang2Q.area(*angles, UB=geom.ub) + qx, qy, qz = calculate_q(geom.built, sc_pos, dc_pos) n = stop - start if n == 1 and qx.ndim == 2: # xrayutilities' Ang2Q.area drops the leading batch axis when diff --git a/src/dashpva/utils/rsm_geometry.py b/src/dashpva/utils/rsm_geometry.py new file mode 100644 index 00000000..8b4543b0 --- /dev/null +++ b/src/dashpva/utils/rsm_geometry.py @@ -0,0 +1,347 @@ +# Copyright (C) UChicago Argonne, LLC +# See LICENSE file for details +"""Shared xrayutilities geometry construction for offline and live RSM.""" + +from __future__ import annotations + +import re +import warnings +from dataclasses import dataclass +from typing import Literal, Mapping, Sequence + +import numpy as np +import xrayutilities as xu + +AxisRole = Literal["sample", "detector"] + +_SAMPLE_DIRECTION = re.compile(r"[xyzk][+-]") +_DETECTOR_DIRECTION = re.compile(r"[xyz][+-]") +_EXPLICIT_SAMPLE_ORIENTATION = re.compile(r"[xyz][+-]") + + +def _vector3(value: Sequence[float], label: str) -> tuple[float, float, float]: + vector = np.asarray(value, dtype=float) + if vector.shape != (3,): + raise ValueError(f"{label} must contain exactly three values.") + if not np.isfinite(vector).all() or np.linalg.norm(vector) == 0: + raise ValueError(f"{label} must be finite and non-zero.") + return tuple(float(component) for component in vector) + + +def _matrix3(value: Sequence[Sequence[float]], label: str) -> tuple[tuple[float, ...], ...]: + matrix = np.asarray(value, dtype=float) + if matrix.shape != (3, 3): + raise ValueError(f"{label} must have shape (3, 3).") + if not np.isfinite(matrix).all() or np.linalg.matrix_rank(matrix) < 3: + raise ValueError(f"{label} must be finite and full rank.") + return tuple(tuple(float(component) for component in row) for row in matrix) + + +def _direction_vector(direction: str) -> np.ndarray: + return np.asarray(xu.math.getVector(direction), dtype=float) + + +def _parallel(left: Sequence[float], right: Sequence[float]) -> bool: + return bool(np.isclose(np.linalg.norm(np.cross(left, right)), 0.0)) + + +@dataclass(frozen=True, slots=True) +class RotationAxis: + """One ordered sample or detector rotation axis. + + ``role`` comes from the profile list containing the axis. ``record_name`` + is its stable machine identity; ``label`` remains human-editable. + """ + + role: AxisRole + direction: str + label: str = "" + record_name: str = "" + source_pv: str = "" + angle_units: str = "deg" + + def __post_init__(self) -> None: + role = str(self.role).strip().lower() + if role not in ("sample", "detector"): + raise ValueError(f"Axis role must be 'sample' or 'detector', got {self.role!r}.") + object.__setattr__(self, "role", role) + + direction = str(self.direction).strip().lower() + pattern = _SAMPLE_DIRECTION if role == "sample" else _DETECTOR_DIRECTION + if pattern.fullmatch(direction) is None: + allowed = "[xyzk][+-]" if role == "sample" else "[xyz][+-]" + raise ValueError( + f"Invalid {role} rotation direction {self.direction!r}; expected {allowed}." + ) + object.__setattr__(self, "direction", direction) + + units = str(self.angle_units).strip().lower() + if units not in ("deg", "degree", "degrees"): + raise ValueError( + f"Axis {self.record_name or self.label or direction!r} uses unsupported " + f"angle units {self.angle_units!r}; PR 1 geometry requires degrees." + ) + object.__setattr__(self, "angle_units", "deg") + + @classmethod + def from_mapping(cls, role: AxisRole, values: Mapping[str, object]) -> RotationAxis: + """Build an axis from one canonical ``IOC_RSM_PARAMETER`` list item.""" + try: + direction = str(values["DIRECTION"]) + except KeyError as exc: + raise ValueError("Canonical rotation axis is missing DIRECTION.") from exc + return cls( + role=role, + direction=direction, + label=str(values.get("LABEL", "")), + record_name=str(values.get("RECORD_NAME", "")), + source_pv=str(values.get("SOURCE_PV", "")), + angle_units=str(values.get("ANGLE_UNITS", "deg")), + ) + + +@dataclass(frozen=True, slots=True) +class DetectorModel: + """Legacy area-detector calibration used by the shared PR 1 builder.""" + + pixel_direction_1: str + pixel_direction_2: str + center_channel: tuple[float, float] + shape: tuple[int, int] + pixel_width: tuple[float, float] + distance: float + roi: tuple[int, int, int, int] | None = None + + def __post_init__(self) -> None: + for field_name in ("pixel_direction_1", "pixel_direction_2"): + value = str(getattr(self, field_name)).strip().lower() + if _DETECTOR_DIRECTION.fullmatch(value) is None: + raise ValueError(f"{field_name} must use [xyz][+-] syntax, got {value!r}.") + object.__setattr__(self, field_name, value) + + center = tuple(float(value) for value in self.center_channel) + if len(center) != 2 or not np.isfinite(center).all(): + raise ValueError("Detector center_channel must contain two finite values.") + object.__setattr__(self, "center_channel", center) + + shape = tuple(int(value) for value in self.shape) + if len(shape) != 2 or any(value <= 0 for value in shape): + raise ValueError("Detector shape must contain two positive integers.") + object.__setattr__(self, "shape", shape) + + pixel_width = tuple(float(value) for value in self.pixel_width) + if ( + len(pixel_width) != 2 + or not np.isfinite(pixel_width).all() + or any(value <= 0 for value in pixel_width) + ): + raise ValueError("Detector pixel_width must contain two finite positive values.") + object.__setattr__(self, "pixel_width", pixel_width) + + distance = float(self.distance) + if not np.isfinite(distance) or distance <= 0: + raise ValueError("Detector distance must be finite and positive.") + object.__setattr__(self, "distance", distance) + + roi = self.roi if self.roi is not None else (0, shape[0], 0, shape[1]) + roi = tuple(int(value) for value in roi) + if len(roi) != 4: + raise ValueError("Detector roi must contain four integer bounds.") + if not (0 <= roi[0] < roi[1] <= shape[0] and 0 <= roi[2] < roi[3] <= shape[1]): + raise ValueError(f"Detector roi {roi!r} falls outside detector shape {shape!r}.") + object.__setattr__(self, "roi", roi) + + +@dataclass(frozen=True, slots=True) +class RSMGeometry: + """Frame-invariant diffraction geometry with ordered rotation axes.""" + + sample_axes: tuple[RotationAxis, ...] + detector_axes: tuple[RotationAxis, ...] + primary_beam_direction: tuple[float, float, float] + inplane_reference_direction: tuple[float, float, float] + sample_surface_normal_direction: tuple[float, float, float] + energy_eV: float + ub_matrix: tuple[tuple[float, ...], ...] + detector: DetectorModel + sample_orientation: str = "det" + + def __post_init__(self) -> None: + sample_axes = tuple(self.sample_axes) + detector_axes = tuple(self.detector_axes) + if any(not isinstance(axis, RotationAxis) or axis.role != "sample" for axis in sample_axes): + raise ValueError("sample_axes may contain only sample RotationAxis values.") + if any( + not isinstance(axis, RotationAxis) or axis.role != "detector" + for axis in detector_axes + ): + raise ValueError("detector_axes may contain only detector RotationAxis values.") + object.__setattr__(self, "sample_axes", sample_axes) + object.__setattr__(self, "detector_axes", detector_axes) + + primary = _vector3(self.primary_beam_direction, "Primary beam direction") + inplane = _vector3(self.inplane_reference_direction, "In-plane reference direction") + surface = _vector3( + self.sample_surface_normal_direction, "Sample surface-normal direction" + ) + if _parallel(inplane, surface): + raise ValueError( + "In-plane reference and sample surface-normal directions must be independent." + ) + cosine = np.dot(inplane, surface) / ( + np.linalg.norm(inplane) * np.linalg.norm(surface) + ) + if not np.isclose(cosine, 0.0): + raise ValueError( + "In-plane reference and sample surface-normal directions must be " + "perpendicular; xrayutilities would otherwise adjust the in-plane vector." + ) + object.__setattr__(self, "primary_beam_direction", primary) + object.__setattr__(self, "inplane_reference_direction", inplane) + object.__setattr__(self, "sample_surface_normal_direction", surface) + + energy = float(self.energy_eV) + if not np.isfinite(energy) or energy <= 0: + raise ValueError("Photon energy must be finite and positive in eV.") + object.__setattr__(self, "energy_eV", energy) + object.__setattr__(self, "ub_matrix", _matrix3(self.ub_matrix, "UB matrix")) + + orientation = str(self.sample_orientation).strip().lower() + validate_sample_orientation(sample_axes, detector_axes, primary, orientation) + object.__setattr__(self, "sample_orientation", orientation) + + if not isinstance(self.detector, DetectorModel): + raise TypeError("detector must be a DetectorModel.") + + +@dataclass(frozen=True, slots=True) +class BuiltRSMGeometry: + """A validated geometry and its initialized xrayutilities object.""" + + model: RSMGeometry + hxrd: xu.HXRD + + @property + def ub(self) -> np.ndarray: + return np.asarray(self.model.ub_matrix, dtype=float) + + @property + def energy_eV(self) -> float: + return self.model.energy_eV + + @property + def shape(self) -> tuple[int, int]: + return self.model.detector.shape + + +def validate_sample_orientation( + sample_axes: Sequence[RotationAxis], + detector_axes: Sequence[RotationAxis], + primary_beam_direction: Sequence[float], + sample_orientation: str, + *, + warn_for_sample_axis: bool = True, +) -> None: + """Reject cases xrayutilities would fail or silently reinterpret.""" + primary = np.asarray(primary_beam_direction, dtype=float) + orientation = str(sample_orientation).strip().lower() + + if orientation == "det": + if not detector_axes: + raise ValueError("SAMPLE_ORIENTATION='det' requires a detector rotation axis.") + innermost = _direction_vector(detector_axes[-1].direction) + if _parallel(innermost, primary): + if len(detector_axes) < 2 or _parallel( + _direction_vector(detector_axes[-2].direction), primary + ): + raise ValueError( + "SAMPLE_ORIENTATION='det' requires an innermost detector rotation " + "not parallel to the primary beam (a final beam-axis rotation is ignored)." + ) + return + + if orientation == "sam": + if not sample_axes: + raise ValueError("SAMPLE_ORIENTATION='sam' requires a sample rotation axis.") + if _parallel(_direction_vector(sample_axes[-1].direction), primary): + raise ValueError( + "SAMPLE_ORIENTATION='sam' requires the innermost sample axis not to be " + "parallel to the primary beam." + ) + if warn_for_sample_axis: + warnings.warn( + "SAMPLE_ORIENTATION='sam' is physically correct only when the innermost " + "sample circle is the azimuth motor.", + UserWarning, + stacklevel=2, + ) + return + + if _EXPLICIT_SAMPLE_ORIENTATION.fullmatch(orientation) is None: + raise ValueError( + "SAMPLE_ORIENTATION must be 'det', 'sam', or explicit [xyz][+-] syntax." + ) + if _parallel(_direction_vector(orientation), primary): + raise ValueError( + f"Explicit SAMPLE_ORIENTATION={orientation!r} is parallel to the primary " + "beam; xrayutilities would silently substitute a different axis." + ) + + +def build_hxrd(model: RSMGeometry) -> BuiltRSMGeometry: + """Validate and initialize one xrayutilities area-detector geometry.""" + q_conversion = xu.experiment.QConversion( + [axis.direction for axis in model.sample_axes], + [axis.direction for axis in model.detector_axes], + model.primary_beam_direction, + ) + hxrd = xu.HXRD( + model.inplane_reference_direction, + model.sample_surface_normal_direction, + en=model.energy_eV, + qconv=q_conversion, + sampleor=model.sample_orientation, + ) + detector = model.detector + hxrd.Ang2Q.init_area( + detector.pixel_direction_1, + detector.pixel_direction_2, + cch1=detector.center_channel[0], + cch2=detector.center_channel[1], + Nch1=detector.shape[0], + Nch2=detector.shape[1], + pwidth1=detector.pixel_width[0], + pwidth2=detector.pixel_width[1], + distance=detector.distance, + roi=list(detector.roi), + ) + return BuiltRSMGeometry(model=model, hxrd=hxrd) + + +def calculate_q( + geometry: BuiltRSMGeometry, + sample_angles: Sequence[object], + detector_angles: Sequence[object], + *, + ub_matrix: Sequence[Sequence[float]] | None = None, +) -> tuple[np.ndarray, np.ndarray, np.ndarray]: + """Convert ordered sample/detector angles with an initialized geometry.""" + if len(sample_angles) != len(geometry.model.sample_axes): + raise ValueError( + f"Expected {len(geometry.model.sample_axes)} sample angles, " + f"received {len(sample_angles)}." + ) + if len(detector_angles) != len(geometry.model.detector_axes): + raise ValueError( + f"Expected {len(geometry.model.detector_axes)} detector angles, " + f"received {len(detector_angles)}." + ) + ub = geometry.ub if ub_matrix is None else np.asarray( + _matrix3(ub_matrix, "UB matrix"), dtype=float + ) + return geometry.hxrd.Ang2Q.area( + *sample_angles, + *detector_angles, + UB=ub, + deg=True, + ) diff --git a/src/dashpva/viewer/area_det/area_det_viewer.py b/src/dashpva/viewer/area_det/area_det_viewer.py index ede07ade..12c591cb 100644 --- a/src/dashpva/viewer/area_det/area_det_viewer.py +++ b/src/dashpva/viewer/area_det/area_det_viewer.py @@ -8,7 +8,6 @@ import numpy as np import pyqtgraph as pg -import xrayutilities as xu from epics import PV, ca, caget, camonitor from PyQt5 import uic from PyQt5.QtCore import ( @@ -44,6 +43,14 @@ from dashpva.utils import HDF5Writer, PVAReader, rotation_cycle from dashpva.utils.mask_manager import MaskManager from dashpva.utils.roi_ops import _extract_roi_subarray +from dashpva.utils.rsm_geometry import ( + DetectorModel, + RotationAxis, + RSMGeometry, + build_hxrd, + calculate_q, + validate_sample_orientation, +) from dashpva.viewer.area_det.docks import ( AnalysisDock, BeamFitDock, @@ -252,7 +259,7 @@ def __init__(self, input_channel='pvapy:image'): self.hkl_config = None self.hkl_pvs = {} self.hkl_data = {} - self.q_conv = None + self.rsm_geometry_ready = False self.qx = None self.qy = None self.qz = None @@ -1893,7 +1900,7 @@ def handle_hkl_data_update(self): if self.reader is not None and not self.stop_hkl.isChecked() and self.hkl_data: try: self.hkl_setup() - if self.q_conv is not None: + if self.rsm_geometry_ready: self.update_rsm() except Exception as e: print(f'[DashPVA] HKL update failed (will retry next frame): {e}') @@ -1978,18 +1985,69 @@ def hkl_setup(self) -> None: raise ValueError("Missing energy PV data") self.energy *= 1000 - # Make sure all values are setup correctly before instantiating QConversion - if all([self.sample_circle_directions, self.det_circle_directions, self.primary_beam_directions]): - self.q_conv = xu.experiment.QConversion(self.sample_circle_directions, - self.det_circle_directions, - self.primary_beam_directions) - else: - self.q_conv = None - raise ValueError("QConversion initialization failed due to missing PV data.") + sample_axes = tuple( + RotationAxis('sample', direction) + for direction in self.sample_circle_directions + ) + detector_axes = tuple( + RotationAxis('detector', direction) + for direction in self.det_circle_directions + ) + validate_sample_orientation( + sample_axes, + detector_axes, + self.primary_beam_directions, + self._sample_orientation(), + warn_for_sample_axis=False, + ) + self.rsm_geometry_ready = True except Exception as e: print(f'[Diffraction Image Viewer] Error Setting up HKL: {e}') - self.q_conv = None # Reset to None on failure to prevent invalid calculations + self.rsm_geometry_ready = False + + def _sample_orientation(self) -> str: + canonical = self.reader.config.get('IOC_RSM_PARAMETER', {}) or {} + return canonical.get('SAMPLE_ORIENTATION', 'det') + + def _build_live_rsm_geometry(self): + """Build the shared geometry model from the current monitored values.""" + if self.reader is None or len(self.reader.shape) < 2: + raise ValueError("Detector frame shape is unavailable.") + ds_cfg = self.hkl_config.get('DETECTOR_SETUP', {}) or {} + cch1, cch2 = self.hkl_data[ds_cfg['CENTER_CHANNEL_PIXEL']][:2] + distance = self.hkl_data[ds_cfg['DISTANCE']] + pixel_dir1 = self.hkl_data[ds_cfg['PIXEL_DIRECTION_1']] + pixel_dir2 = self.hkl_data[ds_cfg['PIXEL_DIRECTION_2']] + nch1, nch2 = self.reader.shape[:2] + size_xy = self.hkl_data[ds_cfg['SIZE']] + detector = DetectorModel( + pixel_direction_1=pixel_dir1, + pixel_direction_2=pixel_dir2, + center_channel=(cch1, cch2), + shape=(nch1, nch2), + pixel_width=(size_xy[0] / nch1, size_xy[1] / nch2), + distance=distance, + roi=(0, nch1, 0, nch2), + ) + model = RSMGeometry( + sample_axes=tuple( + RotationAxis('sample', direction) + for direction in self.sample_circle_directions + ), + detector_axes=tuple( + RotationAxis('detector', direction) + for direction in self.det_circle_directions + ), + primary_beam_direction=self.primary_beam_directions, + inplane_reference_direction=self.inplane_reference_directions, + sample_surface_normal_direction=self.sample_surface_normal_directions, + energy_eV=self.energy, + ub_matrix=self.ub_matrix, + detector=detector, + sample_orientation=self._sample_orientation(), + ) + return build_hxrd(model) def create_rsm(self) -> np.ndarray: """ @@ -2012,39 +2070,17 @@ def create_rsm(self) -> np.ndarray: """ if self.reader is not None and self.hkl_data and (not self.stop_hkl.isChecked()): try: - if (self.q_conv is None or + if (not self.rsm_geometry_ready or not self.inplane_reference_directions or not self.sample_surface_normal_directions or self.energy is None): return None - - hxrd = xu.HXRD(self.inplane_reference_directions, - self.sample_surface_normal_directions, - en=self.energy, - qconv=self.q_conv) - - roi = [0, self.reader.shape[0], 0, self.reader.shape[1]] - # Look up by the PV name in the active HKL config so any prefix - # scheme (xidb:, 6idb1:, none) works without code edits — same - # pattern HpcRsmProcessor uses. - ds_cfg = self.hkl_config.get('DETECTOR_SETUP', {}) or {} - cch1, cch2 = self.hkl_data[ds_cfg['CENTER_CHANNEL_PIXEL']][:2] - distance = self.hkl_data[ds_cfg['DISTANCE']] - pixel_dir1 = self.hkl_data[ds_cfg['PIXEL_DIRECTION_1']] - pixel_dir2 = self.hkl_data[ds_cfg['PIXEL_DIRECTION_2']] - nch1 = self.reader.shape[0] # Number of detector pixels along direction 1 - nch2 = self.reader.shape[1] # Number of detector pixels along direction 2 - size_xy = self.hkl_data[ds_cfg['SIZE']] - pixel_width1 = size_xy[0] / nch1 - pixel_width2 = size_xy[1] / nch2 - - hxrd.Ang2Q.init_area(pixel_dir1, pixel_dir2, cch1=cch1, cch2=cch2, - Nch1=nch1, Nch2=nch2, pwidth1=pixel_width1, - pwidth2=pixel_width2, distance=distance, roi=roi) - - angles = [*self.sample_circle_positions, *self.det_circle_positions] - - return hxrd.Ang2Q.area(*angles, UB=self.ub_matrix) + geometry = self._build_live_rsm_geometry() + return calculate_q( + geometry, + self.sample_circle_positions, + self.det_circle_positions, + ) except Exception as e: print(f'[Diffration Image Viewer] Error Creating RSM: {e}') return @@ -2053,6 +2089,7 @@ def create_rsm(self) -> np.ndarray: def reset_rsm_vars(self) -> None: self.hkl_data = {} + self.rsm_geometry_ready = False self.rois.clear() self._roi_overlays.clear() self.qx = None diff --git a/src/dashpva/workflow/workflow.py b/src/dashpva/workflow/workflow.py index 028977fa..a2911362 100644 --- a/src/dashpva/workflow/workflow.py +++ b/src/dashpva/workflow/workflow.py @@ -2667,7 +2667,7 @@ def run_analysis_consumer(self): '-dc' ] - config_path = app_settings.TOML_FILE + config_path = app_settings.ensure_path() if config_path: cmd.extend(['--processor-args', '{"path": "%s"}' % config_path]) diff --git a/tests/unit/test_config_cas.py b/tests/unit/test_config_cas.py new file mode 100644 index 00000000..029a18e8 --- /dev/null +++ b/tests/unit/test_config_cas.py @@ -0,0 +1,143 @@ +# Copyright (C) UChicago Argonne, LLC +# See LICENSE file for details +"""Compare-and-swap contracts for TOML and database profile sources.""" + +import threading + +import toml + +from dashpva.utils.config.source import ( + ConfigSaveStatus, + ConfigSource, + DbProfileConfigSource, +) + + +def test_toml_stale_revision_cannot_overwrite_newer_save(tmp_path): + path = tmp_path / "profile.toml" + path.write_text(toml.dumps({"owner": "initial"})) + source = ConfigSource(str(path)) + _, first_revision = source.load_snapshot() + _, stale_revision = source.load_snapshot() + + first = source.replace_if_revision({"owner": "first"}, first_revision) + stale = source.replace_if_revision({"owner": "stale"}, stale_revision) + + assert first.status is ConfigSaveStatus.SAVED + assert first.revision + assert stale.status is ConfigSaveStatus.CONFLICT + assert toml.load(path) == {"owner": "first"} + + +def test_toml_concurrent_writers_have_exactly_one_winner(tmp_path): + path = tmp_path / "profile.toml" + path.write_text(toml.dumps({"owner": "initial"})) + source = ConfigSource(str(path)) + _, revision = source.load_snapshot() + barrier = threading.Barrier(3) + results = [] + + def write(owner): + barrier.wait() + results.append( + source.replace_if_revision({"owner": owner}, revision).status + ) + + threads = [ + threading.Thread(target=write, args=("left",)), + threading.Thread(target=write, args=("right",)), + ] + for thread in threads: + thread.start() + barrier.wait() + for thread in threads: + thread.join(timeout=5) + + assert not any(thread.is_alive() for thread in threads) + assert results.count(ConfigSaveStatus.SAVED) == 1 + assert results.count(ConfigSaveStatus.CONFLICT) == 1 + assert toml.load(path)["owner"] in {"left", "right"} + + +def test_toml_failed_atomic_replace_preserves_original(tmp_path, monkeypatch): + import dashpva.utils.config.source as source_module + + path = tmp_path / "profile.toml" + original = toml.dumps({"owner": "initial"}).encode() + path.write_bytes(original) + source = ConfigSource(str(path)) + _, revision = source.load_snapshot() + + def fail_replace(*_args): + raise OSError("injected replacement failure") + + monkeypatch.setattr(source_module.os, "replace", fail_replace) + result = source.replace_if_revision({"owner": "replacement"}, revision) + + assert result.status is ConfigSaveStatus.ERROR + assert "injected replacement failure" in result.error + assert path.read_bytes() == original + + +def test_toml_snapshot_read_does_not_require_a_writable_lock_location( + tmp_path, monkeypatch +): + import dashpva.utils.config.source as source_module + + path = tmp_path / "profile.toml" + path.write_text(toml.dumps({"owner": "read-only"})) + + def fail_if_locked(*_args, **_kwargs): + raise AssertionError("read snapshots must not create a sidecar lock") + + monkeypatch.setattr(source_module, "_config_lock", fail_if_locked) + + config, revision = ConfigSource(str(path)).load_snapshot() + + assert config == {"owner": "read-only"} + assert revision + + +def test_database_stale_revision_cannot_overwrite_newer_save(tmp_db): + profile = tmp_db.create_profile("cas_profile") + assert tmp_db.import_toml_to_profile(profile.id, {"owner": "initial"}) + source = DbProfileConfigSource(tmp_db, profile.id) + _, first_revision = source.load_snapshot() + _, stale_revision = source.load_snapshot() + + first = source.replace_if_revision({"owner": "first"}, first_revision) + stale = source.replace_if_revision({"owner": "stale"}, stale_revision) + + assert first.status is ConfigSaveStatus.SAVED + assert stale.status is ConfigSaveStatus.CONFLICT + assert tmp_db.export_profile_to_toml(profile.id) == {"owner": "first"} + + +def test_database_concurrent_writers_have_exactly_one_winner(tmp_db): + profile = tmp_db.create_profile("concurrent_cas_profile") + assert tmp_db.import_toml_to_profile(profile.id, {"owner": "initial"}) + source = DbProfileConfigSource(tmp_db, profile.id) + _, revision = source.load_snapshot() + barrier = threading.Barrier(3) + results = [] + + def write(owner): + barrier.wait() + results.append( + source.replace_if_revision({"owner": owner}, revision).status + ) + + threads = [ + threading.Thread(target=write, args=("left",)), + threading.Thread(target=write, args=("right",)), + ] + for thread in threads: + thread.start() + barrier.wait() + for thread in threads: + thread.join(timeout=5) + + assert not any(thread.is_alive() for thread in threads) + assert results.count(ConfigSaveStatus.SAVED) == 1 + assert results.count(ConfigSaveStatus.CONFLICT) == 1 + assert tmp_db.export_profile_to_toml(profile.id)["owner"] in {"left", "right"} diff --git a/tests/unit/test_config_resolution.py b/tests/unit/test_config_resolution.py new file mode 100644 index 00000000..a66767a3 --- /dev/null +++ b/tests/unit/test_config_resolution.py @@ -0,0 +1,153 @@ +# Copyright (C) UChicago Argonne, LLC +# See LICENSE file for details +"""Contracts for persisted (raw) versus runtime-effective RSM profiles.""" + +import copy +from pathlib import Path + +import pytest +import toml + +from dashpva.utils.config.resolver import resolve_profile_config + + +def _canonical_profile(): + return { + "IOC_PREFIX": "6idb", + "HKL": { + "SAMPLE_CIRCLE_AXIS_1": { + "POSITION": "hand-authored:position", + "VENDOR_EXTENSION": "keep-me", + }, + "SAMPLE_CIRCLE_AXIS_2": {"POSITION": "obsolete:position"}, + "CUSTOM_EXTENSION": {"ENABLED": True}, + }, + "IOC_RSM_PARAMETER": { + "SAMPLE_AXES": [ + { + "LABEL": "Mu", + "RECORD_NAME": "Mu", + "DIRECTION": "x+", + "SOURCE_PV": "6idb1:m28.RBV", + "ANGLE_UNITS": "deg", + } + ], + "DETECTOR_AXES": [ + { + "LABEL": "Delta", + "RECORD_NAME": "Delta", + "DIRECTION": "z-", + "SOURCE_PV": "6idb1:m18.RBV", + "ANGLE_UNITS": "deg", + } + ], + "ENERGY_SOURCE_PV": "6idb:spec:Energy", + "ENERGY_UNITS": "keV", + }, + } + + +def test_legacy_profile_is_equal_but_deeply_detached(): + raw = { + "IOC_PREFIX": "legacy", + "HKL": {"SPEC": {"ENERGY_VALUE": "hand-authored:energy"}}, + } + + effective = resolve_profile_config(raw) + + assert effective == raw + assert effective is not raw + assert effective["HKL"] is not raw["HKL"] + effective["HKL"]["SPEC"]["ENERGY_VALUE"] = "changed" + assert raw["HKL"]["SPEC"]["ENERGY_VALUE"] == "hand-authored:energy" + + +def test_canonical_profile_generates_managed_hkl_without_mutating_raw(): + raw = _canonical_profile() + before = copy.deepcopy(raw) + + effective = resolve_profile_config(raw) + + assert raw == before + assert effective["IOC_PREFIX"] == "6idb:" + assert effective["HKL"]["SAMPLE_CIRCLE_AXIS_1"] == { + "VENDOR_EXTENSION": "keep-me", + "AXIS_NUMBER": "6idb:Mu:AxisNumber", + "DIRECTION_AXIS": "6idb:Mu:DirectionAxis", + "POSITION": "6idb:Mu:Position", + "SPEC_MOTOR_NAME": "6idb:Mu:SpecMotorName", + } + assert effective["HKL"]["DETECTOR_CIRCLE_AXIS_1"]["POSITION"] == ( + "6idb:Delta:Position" + ) + assert "SAMPLE_CIRCLE_AXIS_2" not in effective["HKL"] + assert effective["HKL"]["CUSTOM_EXTENSION"] == {"ENABLED": True} + assert effective["HKL"]["SPEC"]["ENERGY_VALUE"] == "6idb:spec:Energy:Value" + + +def test_empty_canonical_axis_lists_remove_all_legacy_managed_axes(): + raw = _canonical_profile() + raw["IOC_RSM_PARAMETER"]["SAMPLE_AXES"] = [] + raw["IOC_RSM_PARAMETER"]["DETECTOR_AXES"] = [] + + effective = resolve_profile_config(raw) + + assert not any("_CIRCLE_AXIS_" in name for name in effective["HKL"]) + assert "SAMPLE_CIRCLE_AXIS_1" in raw["HKL"] + + +def test_record_names_must_be_unique_across_sample_and_detector_roles(): + raw = _canonical_profile() + raw["IOC_RSM_PARAMETER"]["DETECTOR_AXES"][0]["RECORD_NAME"] = "Mu" + + with pytest.raises(ValueError, match="duplicate RECORD_NAME"): + resolve_profile_config(raw) + + +def test_prefixed_record_name_is_rejected_instead_of_double_prefixing(): + raw = _canonical_profile() + raw["IOC_RSM_PARAMETER"]["SAMPLE_AXES"][0]["RECORD_NAME"] = "other:Mu" + + with pytest.raises(ValueError, match="unprefixed record stem"): + resolve_profile_config(raw) + + +def test_record_name_rejects_whitespace_that_would_create_an_invalid_pv(): + raw = _canonical_profile() + raw["IOC_RSM_PARAMETER"]["SAMPLE_AXES"][0]["RECORD_NAME"] = "Mu Position" + + with pytest.raises(ValueError, match="record stem"): + resolve_profile_config(raw) + + +def test_canonical_output_never_borrows_detector_prefix(): + raw = _canonical_profile() + raw.pop("IOC_PREFIX") + raw["DETECTOR_PREFIX"] = "detector" + + effective = resolve_profile_config(raw) + + assert effective["IOC_PREFIX"] == "" + assert effective["HKL"]["SAMPLE_CIRCLE_AXIS_1"]["POSITION"] == "Mu:Position" + + +def test_settings_keeps_raw_profile_and_exports_effective_snapshot(tmp_path): + from dashpva.settings import Settings + + raw = _canonical_profile() + profile_path = tmp_path / "canonical.toml" + profile_path.write_text(toml.dumps(raw)) + + settings = Settings.from_toml(str(profile_path)) + snapshot_path = Path(settings.ensure_path()) + try: + snapshot = toml.load(snapshot_path) + assert settings.RAW_CONFIG == raw + assert settings.CONFIG["HKL"]["SAMPLE_CIRCLE_AXIS_1"]["POSITION"] == ( + "6idb:Mu:Position" + ) + assert toml.load(profile_path) == raw + assert snapshot_path != profile_path + assert snapshot == settings.CONFIG + finally: + snapshot_path.unlink(missing_ok=True) diff --git a/tests/unit/test_pr1_geometry_contract.py b/tests/unit/test_pr1_geometry_contract.py new file mode 100644 index 00000000..8e582955 --- /dev/null +++ b/tests/unit/test_pr1_geometry_contract.py @@ -0,0 +1,231 @@ +# Copyright (C) UChicago Argonne, LLC +# See LICENSE file for details +"""Behavior-preservation contracts for the shared PR 1 RSM geometry path.""" + +import h5py +import numpy as np +import pytest +import xrayutilities as xu + +from dashpva.utils.rsm_converter import RSMConverter +from dashpva.utils.rsm_geometry import ( + DetectorModel, + RotationAxis, + RSMGeometry, + build_hxrd, + calculate_q, +) + +from ._synthetic_hdf5 import make_synthetic_scan_h5 + + +def _model( + *, + sample_axes=(RotationAxis("sample", "z-"),), + detector_axes=(RotationAxis("detector", "z-"),), + sample_orientation="det", + primary_beam=(0.0, 1.0, 0.0), + inplane=(1.0, 0.0, 0.0), + surface=(0.0, 0.0, 1.0), +): + return RSMGeometry( + sample_axes=sample_axes, + detector_axes=detector_axes, + primary_beam_direction=primary_beam, + inplane_reference_direction=inplane, + sample_surface_normal_direction=surface, + energy_eV=10000.0, + ub_matrix=np.eye(3), + detector=DetectorModel( + "x+", + "z+", + (1.0, 2.0), + (3, 5), + (1.0, 1.0), + 500.0, + (0, 3, 0, 5), + ), + sample_orientation=sample_orientation, + ) + + +def test_legacy_six_axis_math_remains_bit_identical(tmp_path): + """The shared builder must not perturb the legacy xrayutilities call.""" + path = str(tmp_path / "legacy_geometry.h5") + make_synthetic_scan_h5(path, n_frames=4, shape=(3, 5)) + + sample_directions = ["x+", "z-", "y+", "z-"] + detector_directions = ["x+", "z-"] + sample_angles = [ + np.linspace(0.0, 10.0, 4), + np.full(4, 2.0), + np.full(4, 3.0), + np.full(4, 4.0), + ] + detector_angles = [np.full(4, 20.0), np.full(4, 25.0)] + string_dtype = h5py.string_dtype(encoding="utf-8") + with h5py.File(path, "r+") as h5_file: + hkl = h5_file["entry/data/metadata/HKL"] + del hkl["SAMPLE_CIRCLE_AXIS_1/DIRECTION_AXIS"] + hkl["SAMPLE_CIRCLE_AXIS_1"].create_dataset( + "DIRECTION_AXIS", data=np.array(["x+"], dtype=object), dtype=string_dtype + ) + for number, (direction, positions) in enumerate( + zip(sample_directions[1:], sample_angles[1:]), start=2 + ): + axis = hkl.create_group(f"SAMPLE_CIRCLE_AXIS_{number}") + axis.create_dataset( + "DIRECTION_AXIS", + data=np.array([direction], dtype=object), + dtype=string_dtype, + ) + axis.create_dataset("POSITION", data=positions) + + del hkl["DETECTOR_CIRCLE_AXIS_1/DIRECTION_AXIS"] + hkl["DETECTOR_CIRCLE_AXIS_1"].create_dataset( + "DIRECTION_AXIS", data=np.array(["x+"], dtype=object), dtype=string_dtype + ) + detector_axis_2 = hkl.create_group("DETECTOR_CIRCLE_AXIS_2") + detector_axis_2.create_dataset( + "DIRECTION_AXIS", data=np.array(["z-"], dtype=object), dtype=string_dtype + ) + detector_axis_2.create_dataset("POSITION", data=detector_angles[1]) + + converter = RSMConverter() + with h5py.File(path, "r") as h5_file: + geometry = converter.build_file_geometry(h5_file) + actual = converter.q_for_frames(geometry, h5_file, 0, 4) + + qconv = xu.experiment.QConversion( + sample_directions, + detector_directions, + [0.0, 1.0, 0.0], + ) + reference = xu.HXRD( + [1.0, 0.0, 0.0], + [0.0, 0.0, 1.0], + en=10000.0, + qconv=qconv, + ) + reference.Ang2Q.init_area( + "x+", + "z+", + cch1=1, + cch2=2, + Nch1=3, + Nch2=5, + pwidth1=1.0, + pwidth2=1.0, + distance=500.0, + roi=[0, 3, 0, 5], + ) + expected = reference.Ang2Q.area( + *sample_angles, + *detector_angles, + UB=np.eye(3), + ) + + for actual_axis, expected_axis in zip(actual, expected): + np.testing.assert_array_equal(actual_axis, expected_axis) + + +def test_shared_builder_is_bit_identical_to_direct_xrayutilities(): + model = _model() + sample_angle = np.linspace(0.0, 10.0, 4) + detector_angle = np.full(4, 20.0) + + actual = calculate_q(build_hxrd(model), [sample_angle], [detector_angle]) + + qconv = xu.experiment.QConversion(["z-"], ["z-"], [0.0, 1.0, 0.0]) + reference = xu.HXRD( + [1.0, 0.0, 0.0], + [0.0, 0.0, 1.0], + en=10000.0, + qconv=qconv, + sampleor="det", + ) + reference.Ang2Q.init_area( + "x+", + "z+", + cch1=1.0, + cch2=2.0, + Nch1=3, + Nch2=5, + pwidth1=1.0, + pwidth2=1.0, + distance=500.0, + roi=[0, 3, 0, 5], + ) + expected = reference.Ang2Q.area( + sample_angle, + detector_angle, + UB=np.eye(3), + deg=True, + ) + + for actual_axis, expected_axis in zip(actual, expected): + np.testing.assert_array_equal(actual_axis, expected_axis) + + +@pytest.mark.parametrize( + "detector_axes", + [(), (RotationAxis("detector", "y+"),)], +) +def test_det_orientation_rejects_geometry_without_usable_detector_axis( + detector_axes, +): + with pytest.raises(ValueError, match="SAMPLE_ORIENTATION='det'"): + _model(detector_axes=detector_axes) + + +def test_det_orientation_ignores_one_innermost_beam_axis(): + model = _model( + detector_axes=( + RotationAxis("detector", "z-"), + RotationAxis("detector", "y+"), + ) + ) + + built = build_hxrd(model) + + assert built.model.sample_orientation == "det" + + +def test_sam_orientation_requires_a_sample_axis(): + with pytest.raises(ValueError, match="requires a sample rotation axis"): + _model(sample_axes=(), sample_orientation="sam") + + +def test_sam_orientation_rejects_beam_parallel_innermost_axis(): + with pytest.raises(ValueError, match="innermost sample axis"): + _model( + sample_axes=(RotationAxis("sample", "y+"),), + sample_orientation="sam", + ) + + +def test_sam_orientation_warns_that_innermost_axis_must_be_azimuth(): + with pytest.warns(UserWarning, match="azimuth motor"): + model = _model(sample_orientation="sam") + + assert model.sample_orientation == "sam" + + +def test_explicit_orientation_rejects_primary_beam_parallel_direction(): + with pytest.raises(ValueError, match="silently substitute"): + _model(sample_orientation="y+") + + +def test_inplane_and_surface_directions_must_be_perpendicular(): + with pytest.raises(ValueError, match="perpendicular"): + _model(inplane=(1.0, 0.0, 1.0)) + + +def test_kappa_is_sample_only_and_roles_cannot_cross_lists(): + kappa = RotationAxis("sample", "k+") + assert kappa.direction == "k+" + + with pytest.raises(ValueError, match="Invalid detector rotation direction"): + RotationAxis("detector", "k+") + with pytest.raises(ValueError, match="sample_axes"): + _model(sample_axes=(RotationAxis("detector", "z-"),))