brahmap.lbsim.LBSim_compute_GLS_maps

Computes the Generalized Least Squares (GLS) maps from litebird_sim observations.

Parameters:

Name	Type	Description	Default
nside	int	The HEALPix \(N_{side}\) resolution parameter defining the number of pixels	required
observations	Observation | List[Observation]	An instance of the Observation class or a list of the same	required
pointings	NDArray[number] | List[NDArray[number]] | None	Array of detector pointing indices mapping time samples to observed sky pixels, by default None	None
hwp	HWP | None	The Half-Wave Plate (HWP) angles or configuration, by default None	None
components	str | List[str]	A string or list defining the TOD components to be used for map-making, by default "tod"	'tod'
pointings_flag	NDArray[bool_] | None	Boolean array indicating valid pointing samples, by default None. The True value indicates a valid pointing, and the False value indicates a bad pointing. If set as None, all the pointings are considered valid	None
inv_noise_cov_operator	DTypeNoiseCov | DTypeLBSNoiseCov | None	The inverse noise covariance linear operator (\(N^{-1}\)), by default None	None
threshold	float	The condition number threshold used to flag degenerate or under-sampled pixels, by default 1.0e-5	1e-05
dtype_float	DTypeFloat	The data type to use for floating point arrays, by default np.float64	float64
LBSim_gls_parameters	LBSimGLSParameters	The parameter configuration dictating the map-making behavior, by default LBSimGLSParameters()	LBSimGLSParameters()
x0	NDArray[number] | None	Initial guess for the GLS solution in the form of interleaved maps (e.g. \([I_1, Q_1, U_1, I_2, Q_2, U_2, \dots]\)), by default None	None

Returns:

Type	Description
LBSimGLSResult | tuple[LBSimProcessTimeSamples, LBSimGLSResult]	The dataclass containing the final output from the GLS map-maker, optionally returning the processed samples container

Source code in brahmap/lbsim/lbsim_GLS.py

def LBSim_compute_GLS_maps(
    nside: int,
    observations: lbs.Observation | List[lbs.Observation],
    pointings: npt.NDArray[np.number] | List[npt.NDArray[np.number]] | None = None,
    hwp: lbs.HWP | None = None,
    components: str | List[str] = "tod",
    pointings_flag: npt.NDArray[np.bool_] | None = None,
    inv_noise_cov_operator: DTypeNoiseCov | DTypeLBSNoiseCov | None = None,
    threshold: float = 1.0e-5,
    dtype_float: DTypeFloat = np.float64,
    LBSim_gls_parameters: LBSimGLSParameters = LBSimGLSParameters(),
    x0: npt.NDArray[np.number] | None = None,
) -> LBSimGLSResult | tuple[LBSimProcessTimeSamples, LBSimGLSResult]:
    """Computes the Generalized Least Squares (GLS) maps from
    `litebird_sim` observations.

    Parameters
    ----------
    nside : int
        The HEALPix $N_{side}$ resolution parameter defining the number of pixels
    observations : lbs.Observation | List[lbs.Observation]
        An instance of the `Observation` class or a list of the same
    pointings : npt.NDArray[np.number] | List[npt.NDArray[np.number]] | None, optional
        Array of detector pointing indices mapping time samples to observed sky pixels,
        by default `None`
    hwp : lbs.HWP | None, optional
        The Half-Wave Plate (HWP) angles or configuration, by default `None`
    components : str | List[str], optional
        A string or list defining the TOD components to be used for map-making, by
        default `"tod"`
    pointings_flag : npt.NDArray[np.bool_] | None, optional
        Boolean array indicating valid pointing samples, by default `None`.
        The `True` value indicates a valid pointing, and the `False`
        value indicates a bad pointing. If set as `None`, all the
        pointings are considered valid
    inv_noise_cov_operator : DTypeNoiseCov | DTypeLBSNoiseCov | None, optional
        The inverse noise covariance linear operator ($N^{-1}$), by default `None`
    threshold : float, optional
        The condition number threshold used to flag degenerate or
        under-sampled pixels, by default `1.0e-5`
    dtype_float : DTypeFloat, optional
        The data type to use for floating point arrays, by default
        `np.float64`
    LBSim_gls_parameters : LBSimGLSParameters, optional
        The parameter configuration dictating the map-making behavior, by
        default `LBSimGLSParameters()`
    x0 : npt.NDArray[np.number] | None, optional
        Initial guess for the GLS solution in the form of interleaved
        maps (e.g. $[I_1, Q_1, U_1, I_2, Q_2, U_2, \\dots]$), by default `None`

    Returns
    -------
    LBSimGLSResult | tuple[LBSimProcessTimeSamples, LBSimGLSResult]
        The dataclass containing the final output from the GLS map-maker,
        optionally returning the processed samples container
    """
    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=processed_samples.obs_list,
            target=components[0],
            other_components=components[1:],
            factor=1.0,
        )

    time_ordered_data = np.concatenate(
        [getattr(obs, components[0]) for obs in processed_samples.obs_list], 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,
        x0=x0,
    )

    lbsim_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, lbsim_gls_result
    else:
        del processed_samples
        gc.collect()
        return lbsim_gls_result