brahmap.lbsim.LBSim_InvNoiseCovLO_Circulant

Bases: BlockDiagInvNoiseCovLO

summary

Parameters:

Name	Type	Description	Default
obs	Union[Observation, List[Observation]]	description	required
input	Union[dict, Union[ndarray, List]]	description	required
input_type	Literal['covariance', 'power_spectrum']	description, by default "power_spectrum"	'power_spectrum'
dtype	DTypeFloat	description, by default np.float64	float64

Source code in brahmap/lbsim/lbsim_noise_operators.py

class LBSim_InvNoiseCovLO_Circulant(BlockDiagInvNoiseCovLO):
    """_summary_

    Parameters
    ----------
    obs : Union[lbs.Observation, List[lbs.Observation]]
        _description_
    input : Union[dict, Union[np.ndarray, List]]
        _description_
    input_type : Literal["covariance", "power_spectrum"], optional
        _description_, by default "power_spectrum"
    dtype : DTypeFloat, optional
        _description_, by default np.float64
    """

    def __init__(
        self,
        obs: Union[lbs.Observation, List[lbs.Observation]],
        input: Union[dict, Union[np.ndarray, List]],
        input_type: Literal["covariance", "power_spectrum"] = "power_spectrum",
        dtype: DTypeFloat = np.float64,
    ):
        if isinstance(obs, lbs.Observation):
            obs_list = [obs]
        else:
            obs_list = obs

        block_size = []
        block_input = []

        for obs in obs_list:
            if isinstance(input, dict):
                # if input is a dict
                for det_idx in range(obs.n_detectors):
                    block_size.append(obs.n_samples)
                    resized_input = self._resize_input(
                        new_size=obs.n_samples,
                        input=input[obs.name[det_idx]],
                        input_type=input_type,
                        dtype=dtype,
                    )
                    block_input.append(resized_input)
            else:
                for det_idx in range(obs.n_detectors):
                    # if input is an array or a list, it will be taken as same for all the detectors available in the observation
                    block_size.append(obs.n_samples)
                    resized_input = self._resize_input(
                        new_size=obs.n_samples,
                        input=input,
                        input_type=input_type,
                        dtype=dtype,
                    )
                    block_input.append(resized_input)

        super(LBSim_InvNoiseCovLO_Circulant, self).__init__(
            InvNoiseCovLO_Circulant,
            block_size=block_size,
            block_input=block_input,
            input_type=input_type,
            dtype=dtype,
        )

    def _resize_input(self, new_size, input, input_type, dtype):
        if input_type == "covariance":
            # if the size of the returned array is smaller than new_size, it
            # will be captured by the InvNoiseCovLO_Circulant class
            # automatically

            # Slicing the input array here is probably not the best choice as
            # it breaks the symmetry of the covariance and renders it
            # non-circulant. Same goes for slicing the covariance computed
            # through power spectrum. The best solution would be to create per
            # observation, per detector operators independently and supply them
            # to `BlockDiagInvNoiseCovLO`
            return input[:new_size]
        elif input_type == "power_spectrum":
            input_size = len(input)
            if input_size > new_size:
                new_input = np.fft.ifft(input)[:new_size]  # new covariance
                new_input = np.fft.fft(new_input).real.astype(
                    dtype=dtype,
                    copy=False,
                )  # new ps
                return new_input
            else:
                # If input size is equal to expected size, it will be fine.
                # If it is smaller, InvNoiseCovLO_Circulant class will
                # throw an error automatically
                return input