rtcCompass.py

 
from shesha.init.rtc_init import rtc_init
from shesha.supervisor.components.sourceCompass import SourceCompass
import shesha.constants as scons
import numpy as np
 
class RtcCompass(object):
    """ RTC handler for compass simulation
 
    Attributes:
        rtc : (sutraWrap.Rtc) : Sutra rtc instance
 
        context : (carmaContext) : CarmaContext instance
 
        config : (config module) : Parameters configuration structure module    
 
        cacao : (bool) : CACAO features enabled in the RTC  
    """
    def __init__(self, context, config, tel, wfs, dms, atm, *, cacao=False):
        """ Initialize a RtcCompass component for rtc related supervision
 
        Parameters:
            context : (carmaContext) : CarmaContext instance
 
            config : (config module) : Parameters configuration structure module
 
            tel : (TelescopeCompass) : A TelescopeCompass instance
 
            wfs : (WfsCompass) : A WfsCompass instance
 
            dms : (DmCompass) : A DmCompass instance
 
            atm : (AtmosCompass) : An AtmosCompass instance
 
            cacao : (bool, optional) : If True, enables CACAO features in RTC (Default is False)
                                      /!\ Requires OCTOPUS to be installed
        """
        self.context = context
        self.config = config # Parameters configuration coming from supervisor init
        self.cacao = cacao
        print("->rtc init")
        self.rtc = rtc_init(self.context, tel.tel, wfs.wfs, dms.dms, atm.atmos,
                                self.config.p_wfss, self.config.p_tel,
                                self.config.p_geom, self.config.p_atmos, self.config.p_loop.ittime,
                                self.config.p_centroiders, self.config.p_controllers,
                                self.config.p_dms, cacao=cacao)
 
    def set_perturbation_voltage(self, controller_index: int, name: str,
                                 command: np.ndarray) -> None:
        """ Add circular buffer of offset values to integrator (will be applied at the end of next iteration)
 
        Parameters:
            controller_index : (int) : Controller index
 
            name : (str) : Buffer name
 
            command : (np.ndarray) : perturbation voltage circular buffer
        """
        if len(command.shape) == 1:
            self.rtc.d_control[controller_index].set_perturb_voltage(name, command, 1)
        elif len(command.shape) == 2:
            self.rtc.d_control[controller_index].set_perturb_voltage(
                    name, command, command.shape[0])
        else:
            raise AttributeError("command should be a 1D or 2D array")
 
    def get_slopes(self, controller_index: int) -> np.ndarray:
        """ Return the current slopes vector of the controller_index controller
 
        Parameters:
            controller_index : (int) : controller index handling the slopes
 
        Return:
            slopes : (np.ndarray) : Current slopes vector containing slopes of all
                                    the WFS handled by the specified controller
        """
        return np.array(self.rtc.d_control[controller_index].d_centroids)
 
    def close_loop(self, controller_index: int=None) -> None:
        """ DM receives controller output + pertuVoltage
 
        Parameters:
            controller_index: (int, optional): controller index. 
                                               If None (default), apply on all controllers
        """
        if controller_index is None:
            for controller in self.rtc.d_control:
                controller.set_open_loop(0)
        else:
            self.rtc.d_control[controller_index].set_open_loop(0)  # close_loop
 
    def open_loop(self, controller_index: int=None, reset=True) -> None:
        """ Integrator computation goes to /dev/null but pertuVoltage still applied
 
        Parameters:
            controller_index: (int): controller index.
                                     If None (default), apply on all controllers
 
            reset : (bool, optional) : If True (default), integrator is reset
        """
        if controller_index is None:
            for controller in self.rtc.d_control:
                controller.set_open_loop(1, reset)
        else:
            self.rtc.d_control[controller_index].set_open_loop(1, reset)  # open_loop
 
    def set_ref_slopes(self, ref_slopes: np.ndarray, *, centro_index=None) -> None:
        """ Set given ref slopes in centroider
 
        Parameters:
            ref_slopes : (ndarray) : Reference slopes vectoronly set the reference slop
 
            centro_index : (int, optionnal) : If given, only set the reference slopes vector
                                             used by the specified centroider. If None, the reference
                                             slopes vector must be a concatenation of all the reference
                                             slopes to use for each centroiders handled by the controller
        """
        if (centro_index is None):
            self.rtc.set_centroids_ref(ref_slopes)
        else:
            self.rtc.d_centro[centro_index].set_centroids_ref(ref_slopes)
 
    def get_ref_slopes(self, centro_index=None) -> np.ndarray:
        """ Get the currently used reference slopes
 
        Parameters:
            centro_index : (int, optionnal) : If given, only get the reference slopes vector
                                             used by the specified centroider. If None, the reference
                                             slopes vector returned is a concatenation of all the reference
                                             slopes used for by centroiders in the RTC
 
        Return:
            ref_slopes : (np.ndarray) : Reference slopes vector
        """
        ref_slopes = np.empty(0)
        if (centro_index is None):
            for centro in self.rtc.d_centro:
                ref_slopes = np.append(ref_slopes, np.array(centro.d_centroids_ref))
            return ref_slopes
        else:
            return np.array(self.rtc.d_centro[centro_index].d_centroids_ref)
 
    def set_gain(self, controller_index: int, gain: float) -> None:
        """ Set the scalar gain
 
        Parameters:
            controller_index : (int) : Index of the controller to modify
 
            gain : (float) : scalar gain of modal gain to set
        """
        self.rtc.d_control[controller_index].set_gain(gain)
 
    def get_interaction_matrix(self, controller_index: int):
        """ Return the interaction matrix of the controller
 
        Parameters :
            controller_index: (int): controller index
 
        Return:
            imat : (np.ndarray) : Interaction matrix currently set in the controller
        """
        return np.array(self.rtc.d_control[controller_index].d_imat)
 
    def get_command_matrix(self, controller_index: int):
        """ Return the command matrix of the controller
 
        Parameters:
            controller_index: (int): controller index
 
        Return:
            cmat : (np.ndarray) : Command matrix currently used by the controller
        """
        return np.array(self.rtc.d_control[controller_index].d_cmat)
 
    def set_command_matrix(self, controller_index: int, cmat: np.ndarray) -> None:
        """ Set the command matrix for the controller to use
 
        Parameters:
            controller_index : (int) : Controller index to modify
 
            cmat : (np.ndarray) : command matrix to set
        """
        self.rtc.d_control[controller_index].set_cmat(cmat)
 
    def get_intensities(self) -> np.ndarray:
        """ Return sum of intensities in subaps. Size nSubaps, same order as slopes
        """
        raise NotImplementedError("Not implemented")
 
    def set_flat(self, centro_index: int, flat: np.ndarray,):
        """ Load flat field for the given wfs
 
        Parameters:
            centro_index : (int) : index of the centroider handling the WFS
 
            flat : (np.ndarray) : New WFS flat to use
        """
        self.rtc.d_centro[centro_index].set_flat(flat, flat.shape[0])
 
    def set_dark(self, centro_index: int, dark: np.ndarray):
        """ Load dark for the given wfs
 
        Parameters:
            centro_index : (int) : index of the centroider handling the WFS
 
            dark : (np.ndarray) : New WFS dark to use
        """
        self.rtc.d_centro[centro_index].set_dark(dark, dark.shape[0])
 
    def compute_slopes(self, controller_index: int):
        """ Compute the slopes handled by a controller, and returns it
 
        Parameters:
            controller_index : (int) : Controller index that will compute its slopes
 
        Return:
            slopes : (np.ndarray) : Slopes vector
        """
        self.rtc.do_centroids(controller_index)
        return self.get_slopes(controller_index)
 
    def reset_perturbation_voltage(self, controller_index: int) -> None:
        """ Reset the perturbation voltage of the controller_index controller
        (i.e. will remove ALL perturbation voltages.)
        If you want to reset just one, see the function remove_perturbation_voltage()
 
        Parameters:
            controller_index : (int) : controller index from where to remove the buffer
        """
        self.rtc.d_control[controller_index].reset_perturb_voltage()
 
    def remove_perturbation_voltage(self, controller_index: int, name: str) -> None:
        """ Remove the perturbation voltage called <name>, from the controller number <controller_index>.
        If you want to remove all of them, see function reset_perturbation_voltage()
 
        Parameters:
            controller_index : (int) : controller index from where to remove the buffer
        """
        self.rtc.d_control[controller_index].remove_perturb_voltage(name)
 
    def get_err(self, controller_index: int) -> np.ndarray:
        """ Get integrator increment from controller_index controller
 
        Parameters:
            controller_index : (int) : controller index
        """
        return np.array(self.rtc.d_control[controller_index].d_err)
 
    def get_voltages(self, controller_index: int) -> np.ndarray:
        """ Get voltages vector (i.e. vector sent to the DM) from controller_index controller
 
        Parameters:
            controller_index : (int) : controller index
 
        Return:
            voltages : (np.ndarray) : current voltages vector
 
        """
        return np.array(self.rtc.d_control[controller_index].d_voltage)
 
    def set_integrator_law(self, controller_index: int) -> None:
        """ Set the control law to integrator (controller generic only)
            v[k] = v[k-1] + g.R.s[k]
 
        Parameters:
            controller_index: (int): controller index
        """
        self.rtc.d_control[controller_index].set_commandlaw("integrator")
 
    def set_2matrices_law(self, controller_index: int) -> None:
        """ Set the control law to 2matrices (controller generic only)
        v[k] = decayFactor.E.v[k-1] + g.R.s[k]
 
        Parameters:
            controller_index: (int): controller index
        """
        self.rtc.d_control[controller_index].set_commandlaw("2matrices")
 
    def set_modal_integrator_law(self, controller_index: int) -> None:
        """ Set the control law to 2matrices (controller generic only)
        v[k] = v[k-1] + E.g.R.s[k]
 
        Parameters:
            controller_index: (int): controller index
        """
        self.rtc.d_control[controller_index].set_commandlaw("modal_integrator")
 
    def set_decay_factor(self, controller_index: int, decay : np.ndarray) -> None:
        """ Set the decay factor used in 2matrices command law (controller generic only)
 
        Parameters :
            controller_index: (int): controller index
 
            decay : (np.ndarray) : decay factor vector
        """
        self.rtc.d_control[controller_index].set_decayFactor(decay)
 
    def set_E_matrix(self, controller_index: int, e_matrix : np.ndarray) -> None:
        """ Set the E matrix used in 2matrices or modal command law (controller generic only)
 
        Parameters :
            e_matrix : (np.ndarray) : E matrix to set
 
            controller_index: (int): controller index
        """
        self.rtc.d_control[controller_index].set_matE(e_matrix)
 
    def reset_ref_slopes(self, controller_index: int) -> None:
        """ Reset the reference slopes of each WFS handled by the specified controller
 
        Parameters:
            controller_index: (int): controller index
        """
        for centro in self.rtc.d_centro:
            centro.d_centroids_ref.reset()
 
    def set_centroider_threshold(self, centro_index: int, thresh: float) -> None:
        """ Set the threshold value of a thresholded COG
 
        Parameters:
            centro_index: (int): centroider index
 
            thresh: (float): new threshold value
        """
        self.rtc.d_centro[centro_index].set_threshold(thresh)
 
    def get_pyr_method(self, centro_index : int) -> str:
        """ Get pyramid compute method currently used
 
        Parameters:
            centro_index: (int): centroider index
 
        Return:
            method : (str) : Pyramid compute method currently used
        """
        return self.rtc.d_centro[centro_index].pyr_method
 
    def set_pyr_method(self, centro_index: int, pyr_method : int) -> None:
        """ Set the pyramid method for slopes computation
 
        Parameters :
            centro_index : (int) : centroider index
 
            pyr_method : (int) : new centroiding method (0: nosinus global
                                                1: sinus global
                                                2: nosinus local
                                                3: sinus local)
        """
        self.rtc.d_centro[centro_index].set_pyr_method(pyr_method)  # Sets the pyr method
        self.rtc.do_centroids(0)  # To be ready for the next get_slopess
        print("PYR method set to " + self.rtc.d_centro[centro_index].pyr_method)
 
 
    def set_modal_gains(self, controller_index: int, mgain: np.ndarray) -> None:
        """ Sets the modal gain (when using modal integrator control law)
 
        Parameters:
            controller_index : (int) : Controller index to modify
            
            mgain : (np.ndarray) : Modal gains to set
        """
        self.rtc.d_control[controller_index].set_modal_gains(mgain)
 
    def get_modal_gains(self, controller_index: int) -> np.ndarray:
        """ Returns the modal gains (when using modal integrator control law)
 
        Parameters:
            controller_index : (int) : Controller index to modify
 
        Return:
            mgain : (np.ndarray) : Modal gains vector currently used
        """
        return np.array(self.rtc.d_control[controller_index].d_gain)
 
    def get_masked_pix(self, centro_index: int) -> np.ndarray:
        """ Return the mask of valid pixels used by a maskedpix centroider
 
        Parameters:
            centro_index : (int): Centroider index. Must be a maskedpix centroider
 
        Return:
            mask : (np.ndarray) : Mask used
        """
        if (self.rtc.d_centro[centro_index].type != scons.CentroiderType.MASKEDPIX):
            raise TypeError("Centroider must be a maskedpix one")
        self.rtc.d_centro[centro_index].fill_mask()
        return np.array(self.rtc.d_centro[centro_index].d_mask)
 
    def get_command(self, controller_index: int) -> np.ndarray:
        """ Returns the last computed command before conversion to voltages
 
        Parameters:
            controller_index : (int, optional) : Controller index
 
        Return:
            com : (np.ndarray) : Command vector
        """
        return np.array(self.rtc.d_control[controller_index].d_com)
 
    def set_command(self, controller_index: int, com : np.ndarray) -> np.ndarray:
        """ Returns the last computed command before conversion to voltages
 
        Parameters:
            controller_index : (int, optional) : Controller index
 
            com : (np.ndarray) : Command vector to set
        """
        if(com.size != self.config.p_controllers[controller_index].nactu):
            raise ValueError("Dimension mismatch")
        self.rtc.d_control[controller_index].set_com(com, com.size)
 
    def reset_command(self, controller_index: int = None) -> None:
        """ Reset the controller_index Controller command buffer, reset all controllers if controller_index is None
 
        Parameters:
            controller_index : (int, optional) : Controller index
                                      Default is None, i.e. all controllers are reset
        """
        if (controller_index is None):  # All Dms reset
            for control in self.rtc.d_control:
                control.d_com.reset()
        else:
            self.rtc.d_control[controller_index].d_com.reset()
 
    def get_slopes_geom(self, controller_index: int) -> np.ndarray:
        """ Computes and return the slopes geom from the specified controller
 
        Parameters:
            controller_index : (int) : controller index
 
        Return:
            slopes_geom : (np.ndarray) : geometrically computed slopes
        """
        self.rtc.do_centroids_geom(controller_index)
        slopes_geom = np.array(self.rtc.d_control[controller_index].d_centroids)
        self.rtc.do_centroids(controller_index)  # To return in non-geo state
        return slopes_geom
 
    def get_selected_pix(self) -> np.ndarray:
        """ Return the pyramid image with only the selected pixels used by the full pixels centroider
 
        Return:
            selected_pix : (np.ndarray) : PWFS image with only selected pixels
        """
        if (self.config.p_centroiders[0].type != scons.CentroiderType.MASKEDPIX):
            raise TypeError("Centroider must be maskedPix")
 
        carma_centroids = self.rtc.d_control[0].d_centroids
        self.rtc.d_centro[0].fill_selected_pix(carma_centroids)
 
        return np.array(self.rtc.d_centro[0].d_selected_pix)
 
    def do_ref_slopes(self, controller_index: int) -> None:
        """ Computes and set a new reference slopes for each WFS handled by
        the specified controller
 
        Parameters:
            controller_index: (int): controller index
        """
        print("Doing reference slopes...")
        self.rtc.do_centroids_ref(controller_index)
        print("Reference slopes done")
 
    def do_control(self, controller_index: int, *, sources : SourceCompass = None, source_index : int = 0, is_wfs_phase : bool = False) -> None:
        """Computes the command from the Wfs slopes
 
        Parameters :
            controller_index: (int): controller index
 
            sources : (SourceCompass, optional) : List of phase screens of a wfs or target sutra object
                                                  If the controller is a GEO one, specify a SourceCompass instance
                                                  from WfsCompass or TargetCompass to project the corresponding phase
            
            source_index : (int, optional) : Index of the phase screen to consider inside <sources>. Default is 0
 
            is_wfs_phase : (bool, optional) : If True, sources[source_index] is a WFS phase screen.
                                              Else, it is a Target phase screen (Default)
        """
        if (self.rtc.d_control[controller_index].type == scons.ControllerType.GEO):
            if (sources is not None):
                self.rtc.d_control[controller_index].comp_dphi(sources[source_index],
                                                       is_wfs_phase)
        self.rtc.do_control(controller_index)
 
    def do_calibrate_img(self, controller_index: int) -> None:
        """ Computes the calibrated image from the Wfs image
 
        Parameters :
            controller_index: (int): controller index
        """
        self.rtc.do_calibrate_img(controller_index)
 
    def do_centroids(self, controller_index: int) -> None:
        """ Computes the centroids from the Wfs image
 
        Parameters :
            controller_index: (int): controller index
        """
        self.rtc.do_centroids(controller_index)
 
    def do_centroids_geom(self, controller_index: int) -> None:
        """ Computes the centroids geom from the Wfs image
 
        Parameters :
            controller_index: (int): controller index
        """
        self.rtc.do_centroids_geom(controller_index)
 
    def apply_control(self, controller_index: int, *, comp_voltage: bool = True) -> None:
        """ Computes the final voltage vector to apply on the DM by taking into account delay and perturbation voltages, and shape the DMs
 
        Parameters :
            controller_index: (int): controller index
 
            comp_voltage: (bool): If True (default), computes the voltage vector from the command one (delay + perturb). Else, directly applies the current voltage vector
        """
        self.rtc.apply_control(controller_index, comp_voltage)
 
    def do_clipping(self, controller_index: int) -> None:
        """ Clip the commands between vmin and vmax values set in the RTC
 
        Parameters :
            controller_index: (int): controller index
        """
        self.rtc.do_clipping(controller_index)
 
    def set_scale(self, centroider_index : int, scale : float) -> None:
        """ Update the scale factor of the centroider
 
        Parameters:
            centroider_index : (int) : Index of the centroider to update
 
            scale : (float) : scale factor to apply on slopes
        """
        self.rtc.d_centro[centroider_index].set_scale(scale)
 
    def publish(self) -> None:
        """ Publish loop data on DDS topics
 
        /!\ only with cacao enabled, requires OCTOPUS
        """
        if self.cacao:
            self.rtc.rtc.publish()
        else:
            raise AttributeError("CACAO must be enabled")