Skip to content

brahmap.core.compute_GLS_maps

Source code in brahmap/core/GLS.py
def compute_GLS_maps(
    npix: int,
    pointings: np.ndarray,
    time_ordered_data: np.ndarray,
    pointings_flag: Union[np.ndarray, None] = None,
    pol_angles: Union[np.ndarray, None] = None,
    inv_noise_cov_operator: Union[DTypeNoiseCov, None] = None,
    threshold: float = 1.0e-5,
    dtype_float: Union[DTypeFloat, None] = None,
    update_pointings_inplace: bool = True,
    gls_parameters: GLSParameters = GLSParameters(),
) -> Union[GLSResult, tuple[ProcessTimeSamples, GLSResult]]:
    if dtype_float is None:
        if pol_angles is None:
            dtype_float = time_ordered_data.dtype
        else:
            dtype_float = np.promote_types(pol_angles.dtype, time_ordered_data.dtype)

    if pol_angles is not None:
        pol_angles = pol_angles.astype(dtype=dtype_float, copy=False)

    if inv_noise_cov_operator is None:
        noise_weights = None
    else:
        noise_weights = inv_noise_cov_operator.diag

    processed_samples = ProcessTimeSamples(
        npix=npix,
        pointings=pointings,
        pointings_flag=pointings_flag,
        solver_type=gls_parameters.solver_type,
        pol_angles=pol_angles,
        noise_weights=noise_weights,
        threshold=threshold,
        dtype_float=dtype_float,
        update_pointings_inplace=update_pointings_inplace,
    )

    gls_result = compute_GLS_maps_from_PTS(
        processed_samples=processed_samples,
        time_ordered_data=time_ordered_data.astype(dtype=dtype_float, copy=False),
        inv_noise_cov_operator=inv_noise_cov_operator,
        gls_parameters=gls_parameters,
    )

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