brahmap.lbsim.LBSim_InvNoiseCovLO_Toeplitz

Bases: BlockDiagInvNoiseCovLO

summary

Note that the observation length is either n or n-1.

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'
operator	InvNoiseCovLinearOperator	description, by default InvNoiseCovLO_Toeplitz01	InvNoiseCovLO_Toeplitz01
dtype	DTypeFloat	description, by default np.float64	float64
extra_kwargs	Dict[str, Any]	description, by default {}	{}

Source code in brahmap/lbsim/lbsim_noise_operators.py

class LBSim_InvNoiseCovLO_Toeplitz(BlockDiagInvNoiseCovLO):
    """_summary_

    Note that the observation length is either n or n-1.

    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"
    operator : InvNoiseCovLinearOperator, optional
        _description_, by default InvNoiseCovLO_Toeplitz01
    dtype : DTypeFloat, optional
        _description_, by default np.float64
    extra_kwargs : Dict[str, Any], optional
        _description_, by default {}
    """

    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",
        operator: InvNoiseCovLinearOperator = InvNoiseCovLO_Toeplitz01,
        dtype: DTypeFloat = np.float64,
        extra_kwargs: Dict[str, Any] = {},
    ):
        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:
                # if input is an array or a list, it will be taken as same for all the detectors available in the observation
                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,
                        input_type=input_type,
                        dtype=dtype,
                    )
                    block_input.append(resized_input)

        super(LBSim_InvNoiseCovLO_Toeplitz, self).__init__(
            operator,
            block_size=block_size,
            block_input=block_input,
            input_type=input_type,
            dtype=dtype,
            extra_kwargs=extra_kwargs,
        )

    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_Toeplitz0x class
            # automatically
            return input[:new_size]
        elif input_type == "power_spectrum":
            input_size = len(input)
            ex_size1 = 2 * new_size - 1  # expected size of ps array (2n-1)
            ex_size2 = 2 * new_size - 2  # expected size of ps array (2n-2)
            if input_size > ex_size2 and input_size > ex_size1:
                new_input = np.fft.ifft(input)[
                    :new_size
                ]  # covariance of size `new_size`
                new_input = np.concatenate(
                    [new_input, new_input[1:-1][::-1]]
                )  # full covariance of size `2*new_size - 2`
                new_input = np.fft.fft(new_input).real.astype(
                    dtype=dtype, copy=False
                )  # full ps of size `2*new_size - 2`
                return new_input
            else:
                # If input size is equal to expected size, it will be fine.
                # If it is smaller, InvNoiseCovLO_Toeplitz0x class will
                # throw an error automatically
                return input