brahmap.lbsim.LBSim_compute_GLS_maps

summary

Parameters:

Name	Type	Description	Default
nside	int	description	required
observations	Union[Observation, List[Observation]]	description	required
pointings	Union[ndarray, List[ndarray], None]	description, by default None	None
hwp	Optional[HWP]	description, by default None	None
components	Union[str, List[str]]	description, by default "tod"	'tod'
pointings_flag	Optional[ndarray]	description, by default None	None
inv_noise_cov_operator	Union[DTypeNoiseCov, DTypeLBSNoiseCov, None]	description, by default None	None
threshold	float	description, by default 1.0e-5	1e-05
dtype_float	Optional[DTypeFloat]	description, by default None	None
LBSim_gls_parameters	LBSimGLSParameters	description, by default LBSimGLSParameters()	LBSimGLSParameters()

Returns:

Type	Description
Union[LBSimGLSResult, tuple[LBSimProcessTimeSamples, LBSimGLSResult]]	description

Source code in brahmap/lbsim/lbsim_GLS.py

def LBSim_compute_GLS_maps(
    nside: int,
    observations: Union[lbs.Observation, List[lbs.Observation]],
    pointings: Union[np.ndarray, List[np.ndarray], None] = None,
    hwp: Optional[lbs.HWP] = None,
    components: Union[str, List[str]] = "tod",
    pointings_flag: Optional[np.ndarray] = None,
    inv_noise_cov_operator: Union[DTypeNoiseCov, DTypeLBSNoiseCov, None] = None,
    threshold: float = 1.0e-5,
    dtype_float: Optional[DTypeFloat] = None,
    LBSim_gls_parameters: LBSimGLSParameters = LBSimGLSParameters(),
) -> Union[LBSimGLSResult, tuple[LBSimProcessTimeSamples, LBSimGLSResult]]:
    """_summary_

    Parameters
    ----------
    nside : int
        _description_
    observations : Union[lbs.Observation, List[lbs.Observation]]
        _description_
    pointings : Union[np.ndarray, List[np.ndarray], None], optional
        _description_, by default None
    hwp : Optional[lbs.HWP], optional
        _description_, by default None
    components : Union[str, List[str]], optional
        _description_, by default "tod"
    pointings_flag : Optional[np.ndarray], optional
        _description_, by default None
    inv_noise_cov_operator : Union[DTypeNoiseCov, DTypeLBSNoiseCov, None], optional
        _description_, by default None
    threshold : float, optional
        _description_, by default 1.0e-5
    dtype_float : Optional[DTypeFloat], optional
        _description_, by default None
    LBSim_gls_parameters : LBSimGLSParameters, optional
        _description_, by default LBSimGLSParameters()

    Returns
    -------
    Union[LBSimGLSResult, tuple[LBSimProcessTimeSamples, LBSimGLSResult]]
        _description_
    """
    if inv_noise_cov_operator is None:
        noise_weights = None
    else:
        noise_weights = inv_noise_cov_operator.diag

    processed_samples = LBSimProcessTimeSamples(
        nside=nside,
        observations=observations,
        pointings=pointings,
        hwp=hwp,
        pointings_flag=pointings_flag,
        solver_type=LBSim_gls_parameters.solver_type,
        noise_weights=noise_weights,
        output_coordinate_system=LBSim_gls_parameters.output_coordinate_system,
        threshold=threshold,
        dtype_float=dtype_float,
    )

    if isinstance(components, str):
        components = [components]

    if len(components) > 1:
        lbs.mapmaking.destriper._sum_components_into_obs(
            obs_list=observations,
            target=components[0],
            other_components=components[1:],
            factor=1.0,
        )

    time_ordered_data = np.concatenate(
        [getattr(obs, components[0]) for obs in observations], axis=None
    )

    gls_result = compute_GLS_maps_from_PTS(
        processed_samples=processed_samples,
        time_ordered_data=time_ordered_data,
        inv_noise_cov_operator=inv_noise_cov_operator,
        gls_parameters=LBSim_gls_parameters,
    )

    gls_result = LBSimGLSResult(
        nside=nside,
        coordinate_system=LBSim_gls_parameters.output_coordinate_system,
        **asdict(gls_result),
    )

    if LBSim_gls_parameters.return_processed_samples:
        return processed_samples, gls_result
    else:
        del processed_samples
        gc.collect()
        return gls_result