Merge pull request #822 from lnls-sirius/improve-sofb

Allow removal of correctors in PSSOFB mode

siriuspy/siriuspy/VERSION

@@ -1 +1 @@
-2.49.0
+2.50.0

siriuspy/siriuspy/magnet/excdata.py

@@ -96,22 +96,22 @@ def interp_curr2mult(self, currents, only_main_harmonic=False):
             for harm in harmonics:
                 # normal component
                 mpole = self.multipoles['normal'][harm]
-                interp = ExcitationData._calc_interp(currents, curr, mpole)
+                interp = _util.linear_interpolation(currents, curr, mpole)
                 multipoles['normal'][harm] = interp[0]
                 # skew component
                 mpole = self.multipoles['skew'][harm]
-                interp = ExcitationData._calc_interp(currents, curr, mpole)
+                interp = _util.linear_interpolation(currents, curr, mpole)
                 multipoles['skew'][harm] = interp[0]
         else:
             currents = _np.array(currents)
             for harm in harmonics:
                 # normal component
                 mpole = self.multipoles['normal'][harm]
-                interp = ExcitationData._calc_interp(currents, curr, mpole)
+                interp = _util.linear_interpolation(currents, curr, mpole)
                 multipoles['normal'][harm] = interp
                 # skew component
                 mpole = self.multipoles['skew'][harm]
-                interp = ExcitationData._calc_interp(currents, curr, mpole)
+                interp = _util.linear_interpolation(currents, curr, mpole)
                 multipoles['skew'][harm] = interp
         return multipoles
 
@@ -126,11 +126,11 @@ def interp_mult2curr(self, multipoles, harmonic, multipole_type):
         # do conversion
         if _np.isscalar(multipoles):
             multipoles = _np.array([multipoles])
-            interp = ExcitationData._calc_interp(multipoles, mpole, curr)
+            interp = _util.linear_interpolation(multipoles, mpole, curr)
             currents = interp[0]
         else:
             multipoles = _np.array(multipoles)
-            interp = ExcitationData._calc_interp(multipoles, mpole, curr)
+            interp = _util.linear_interpolation(multipoles, mpole, curr)
             currents = interp
         return currents
 
@@ -167,20 +167,6 @@ def __str__(self):
 
     # --- private methods ---
 
-    @staticmethod
-    def _calc_interp(xvals, xtab, ytab):
-        interp = _np.interp(
-            xvals, xtab, ytab, left=float('nan'), right=float('inf'))
-        nan = _np.isnan(interp)
-        inf = _np.isinf(interp)
-        vec = _util.linear_extrapolation(
-            xvals[nan], xtab[0], xtab[1], ytab[0], ytab[1])
-        interp[nan] = vec
-        vec = _util.linear_extrapolation(
-            xvals[inf], xtab[-1], xtab[-2], ytab[-1], ytab[-2])
-        interp[inf] = vec
-        return interp
-
     def _process_comment_line(self, line):
         if line[1:].strip():
             token, *words = line[1:].split()

siriuspy/siriuspy/magnet/util.py

@@ -72,6 +72,21 @@ def get_multipole_si_units(harmonic, power=None, product=None):
         return 'T/m{0:s}{1:d}'.format(power, harmonic-1)
 
 
+def linear_interpolation(xvals, xtab, ytab):
+    """Return linear interpolation function value."""
+    interp = _numpy.interp(
+        xvals, xtab, ytab, left=-_numpy.inf, right=_numpy.inf)
+    neg = _numpy.isneginf(interp)
+    pos = _numpy.isposinf(interp)
+    if neg.any():
+        interp[neg] = linear_extrapolation(
+            xvals[neg], xtab[0], xtab[1], ytab[0], ytab[1])
+    if pos.any():
+        interp[pos] = linear_extrapolation(
+            xvals[pos], xtab[-1], xtab[-2], ytab[-1], ytab[-2])
+    return interp
+
+
 def linear_extrapolation(x, x1, x2, y1, y2):
     """Return linear extrapolation function value."""
     if x2 == x1:

siriuspy/siriuspy/pwrsupply/bsmp/commands.py

@@ -573,8 +573,25 @@ def sofb_ps_func_return(self):
 
     def sofb_ps_setpoint_set(self, value):
         """."""
-        self._sofb_ps_setpoint = value
-        ack, func_resp = self.ps_function_set_slowref_fbp_readback_ref(value)
+        if self._sofb_ps_setpoint is None:
+            self._sofb_ps_setpoint = _np.zeros(
+                _const_psbsmp.UDC_MAX_NR_DEV, dtype=float)
+
+        # NOTE: NaNs in current vector are a way to avoid communication with
+        #   udcs. In case only a few inputs are NaN they will be replaced by
+        #   old values or zero. If all of them are NaN, them no communication
+        #   is performed.
+        nan = _np.isnan(value)
+        self._sofb_ps_setpoint[~nan] = value[~nan]
+
+        # NOTE: In case all values are NaN, no communication is made:
+        if nan.all():
+            ack = self.CONST_BSMP.ACK_OK
+            func_resp = self._sofb_ps_readback_ref
+        else:
+            ack, func_resp = self.ps_function_set_slowref_fbp_readback_ref(
+                self._sofb_ps_setpoint)
+
         if ack == self.CONST_BSMP.ACK_OK:
             self._sofb_ps_readback_ref = func_resp
             self._sofb_ps_func_return = FBP._ACK_OK

siriuspy/siriuspy/pwrsupply/pructrl/udc.py

@@ -200,9 +200,9 @@ def sofb_current_set(self, value):
         """Set SOFB Current."""
         if self._is_fbp:
             # set value
-            self._dev_first.sofb_ps_setpoint_set(value[:4])
+            self._dev_first.sofb_ps_setpoint_set(value[:_UDC_MAX_NR_DEV])
             if self._dev_second:
-                self._dev_second.sofb_ps_setpoint_set(value[4:])
+                self._dev_second.sofb_ps_setpoint_set(value[_UDC_MAX_NR_DEV:])
 
     def sofb_update(self):
         """Update sofb."""

siriuspy/siriuspy/pwrsupply/pssofb.py

@@ -392,19 +392,11 @@ def _bsmp_current_setpoint(self, bbbname, curr_sp):
         indcs_sofb = self.indcs_sofb[bbbname]
         indcs_bsmp = self.indcs_bsmp[bbbname]
 
-        # get valid current setpoints from sofb array
-        current = curr_sp[indcs_sofb]
-
         # initialize setpoint
-        # read last setpoint already stored in PSBSMP object:
-        readback = udc.sofb_current_rb_get()
-        if readback is None:
-            setpoint = _np.zeros(PSConnSOFB.MAX_NR_DEVS)
-        else:
-            setpoint = _np.asarray(readback)
-
-        # update setpoint
-        setpoint[indcs_bsmp] = current
+        # NOTE: curr_sp may contain NaNs. They will be handled by low level
+        #    classes.
+        setpoint = _np.full(PSConnSOFB.MAX_NR_DEVS, _np.nan)
+        setpoint[indcs_bsmp] = curr_sp[indcs_sofb]
 
         # --- bsmp communication ---
         try:
@@ -427,7 +419,8 @@ def _bsmp_current_setpoint(self, bbbname, curr_sp):
         # send signal to IOC to update one power supply state
         if self._sofb_update_iocs:
             pvobj = self._pvobjs[bbbname]
-            pvobj.put(1, wait=False)  # send signal to IOC
+            if pvobj.connected and not pvobj.status:
+                pvobj.put(1, wait=False)  # send signal to IOC
 
     def _bsmp_current_setpoint_update(self, bbbname, curr_sp):
         """."""

siriuspy/siriuspy/sofb/base_class.py

@@ -104,16 +104,18 @@ def __init__(self, acc, callback=None):
         self._config_ok_vals = {}
         self._config_pvs_rb = {}
         self._config_pvs_sp = {}
+        self.put_enable = True
 
     @property
     def connected(self):
         """Status connected."""
         conn = True
-        for k, pv in self._config_pvs_rb.items():
+        for pv in self._config_pvs_rb.values():
             if not pv.connected:
                 _log.debug('NOT CONN: ' + pv.pvname)
             conn &= pv.connected
-        for k, pv in self._config_pvs_sp.items():
+
+        for pv in self._config_pvs_sp.values():
             if not pv.connected:
                 _log.debug('NOT CONN: ' + pv.pvname)
             conn &= pv.connected
@@ -146,6 +148,9 @@ def configure(self):
         """Configure method."""
         if not self.connected:
             return False
+
+        if not self.put_enable:
+            return True
         for k, pvo in self._config_pvs_sp.items():
             if k in self._config_ok_vals:
                 pvo.put(self._config_ok_vals[k], wait=False)