Add source names to DetectorData

Author: philsmt

src/extra/calibration.py

@@ -1,3 +1,5 @@
+
+import ast
 import json
 import re
 from collections.abc import Mapping
@@ -8,6 +10,7 @@
 from functools import lru_cache
 from operator import index
 from pathlib import Path
+from string import Formatter
 from typing import Dict, List, Optional, Union
 from urllib.parse import urljoin
 from warnings import warn
@@ -1368,6 +1371,21 @@ class ShimadzuHPVX2Conditions(ConditionsBase):
     }
 
 
+class SourceExprChecker(ast.NodeVisitor):
+    def visit_Call(self, node):
+        raise ValueError("Function calls not allowed in source name patterns")
+
+
+class SourceNameFormatter(Formatter):
+    """String formatter that evaluates simple operations like {modno + 2}"""
+
+    def get_field(self, field_name, args, kwargs):
+        node = ast.parse(field_name, "<source pattern>", "eval")
+        SourceExprChecker().visit(node)
+        obj = eval(compile(node, "<source pattern>", "eval"), kwargs)
+        return obj, 0
+
+
 @dataclass
 class DetectorModule:
     """Detector module.
@@ -1402,12 +1420,17 @@ class DetectorModule:
     module_number: int | None
     detector_type: str
     legacy_uuid: int | None  # Deprecated, do not use
+    source_name: str | None
 
     def __post_init__(self):
         if self.module_number is None:
             # Try to fill in module number if missing.
             self.module_number = int(re.findall(r"\d+", self.aggregator)[-1])
 
+        if self.source_name is not None:
+            self.source_name = SourceNameFormatter().format(
+                self.source_name, modno=self.module_number)
+
     @property
     def ccv_params(self):
         """PDU arguments as needed for write_ccv()."""
@@ -1465,7 +1488,7 @@ def get_da(x):
                     item['id'], item['physical_name'], item['karabo_da'],
                     self.identifier, item['virtual_device_name'], i,
                     item['module_number'], item['detector_type']['name'],
-                    item['uuid'])
+                    item['uuid'], self._source_name_pattern)
             else:
                 item.module_index = i
 
@@ -1598,6 +1621,13 @@ def source_name_pattern(self) -> str:
             'incomplete detector entry in CalCat'
         return self._source_name_pattern
 
+    @property
+    def source_names(self) -> str:
+        """Source names."""
+        assert self._source_name_pattern is not None, \
+            'incomplete detector entry in CalCat'
+        return [pdu.source_name for pdu in self.pdus]
+
     @property
     def first_module_index(self) -> int:
         """Module index of the first module."""

tests/test_calibration.py

@@ -18,7 +18,8 @@
     LPDConditions,
     SingleConstant,
     DetectorData,
-    DetectorModule
+    DetectorModule,
+    SourceNameFormatter
 )
 
 # Most of these tests use saved HTTP responses by default (with pytest-recording).
@@ -399,12 +400,16 @@ def test_DetectorData_from_identifier():
     assert list(agipd)[0] == next(iter(agipd.keys())) == 'AGIPD00'
     assert isinstance(next(iter(agipd.values())), DetectorModule)
     assert agipd[0] == agipd['AGIPD00']
+    assert agipd.source_names == [
+        f'SPB_DET_AGIPD1M-1/DET/{i}CH0:xtdf'
+        for i, name in range(len(agipd.source_names))]
 
     # PDU
     pdu = agipd[0]
     assert pdu.aggregator == 'AGIPD00'
     assert pdu.ccv_params == (
         'AGIPD_SIV1_AGIPDV11_M517', 101003000000, 'AGIPD-Type')
+    assert pdu.source_name == 'SPB_DET_AGIPD1M-1/DET/0CH0:xtdf'
 
     # FXE-JFHZ, single-module detector with partial CalCat entries
     jfhz = DetectorData.from_identifier('FXE_XAD_JFHZ', **pdu_date_kw)
@@ -417,6 +422,9 @@ def test_DetectorData_from_identifier():
     with pytest.raises(AssertionError):
         jfhz.first_module_index
 
+    with pytest.raises(AssertionError):
+        jfhz.source_names
+
     # SQS-DSSC, PDU-less detector
     dssc = DetectorData.from_identifier('SQS_DET_DSSC1M-1', **pdu_date_kw)
     repr(dssc)
@@ -456,3 +464,14 @@ def test_DetectorData_list_by_instrument():
 def test_DetectorData_from_CalibrationData():
     agipd_cd = CalibrationData.from_report(3757)
     assert agipd_cd.detector.identifier == 'SPB_DET_AGIPD1M-1'
+
+
+def test_DetectorData_SourceNameFormatter():
+    fmt = SourceNameFormatter()
+
+    assert 'SPB_DET_AGIPD1M-1/DET/4CH0:xtdf' == \
+        fmt.format('SPB_DET_AGIPD1M-1/DET/{modno}CH0:xtdf', modno=4)
+    assert 'SPB_IRDA_JF4M/DET/JNGFR03:daqOutput' == \
+        fmt.format('SPB_IRDA_JF4M/DET/JNGFR{modno:02d}:daqOutput', modno=3)
+    assert 'HED_TST_AGIPDHZ3/DET/84CH0:xtdf' == \
+        fmt.format('HED_TST_AGIPDHZ3/DET/{modno+83}CH0:xtdf', modno=1)