refactor: safe configuration resolution and shared RSM geometry (1/4)

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'):

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:

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:

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)
 

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()