Add ACR status option for CCM energy moves

docs/source/upcoming_release_notes/1194-Add_ACR_status_option_for_CCM_energy_moves.rst

@@ -0,0 +1,34 @@
+1194 Add ACR status option for CCM energy moves
+#################
+
+API Breaks
+----------
+- N/A
+
+Features
+--------
+- N/A
+
+Device Updates
+--------------
+- Add a CCMEnergyWithACRStatus class to ccm.py
+- Add a energy_with_acr_status instance to CCM
+- Update BeamEnergyRequest argument from bunch to pv_index to better reflect the broader use cases.
+  A backward compatible warning is now returned if the old bunch kwarg is used.
+- Update atol in BeamEnergyRequestNoWait to 0.5 (was 5). This is needed for self-seeding
+
+New Devices
+-----------
+- Add a convenience decorator to re-arg a function in utils.py
+
+Bugfixes
+--------
+- N/A
+
+Maintenance
+-----------
+- N/A
+
+Contributors
+------------
+- vespos

pcdsdevices/beam_stats.py

@@ -12,6 +12,7 @@
 from .interface import BaseInterface, FltMvInterface
 from .pv_positioner import PVPositionerDone
 from .signal import AvgSignal
+from .utils import re_arg
 
 logger = logging.getLogger(__name__)
 
@@ -89,9 +90,9 @@ class BeamEnergyRequest(FltMvInterface, Device, PositionerBase):
         in them, the L line PVs just have "EPHOT". This will default to the
         line associated with the prefix hutch name, or to L line failing that.
 
-    bunch : int, optional
-        Whether to move the first bunch (1) or the second bunch (2). This is
-        only relevant for 2-color mode. Defaults to bunch 1.
+    pv_index : int, optional
+        Whether to move the first PV (1) or the second PV (2). This is relevant
+        2-color mode or when scanning combined K and Vernier. Defaults to 1.
 
     acr_status_suffix : str, optional
         If provided, we'll wait on the ACR PV specified by
@@ -100,17 +101,19 @@ class BeamEnergyRequest(FltMvInterface, Device, PositionerBase):
     """
     setpoint = FCpt(
         EpicsSignal,
-        '{prefix}:USER:MCC:EPHOT{line_text}:SET{bunch}',
+        '{prefix}:USER:MCC:EPHOT{line_text}:SET{pv_index}',
         kind='hinted',
+        add_prefix=('suffix', 'write_pv', 'line_text', 'pv_index'),
         doc=(
             'The setpoint PV that acr listens on to update the '
             'vernier or undulator PVs as appropriate.'
         ),
     )
     ref = FCpt(
         EpicsSignal,
-        '{prefix}:USER:MCC:EPHOT{line_text}:REF{bunch}',
+        '{prefix}:USER:MCC:EPHOT{line_text}:REF{pv_index}',
         kind='normal',
+        add_prefix=('suffix', 'write_pv', 'line_text', 'pv_index'),
         doc=(
             'A reference PV for the photon energy at the nominal '
             'position of the vernier or undulator.'
@@ -138,18 +141,19 @@ def __new__(
             return super().__new__(BeamEnergyRequestNoWait)
         return super().__new__(BeamEnergyRequestACRWait)
 
+    @re_arg({"bunch": "pv_index"})
     def __init__(
         self,
         prefix: str,
         *,
         name: str,
         line: Optional[str] = None,
-        bunch: int = 1,
+        pv_index: int = 1,
         acr_status_suffix: Optional[str] = None,
         **kwargs
     ):
         self.line_text = self.line_text_dict.get(line or prefix, '')
-        self.bunch = bunch
+        self.pv_index = pv_index
         self.acr_status_suffix = acr_status_suffix
         super().__init__(prefix, name=name, **kwargs)
 
@@ -161,7 +165,7 @@ class BeamEnergyRequestNoWait(BeamEnergyRequest, PVPositionerDone):
     It will report done immediately and ignore moves that are smaller than
     atol.
     """
-    atol = 5
+    atol = 0.5
 
     # All done-related functionality is inherited from PVPositionerDone
     # Just implement skip_small_moves's default
@@ -181,6 +185,7 @@ class BeamEnergyRequestACRWait(BeamEnergyRequest, PVPositioner):
         EpicsSignal,
         'SIOC:SYS0:ML07:{acr_status_suffix}',
         kind='normal',
+        add_prefix=('suffix', 'write_pv', 'acr_status_suffix'),
         doc=(
             'PV that is 0 while the motors are moving and 1 when ACR is '
             'ready for a new request. ACR can pick which of these PVs '

pcdsdevices/ccm.py

@@ -722,8 +722,8 @@ class CCMEnergyWithVernier(CCMEnergy):
         PVs to write to. If omitted, we can guess this from the
         prefix.
     """
-    vernier = FCpt(BeamEnergyRequest, '{hutch}', kind='normal',
-                   doc='Requests ACR to move the Vernier.')
+    acr_energy = FCpt(BeamEnergyRequest, '{hutch}', kind='normal',
+                      doc='Requests ACR to move the Vernier.')
 
     # These are duplicate warnings with main energy motor
     _enable_warn_constants: bool = False
@@ -759,7 +759,7 @@ def forward(self, pseudo_pos: namedtuple) -> namedtuple:
         energy = pseudo_pos.energy
         alio = self.energy_to_alio(energy)
         vernier = energy * 1000
-        return self.RealPosition(alio=alio, vernier=vernier)
+        return self.RealPosition(alio=alio, acr_energy=vernier)
 
     def inverse(self, real_pos: namedtuple) -> namedtuple:
         """
@@ -773,6 +773,46 @@ def inverse(self, real_pos: namedtuple) -> namedtuple:
         return self.PseudoPosition(energy=energy)
 
 
+class CCMEnergyWithACRStatus(CCMEnergyWithVernier):
+    """
+    CCM energy motor and ACR beam energy request with status.
+    Note that in this case vernier indicates any ways that ACR will act on the
+    photon energy request. This includes the Vernier, but can also lead to
+    motion of the undulators or the K.
+
+    Parameters
+    ----------
+    prefix : str
+        The PV prefix of the Alio motor, e.g. XPP:MON:MPZ:07A
+    hutch : str, optional
+        The hutch we're in. This informs us as to which vernier
+        PVs to write to. If omitted, we can guess this from the
+        prefix.
+    acr_status_sufix : str
+        Prefix to the SIOC PV that ACR uses to report the move status.
+        For HXR this usually is 'AO805'.
+    """
+    acr_energy = FCpt(BeamEnergyRequest, '{hutch}',
+                      pv_index='{pv_index}',
+                      acr_status_suffix='{acr_status_suffix}',
+                      add_prefix=('suffix', 'write_pv', 'pv_index',
+                                  'acr_status_suffix'),
+                      kind='normal',
+                      doc='Requests ACR to move the energy.')
+
+    def __init__(
+        self,
+        prefix: str,
+        hutch: typing.Optional[str] = None,
+        acr_status_suffix='AO805',
+        pv_index=2,
+        **kwargs
+    ):
+        self.acr_status_suffix = acr_status_suffix
+        self.pv_index = pv_index
+        super().__init__(prefix, **kwargs)
+
+
 class CCMX(SyncAxis):
     """
     Combined motion of the CCM X motors.
@@ -892,6 +932,11 @@ class CCM(BaseInterface, GroupDevice, LightpathMixin, CCMConstantsMixin):
         The prefix for the north upstream ccm y translation motor (y2).
     y_up_south_prefix : str, required keyword
         The prefix for the south upstream ccm y translation motor (y3).
+    acr_status_pv_index : int
+        The index for the energy request PV in the case of the acr status
+        wait. Default: 2.
+    acr_status_suffix : str
+        The suffix for the ACR status energy change move. Default to 'AO805'
     """
     energy = Cpt(
         CCMEnergy, '', kind='hinted',
@@ -903,11 +948,22 @@ class CCM(BaseInterface, GroupDevice, LightpathMixin, CCMConstantsMixin):
     energy_with_vernier = Cpt(
         CCMEnergyWithVernier, '', kind='normal',
         doc=(
-            'PsuedoPositioner that moves the alio in '
+            'PseudoPositioner that moves the alio in '
             'terms of the calculated CCM energy while '
             'also requesting a vernier move.'
         ),
     )
+    energy_with_acr_status = FCpt(
+        CCMEnergyWithACRStatus, '{prefix}', kind='normal',
+        acr_status_suffix='{acr_status_suffix}',
+        add_prefix=('suffix', 'write_pv', 'acr_status_suffix'),
+        doc=(
+            'PseudoPositioner that moves the alio in '
+            'terms of the calculated CCM energy while '
+            'also requesting an energy change to ACR. '
+            'This will wait on ACR to complete the move.'
+        ),
+    )
 
     alio = UCpt(CCMAlio, kind='normal',
                 doc='The motor that rotates the CCM crystal.')
@@ -939,8 +995,9 @@ class CCM(BaseInterface, GroupDevice, LightpathMixin, CCMConstantsMixin):
 
     lightpath_cpts = ['x.up.user_readback']
     tab_whitelist = ['x1', 'x2', 'y1', 'y2', 'y3', 'E', 'E_Vernier',
-                     'th2fine', 'alio2E', 'E2alio', 'alio', 'home',
-                     'kill', 'insert', 'remove', 'inserted', 'removed']
+                     'energy_with_acr_status', 'th2fine', 'alio2E', 'E2alio',
+                     'alio', 'home', 'kill', 'insert', 'remove', 'inserted',
+                     'removed']
 
     _in_pos: float
     _out_pos: float
@@ -957,6 +1014,8 @@ def __init__(
         self._in_pos = in_pos
         self._out_pos = out_pos
         prefix = prefix or self.unrelated_prefixes['alio_prefix']
+        self.acr_status_suffix = kwargs.get('acr_status_suffix', 'AO805')
+        self.acr_status_pv_index = kwargs.get('acr_status_suffix', 2)
         super().__init__(prefix, **kwargs)
 
         # Aliases: defined by the scientists

pcdsdevices/tests/test_beam_stats.py

@@ -107,14 +107,14 @@ def test_beam_energy_request_args():
         'TST',
         name='tst_k1_request',
         line='k',
-        bunch=1,
+        pv_index=1,
     )
     assert tst_k1_request.setpoint.pvname == 'TST:USER:MCC:EPHOTK:SET1'
     tst_l2_request = BeamEnergyRequest(
         'TST',
         name='tst_l2_request',
         line='L',
-        bunch=2,
+        pv_index=2,
     )
     assert tst_l2_request.setpoint.pvname == 'TST:USER:MCC:EPHOT:SET2'
     # let's test the class splitting here too

pcdsdevices/tests/test_ccm.py

@@ -86,7 +86,7 @@ def init_pos(mot, pos=0):
     fake_ccm.alio.set(SAMPLE_ALIO)
     fake_ccm.energy.alio.set(SAMPLE_ALIO)
     fake_ccm.energy_with_vernier.alio.set(SAMPLE_ALIO)
-    fake_ccm.energy_with_vernier.vernier.setpoint.sim_put(0)
+    fake_ccm.energy_with_vernier.acr_energy.setpoint.sim_put(0)
 
     return fake_ccm
 
@@ -165,26 +165,26 @@ def test_vernier(fake_ccm):
     # Moving with vernier should move the energy request motor too
     pseudopos.move(7, wait=False)
     assert np.isclose(pseudopos.energy.position, 7)
-    assert pseudopos.vernier.position == 7000
+    assert pseudopos.acr_energy.position == 7000
 
     pseudopos.move(8, wait=False)
     assert np.isclose(pseudopos.energy.position, 8)
-    assert pseudopos.vernier.position == 8000
+    assert pseudopos.acr_energy.position == 8000
 
     pseudopos.move(9, wait=False)
     assert np.isclose(pseudopos.energy.position, 9)
-    assert pseudopos.vernier.position == 9000
+    assert pseudopos.acr_energy.position == 9000
 
     # Small moves (less than 30eV) should be skipped on the energy request
-    pseudopos.move(9.001, wait=False)
-    assert np.isclose(pseudopos.energy.position, 9.001)
-    assert pseudopos.vernier.position == 9000
+    pseudopos.move(9.0001, wait=False)
+    assert np.isclose(pseudopos.energy.position, 9.0001)
+    assert pseudopos.acr_energy.position == 9000
 
     # Unless we set the option for not skipping them
-    pseudopos.vernier.skip_small_moves = False
+    pseudopos.acr_energy.skip_small_moves = False
     pseudopos.move(9.002, wait=False)
     assert np.isclose(pseudopos.energy.position, 9.002)
-    assert pseudopos.vernier.position == 9002
+    assert pseudopos.acr_energy.position == 9002
 
 
 @pytest.mark.timeout(5)

pcdsdevices/utils.py

@@ -53,6 +53,27 @@
 minus = '-'
 
 
+def re_arg(kwarg_map):
+    """
+    Decorator to redefine a kwarg to a function, while still supporting the old kwarg, and warning the end user.
+
+    Usage:
+    @re_arg({"new_kwarg": "old_kwarg"})
+    def myfunc(*args, new_kwarg=<a_value>, **kwargs):
+        ...
+    """
+    def decorator(func):
+        def wrapped(*args, **kwargs):
+            new_kwargs = {}
+            for k, v in kwargs.items():
+                if k in kwarg_map:
+                    print(f"DEPRECATION WARNING: keyword argument '{k}' is no longer valid. Use '{kwarg_map[k]}' instead.")
+                new_kwargs[kwarg_map.get(k, k)] = v
+            return func(*args, **new_kwargs)
+        return wrapped
+    return decorator
+
+
 def is_input():
     """
     Utility to check if there is input available.