voidfinder

voidfinder

Author: Corentin Ravoux

Description : Void and Over-density finder for Lya Tomographic maps. Watershed and Simple Spherical techniques are available. Tested on irene and cobalt (CCRT)

VoidFinder

VoidFinder(pwd, map_name, params_void_finder, map_shape=None, map_size=None, map_property_file=None, number_core=1, find_cluster=False, split_map=None, split_overlap=None, restart=False, delete_option='CLUSTERS')

Bases: object

Detect voids (or over-density clusters) in a tomographic map.

Two algorithms are available: a simple spherical growth finder (SPHERICAL) and a watershed segmentation (WATERSHED). Large maps can be split into overlapping chunks processed in parallel and merged, with optional restart from temporary per-chunk catalogs.

Attributes:

Name	Type	Description
params_void_finder	dict	Finder parameters (method, threshold, average, minimal_radius, maximal_radius, radius_step, dist_clusters).
find_cluster	bool	Search over-densities instead of voids if True.
delete_option	str	Overlap resolution (CLUSTERS/ITERATION/NONE).

Initialise the void finder and its report log.

Parameters:

Name	Type	Description	Default
pwd	str	Output directory for the catalog and log.	required
map_name	str	Path to the tomographic map binary.	required
params_void_finder	dict	Finder parameters (see class docstring).	required
map_shape	tuple[int]	Map pixel shape (read if None).	None
map_size	tuple[float]	Map physical size (read if None).	None
map_property_file	str	Map property/pickle file.	None
number_core	int	Number of parallel processes.	1
find_cluster	bool	Find over-densities instead of voids.	False
split_map	tuple[int]	Chunk grid, e.g. (2, 2).	None
split_overlap	tuple[int]	Chunk overlap in pixels.	None
restart	bool	Resume from temporary chunk catalogs.	False
delete_option	str	Overlap resolution policy.	'CLUSTERS'

Source code in lelantos/voidfinder.py

def __init__(
    self,
    pwd,
    map_name,
    params_void_finder,
    map_shape=None,
    map_size=None,
    map_property_file=None,
    number_core=1,
    find_cluster=False,
    split_map=None,
    split_overlap=None,
    restart=False,
    delete_option="CLUSTERS",
):
    """Initialise the void finder and its report log.

    Args:
        pwd (str): Output directory for the catalog and log.
        map_name (str): Path to the tomographic map binary.
        params_void_finder (dict): Finder parameters (see class docstring).
        map_shape (tuple[int], optional): Map pixel shape (read if None).
        map_size (tuple[float], optional): Map physical size (read if None).
        map_property_file (str, optional): Map property/pickle file.
        number_core (int, optional): Number of parallel processes.
        find_cluster (bool, optional): Find over-densities instead of voids.
        split_map (tuple[int], optional): Chunk grid, e.g. ``(2, 2)``.
        split_overlap (tuple[int], optional): Chunk overlap in pixels.
        restart (bool, optional): Resume from temporary chunk catalogs.
        delete_option (str, optional): Overlap resolution policy.
    """
    self.pwd = pwd
    self.params_void_finder = params_void_finder
    self.number_core = number_core
    self.find_cluster = find_cluster
    self.split_map = split_map
    self.split_overlap = split_overlap
    self.delete_option = delete_option
    self.restart = restart

    self.map_name = map_name
    self.map_shape = map_shape
    self.map_size = map_size
    self.map_property_file = map_property_file
    self.map_mpc_per_pixel = None
    self.map_coordinate_transform = None
    self.map_Omega_m = None
    self.map_boundary_cartesian_coord = None
    self.map_boundary_sky_coord = None

    log_name = f"void_finder_report_{self.get_name_catalog()}.txt"
    self.log = utils.create_report_log(name=os.path.join(self.pwd, log_name))

    self.save_temporary_file = False
    self.temporary_file_name = (
        "{}" + f"_temporary_void_catalog_{self.get_name_catalog()}.fits"
    )

initialize_finder

initialize_finder()

Load the map and cache its geometry/cosmology metadata.

Returns:

Type	Description
	numpy.ndarray: The map array to run the finder on.

Source code in lelantos/voidfinder.py

def initialize_finder(self):
    """Load the map and cache its geometry/cosmology metadata.

    Returns:
        numpy.ndarray: The map array to run the finder on.
    """
    tomographic_map = tomographic_objects.TomographicMap.init_classic(
        name=self.map_name,
        shape=self.map_shape,
        size=self.map_size,
        property_file=self.map_property_file,
    )
    tomographic_map.read()
    if self.map_shape is None:
        self.map_shape = tomographic_map.shape
    if self.map_size is None:
        self.map_size = tomographic_map.size
    self.map_mpc_per_pixel = tomographic_map.mpc_per_pixel
    self.map_coordinate_transform = tomographic_map.coordinate_transform
    self.map_Omega_m = tomographic_map.Omega_m
    self.map_boundary_cartesian_coord = tomographic_map.boundary_cartesian_coord
    self.map_boundary_sky_coord = tomographic_map.boundary_sky_coord
    map_array = tomographic_map.map_array
    del tomographic_map
    return map_array

find_voids

find_voids()

Run the finder on the whole map or on chunks and write the catalog.

Returns:

Name	Type	Description
str		Path of the written void catalog.

Source code in lelantos/voidfinder.py

def find_voids(self):
    """Run the finder on the whole map or on chunks and write the catalog.

    Returns:
        str: Path of the written void catalog.
    """
    map_array = self.initialize_finder()
    if self.split_map is None:
        return self.find_voids_single_map(map_array)
    else:
        return self.find_voids_map_split(map_array)

find_voids_single_map

find_voids_single_map(map_array)

Run the finder on the full map (no splitting) and save the catalog.

Parameters:

Name	Type	Description	Default
map_array	ndarray	The full map array.	required

Returns:

Name	Type	Description
str		Path of the written void catalog.

Raises:

Type	Description
ValueError	If the finder method is neither WATERSHED nor SPHERICAL.

Source code in lelantos/voidfinder.py

def find_voids_single_map(self, map_array):
    """Run the finder on the full map (no splitting) and save the catalog.

    Args:
        map_array (numpy.ndarray): The full map array.

    Returns:
        str: Path of the written void catalog.

    Raises:
        ValueError: If the finder method is neither WATERSHED nor SPHERICAL.
    """
    if self.params_void_finder["method"].upper() == "WATERSHED":
        (radius, coord, other_array, other_array_name) = self.find_voids_watershed(
            (self.map_name, self.map_mpc_per_pixel, map_array)
        )
    elif self.params_void_finder["method"].upper() == "SPHERICAL":
        (radius, coord, other_array, other_array_name) = self.find_voids_sphere(
            (self.map_name, self.map_mpc_per_pixel, map_array)
        )
    else:
        raise ValueError(
            "The method_void chosen is not implemented, try : WATERSHED or SPHERICAL"
        )
    del map_array
    name = self.save_voids(
        radius,
        coord,
        other_array,
        other_array_name,
        self.map_coordinate_transform,
        self.map_Omega_m,
        self.map_boundary_cartesian_coord,
        self.map_boundary_sky_coord,
    )
    return name

find_voids_map_split

find_voids_map_split(map_array)

Run the finder on overlapping chunks, then merge and clean voids.

Splits the map, runs the finder per chunk (optionally in parallel and with restart from temporary catalogs), merges the chunk results while removing overlap duplicates, and writes the final catalog.

Parameters:

Name	Type	Description	Default
map_array	ndarray	The full map array.	required

Returns:

Name	Type	Description
str		Path of the written void catalog.

Raises:

Type	Description
ValueError	If the finder method is neither WATERSHED nor SPHERICAL.

Source code in lelantos/voidfinder.py

def find_voids_map_split(self, map_array):
    """Run the finder on overlapping chunks, then merge and clean voids.

    Splits the map, runs the finder per chunk (optionally in parallel and
    with restart from temporary catalogs), merges the chunk results while
    removing overlap duplicates, and writes the final catalog.

    Args:
        map_array (numpy.ndarray): The full map array.

    Returns:
        str: Path of the written void catalog.

    Raises:
        ValueError: If the finder method is neither WATERSHED nor SPHERICAL.
    """
    self.save_temporary_file = True
    self.log.add(f"Splitting of the map in chunks: {self.split_map}")
    self.log.add("Saving of temporary files activated")
    map_chunks = self.split_map_in_chunks(map_array)
    del map_array
    list_index_map_chunks = [
        f"{i:03d}" + f"{j:03d}"
        for j in range(self.split_map[1])
        for i in range(self.split_map[0])
    ]
    if self.restart:
        self.log.add(
            "Restarting of the void finder, searching for temporary files..."
        )
        other_array_name_restart = [
            "VALUE",
            "MEAN",
        ]
        (map_chunks, list_index_map_chunks_restart) = self.restart_calculation(
            map_chunks, list_index_map_chunks, other_array_name_restart
        )
        list_index_map_chunks = list_index_map_chunks_restart
    if self.params_void_finder["method"] == "WATERSHED":
        if self.number_core > 1:
            pool = mp.Pool(self.number_core)
            list_map_name = [
                map_chunks[list_index_map_chunks[i]]["map_name"]
                for i in range(len(list_index_map_chunks))
            ]
            list_map_mpc_per_pixel = [
                map_chunks[list_index_map_chunks[i]]["map_mpc_per_pixel"]
                for i in range(len(list_index_map_chunks))
            ]
            list_map_array = [
                map_chunks[list_index_map_chunks[i]]["map_array"]
                for i in range(len(list_index_map_chunks))
            ]
            out_finder = pool.map(
                self.find_voids_watershed,
                zip(list_map_name, list_map_mpc_per_pixel, list_map_array),
            )
            del list_map_name, list_map_mpc_per_pixel, list_map_array
            for i in range(len(list_index_map_chunks)):
                map_chunks[list_index_map_chunks[i]]["radius"] = out_finder[i][0]
                map_chunks[list_index_map_chunks[i]]["coord"] = out_finder[i][1]
                map_chunks[list_index_map_chunks[i]]["other_array"] = out_finder[i][
                    2
                ]
                map_chunks[list_index_map_chunks[i]][
                    "other_array_name"
                ] = out_finder[i][3]
        else:
            for i in range(len(list_index_map_chunks)):
                out_finder = self.find_voids_watershed(
                    (
                        map_chunks[list_index_map_chunks[i]]["map_name"],
                        map_chunks[list_index_map_chunks[i]]["map_mpc_per_pixel"],
                        map_chunks[list_index_map_chunks[i]]["map_array"],
                    )
                )
                map_chunks[list_index_map_chunks[i]]["radius"] = out_finder[0]
                map_chunks[list_index_map_chunks[i]]["coord"] = out_finder[1]
                map_chunks[list_index_map_chunks[i]]["other_array"] = out_finder[2]
                map_chunks[list_index_map_chunks[i]][
                    "other_array_name"
                ] = out_finder[3]
    elif self.params_void_finder["method"] == "SPHERICAL":
        for i in range(len(list_index_map_chunks)):
            out_finder = self.find_voids_sphere(
                (
                    map_chunks[list_index_map_chunks[i]]["map_name"],
                    map_chunks[list_index_map_chunks[i]]["map_mpc_per_pixel"],
                    map_chunks[list_index_map_chunks[i]]["map_array"],
                )
            )
            map_chunks[list_index_map_chunks[i]]["radius"] = out_finder[0]
            map_chunks[list_index_map_chunks[i]]["coord"] = out_finder[1]
            map_chunks[list_index_map_chunks[i]]["other_array"] = out_finder[2]
            map_chunks[list_index_map_chunks[i]]["other_array_name"] = out_finder[3]
    else:
        raise ValueError(
            "The method_void chosen is not implemented, try : WATERSHED or SPHERICAL"
        )
    self.log.add("End of the split finding procedure. Merging catalogs")
    (radius, coord, other_array, other_array_name) = self.merge_chunks(map_chunks)
    coord_clean, radius_clean, other_array_clean = self.delete_voids(
        self.map_mpc_per_pixel, radius, coord, other_array=other_array, mpc=True
    )
    name = self.save_voids(
        radius_clean,
        coord_clean,
        other_array_clean,
        other_array_name,
        self.map_coordinate_transform,
        self.map_Omega_m,
        self.map_boundary_cartesian_coord,
        self.map_boundary_sky_coord,
    )
    self.delete_temporary_files(map_chunks, list_index_map_chunks)
    del map_chunks, list_index_map_chunks
    return name

split_map_in_chunks

split_map_in_chunks(map_array)

Split the map into overlapping transverse chunks.

Parameters:

Name	Type	Description	Default
map_array	ndarray	The full map array.	required

Returns:

Name	Type	Description
dict		Per-chunk dict keyed by "iiijjj" holding the sub-array,
		its name, size, pixel scale and cartesian origin (map_min).

Source code in lelantos/voidfinder.py

def split_map_in_chunks(self, map_array):
    """Split the map into overlapping transverse chunks.

    Args:
        map_array (numpy.ndarray): The full map array.

    Returns:
        dict: Per-chunk dict keyed by ``"iiijjj"`` holding the sub-array,
        its name, size, pixel scale and cartesian origin (``map_min``).
    """
    subIntervalx = self.map_shape[0] // self.split_map[0]
    subIntervaly = self.map_shape[1] // self.split_map[1]
    if self.split_overlap is None:
        overlaping_x, overlaping_y = 0, 0
    elif type(self.split_overlap) == int:
        overlaping_x, overlaping_y = self.split_overlap, self.split_overlap
    else:
        overlaping_x, overlaping_y = self.split_overlap[0], self.split_overlap[1]
    map_chunks = {}
    for i in range(self.split_map[0]):
        for j in range(self.split_map[1]):
            map_chunks[f"{i:03d}" + f"{j:03d}"] = {}
            if (i == self.split_map[0] - 1) & (i == 0):
                pixel_x_interval = [0, self.map_shape[0]]
            elif i == 0:
                pixel_x_interval = [0, subIntervalx + overlaping_x]
            elif i == self.split_map[0] - 1:
                pixel_x_interval = [
                    i * subIntervalx - overlaping_x,
                    self.map_shape[0],
                ]
            else:
                pixel_x_interval = [
                    i * subIntervalx - overlaping_x,
                    (i + 1) * subIntervalx + overlaping_x,
                ]
            if (j == self.split_map[1] - 1) & (j == 0):
                pixel_y_interval = [0, self.map_shape[1]]
            elif j == 0:
                pixel_y_interval = [0, subIntervaly + overlaping_y]
            elif j == self.split_map[1] - 1:
                pixel_y_interval = [
                    j * subIntervaly - overlaping_y,
                    self.map_shape[1],
                ]
            else:
                pixel_y_interval = [
                    j * subIntervaly - overlaping_y,
                    (j + 1) * subIntervaly + overlaping_y,
                ]
            min_x_interval = pixel_x_interval[0] * self.map_mpc_per_pixel[0]
            min_y_interval = pixel_y_interval[0] * self.map_mpc_per_pixel[1]

            map_shape = (
                pixel_x_interval[1] - pixel_x_interval[0],
                pixel_y_interval[1] - pixel_y_interval[0],
                self.map_shape[2],
            )
            map_size = utils.get_map_size(map_shape, self.map_mpc_per_pixel)
            map_chunks[f"{i:03d}" + f"{j:03d}"][
                "map_name"
            ] = f"{self.map_name}_{i:03d}{j:03d}"
            map_chunks[f"{i:03d}" + f"{j:03d}"][
                "map_mpc_per_pixel"
            ] = self.map_mpc_per_pixel
            map_chunks[f"{i:03d}" + f"{j:03d}"]["map_array"] = map_array[
                pixel_x_interval[0] : pixel_x_interval[1],
                pixel_y_interval[0] : pixel_y_interval[1],
                :,
            ]
            map_chunks[f"{i:03d}" + f"{j:03d}"]["map_size"] = map_size
            map_chunks[f"{i:03d}" + f"{j:03d}"]["map_min"] = (
                min_x_interval,
                min_y_interval,
                0,
            )
    return map_chunks

merge_chunks

merge_chunks(map_chunks)

Concatenate per-chunk voids, dropping those inside overlap borders.

Parameters:

Name	Type	Description	Default
map_chunks	dict	Per-chunk results from the finder.	required

Returns:

Name	Type	Description
tuple		(radius, coord, other_array, other_array_name) for the
		merged catalog (coordinates shifted to the global frame).

Source code in lelantos/voidfinder.py

def merge_chunks(self, map_chunks):
    """Concatenate per-chunk voids, dropping those inside overlap borders.

    Args:
        map_chunks (dict): Per-chunk results from the finder.

    Returns:
        tuple: ``(radius, coord, other_array, other_array_name)`` for the
        merged catalog (coordinates shifted to the global frame).
    """
    radius_to_contatenate = []
    coord_to_contatenate = []
    if self.split_overlap is not None:
        if type(self.split_overlap) == int:
            overlaping_x, overlaping_y = self.split_overlap, self.split_overlap
        else:
            overlaping_x, overlaping_y = (
                self.split_overlap[0],
                self.split_overlap[1],
            )
    other_array_name = map_chunks[list(map_chunks.keys())[0]]["other_array_name"]
    other_array = [[] for i in range(len(other_array_name))]
    for i in range(self.split_map[0]):
        for j in range(self.split_map[1]):
            coord_chunks = map_chunks[f"{i:03d}" + f"{j:03d}"]["coord"]
            mpc_per_pixel = map_chunks[f"{i:03d}" + f"{j:03d}"]["map_mpc_per_pixel"]
            if coord_chunks.shape[0] != 0:
                if self.split_overlap is not None:
                    if (i == self.split_map[0] - 1) & (i == 0):
                        pixel_x_interval = [0.0, 0.0]
                    elif i == 0:
                        pixel_x_interval = [0, overlaping_x]
                    elif i == self.split_map[0] - 1:
                        pixel_x_interval = [overlaping_x, 0]
                    else:
                        pixel_x_interval = [overlaping_x, overlaping_x]
                    if (j == self.split_map[1] - 1) & (j == 0):
                        pixel_y_interval = [0, 0]
                    elif j == 0:
                        pixel_y_interval = [0, overlaping_y]
                    elif j == self.split_map[1] - 1:
                        pixel_y_interval = [overlaping_y, 0]
                    else:
                        pixel_y_interval = [overlaping_y, overlaping_y]
                    mask = (
                        coord_chunks[:, 0] > pixel_x_interval[0] * mpc_per_pixel[0]
                    )
                    mask &= coord_chunks[:, 0] < map_chunks[
                        f"{i:03d}" + f"{j:03d}"
                    ]["map_size"][0] - (pixel_x_interval[1] * mpc_per_pixel[0])
                    mask &= (
                        coord_chunks[:, 1] > pixel_y_interval[0] * mpc_per_pixel[1]
                    )
                    mask &= coord_chunks[:, 1] < map_chunks[
                        f"{i:03d}" + f"{j:03d}"
                    ]["map_size"][1] - (pixel_y_interval[1] * mpc_per_pixel[1])
                else:
                    mask = np.full(coord_chunks.shape[0], True)
                radius_to_contatenate.append(
                    map_chunks[f"{i:03d}" + f"{j:03d}"]["radius"][mask]
                )
                for k in range(len(other_array_name)):
                    other_array[k].append(
                        np.array(
                            map_chunks[f"{i:03d}" + f"{j:03d}"]["other_array"][k]
                        )[mask]
                    )
                coord_to_contatenate.append(
                    (
                        map_chunks[f"{i:03d}" + f"{j:03d}"]["coord"]
                        + np.array(map_chunks[f"{i:03d}" + f"{j:03d}"]["map_min"])
                    )[mask]
                )
    if len(radius_to_contatenate) == 0:
        radius = np.empty(0)
        coord = np.empty(0)
    else:
        radius = np.concatenate(radius_to_contatenate, axis=0)
        coord = np.concatenate(coord_to_contatenate, axis=0)
    for k in range(len(other_array_name)):
        other_array[k] = np.concatenate(other_array[k], axis=0)
    return (radius, coord, other_array, other_array_name)

find_voids_watershed

find_voids_watershed(parameters)

Detect voids by watershed clustering of under-density pixels.

Under-threshold pixels are clustered; each cluster gives a void centre (extremum pixel), an effective radius (from its volume) and its max/mean delta. Small voids are removed and overlaps cleaned.

Parameters:

Name	Type	Description	Default
parameters	tuple	(map_name, map_mpc_per_pixel, map_array).	required

Returns:

Name	Type	Description
tuple		(radius, coord, other_array, other_array_name) with
		other_array_name = ["VALUE", "MEAN"].

Source code in lelantos/voidfinder.py

def find_voids_watershed(self, parameters):
    """Detect voids by watershed clustering of under-density pixels.

    Under-threshold pixels are clustered; each cluster gives a void centre
    (extremum pixel), an effective radius (from its volume) and its max/mean
    delta. Small voids are removed and overlaps cleaned.

    Args:
        parameters (tuple): ``(map_name, map_mpc_per_pixel, map_array)``.

    Returns:
        tuple: ``(radius, coord, other_array, other_array_name)`` with
        ``other_array_name = ["VALUE", "MEAN"]``.
    """
    map_name, map_mpc_per_pixel, map_array = parameters
    self.log.add(f"Beginning of the Watershed finding for the map {map_name}")
    if self.find_cluster:
        mask = map_array < self.params_void_finder["threshold"]
    else:
        mask = map_array > self.params_void_finder["threshold"]
    index_under_density = np.argwhere(mask)
    self.log.add(
        f"Number of pixels for the map {map_name} = {len(index_under_density)}"
    )
    map_under_density = map_array[mask]
    del map_array
    cluster_map, clusters = self.create_watershed_clusters(index_under_density)
    self.log.add(f"Pixel clusters created for the map {map_name}")
    centers = np.zeros((len(clusters), 3))
    radius_shed = np.zeros(len(clusters))
    delta_max = np.zeros((len(clusters)))
    delta_mean = np.zeros((len(clusters)))
    volume_cell = map_mpc_per_pixel[0] * map_mpc_per_pixel[1] * map_mpc_per_pixel[2]
    mask_clust = None
    for i in range(len(clusters)):
        mask_clust = cluster_map == clusters[i]
        if self.find_cluster:
            arg_center = np.argmin(map_under_density[mask_clust])
        else:
            arg_center = np.argmax(map_under_density[mask_clust])
        delta_max[i] = map_under_density[mask_clust][arg_center]
        delta_mean[i] = np.mean(map_under_density[mask_clust])
        centers[i] = index_under_density[mask_clust][arg_center]
        volume_shed = len(map_under_density[mask_clust]) * volume_cell
        radius_shed[i] = ((3 * volume_shed) / (4 * np.pi)) ** (1 / 3)
    self.log.add(
        f"Computation of radius and center finished for the map {map_name}"
    )
    mask_radius = radius_shed > self.params_void_finder["minimal_radius"]
    radius = radius_shed[mask_radius]
    coord = centers[mask_radius]
    delta_max = delta_max[mask_radius]
    delta_mean = delta_mean[mask_radius]
    self.log.add(f"Masking of low radius done for the map {map_name}")
    coord_clean, radius_clean, other_array_clean = self.delete_voids(
        map_mpc_per_pixel, radius, coord, other_array=[delta_max, delta_mean]
    )
    other_array_name = ["VALUE", "MEAN"]
    coord_clean = self.convert_to_Mpc(map_mpc_per_pixel, coord_clean)
    del (
        mask,
        mask_clust,
        mask_radius,
        cluster_map,
        clusters,
        map_under_density,
        centers,
        index_under_density,
        radius_shed,
    )
    self.log.add(f"End of the Watershed finding for the map {map_name}")
    if self.save_temporary_file:
        self.save_temporary_catalog(
            map_name, radius_clean, coord_clean, other_array_clean, other_array_name
        )
    return (radius_clean, coord_clean, other_array_clean, other_array_name)

create_watershed_clusters

create_watershed_clusters(indices)

Group pixel indices into clusters by a distance threshold (custom).

Parameters:

Name	Type	Description	Default
indices	ndarray	(N, 3) pixel indices to cluster.	required

Returns:

Name	Type	Description
tuple		(cluster_map, clusters) — per-pixel cluster labels and
		the array of unique cluster ids.

Source code in lelantos/voidfinder.py

def create_watershed_clusters(self, indices):
    """Group pixel indices into clusters by a distance threshold (custom).

    Args:
        indices (numpy.ndarray): ``(N, 3)`` pixel indices to cluster.

    Returns:
        tuple: ``(cluster_map, clusters)`` — per-pixel cluster labels and
        the array of unique cluster ids.
    """
    cluster_map = np.zeros(indices[:, 0].shape, dtype=np.int64)
    mask_clusters = cluster_map == 0
    cluster_number = 0
    while len(cluster_map[mask_clusters]) != 0:
        indice_normalized = indices - indices[np.argwhere(mask_clusters)[0][0]]
        dist_index = np.sqrt(
            indice_normalized[:, 0] ** 2
            + indice_normalized[:, 1] ** 2
            + indice_normalized[:, 2] ** 2
        )
        mask_dist = dist_index <= self.params_void_finder["dist_clusters"]
        clust = np.unique(cluster_map[mask_dist])
        clust = clust[clust != 0]
        if len(clust) == 0:
            cluster_number += 1
            cluster_map[mask_dist] = cluster_number
        else:
            if len(clust) == 1:
                mask_clust = mask_dist & (cluster_map == 0)
                cluster_map[mask_clust] = clust[0]
            else:
                cluster_number += 1
                for c in clust:
                    mask_clust = cluster_map == c
                    cluster_map[mask_clust] = cluster_number
        mask_clusters = cluster_map == 0
    clusters = np.unique(cluster_map)
    return (cluster_map, clusters)

create_watershed_clusters2

create_watershed_clusters2(indices)

Cluster pixel indices with scikit-learn agglomerative clustering.

Parameters:

Name	Type	Description	Default
indices	ndarray	(N, 3) pixel indices to cluster.	required

Returns:

Name	Type	Description
tuple		(cluster_map, clusters).

Source code in lelantos/voidfinder.py

def create_watershed_clusters2(self, indices):
    """Cluster pixel indices with scikit-learn agglomerative clustering.

    Args:
        indices (numpy.ndarray): ``(N, 3)`` pixel indices to cluster.

    Returns:
        tuple: ``(cluster_map, clusters)``.
    """
    from sklearn.cluster import AgglomerativeClustering

    linkage = "simple"
    cluster_map = (
        AgglomerativeClustering(
            distance_threshold=1.5, n_clusters=None, linkage=linkage
        )
        .fit(indices)
        .labels_
    )
    clusters = np.unique(cluster_map)
    return (cluster_map, clusters)

create_watershed_clusters3

create_watershed_clusters3(indices)

Cluster pixel indices with scipy hierarchical fclusterdata.

Parameters:

Name	Type	Description	Default
indices	ndarray	(N, 3) pixel indices to cluster.	required

Returns:

Name	Type	Description
tuple		(cluster_map, clusters).

Source code in lelantos/voidfinder.py

def create_watershed_clusters3(self, indices):
    """Cluster pixel indices with scipy hierarchical ``fclusterdata``.

    Args:
        indices (numpy.ndarray): ``(N, 3)`` pixel indices to cluster.

    Returns:
        tuple: ``(cluster_map, clusters)``.
    """
    from scipy.cluster.hierarchy import fclusterdata

    cluster_map = fclusterdata(indices, t=1, criterion="distance")
    clusters = np.unique(cluster_map)
    return (cluster_map, clusters)

find_voids_sphere

find_voids_sphere(parameters)

Detect voids by growing spheres around under-threshold seed pixels.

For each seed pixel a sphere is grown while the enclosed mean stays below (voids) / above (clusters) average, up to maximal_radius. Zero-radius seeds are dropped and overlaps cleaned. Uses module-level globals so the per-seed work can be dispatched to a process pool.

Parameters:

Name	Type	Description	Default
parameters	tuple	(map_name, map_mpc_per_pixel, map_array).	required

Returns:

Name	Type	Description
tuple		(radius, coord, other_array, other_array_name) with
		other_array_name = ["MEAN", "VALUE"].

Source code in lelantos/voidfinder.py

def find_voids_sphere(self, parameters):
    """Detect voids by growing spheres around under-threshold seed pixels.

    For each seed pixel a sphere is grown while the enclosed mean stays
    below (voids) / above (clusters) ``average``, up to ``maximal_radius``.
    Zero-radius seeds are dropped and overlaps cleaned. Uses module-level
    globals so the per-seed work can be dispatched to a process pool.

    Args:
        parameters (tuple): ``(map_name, map_mpc_per_pixel, map_array)``.

    Returns:
        tuple: ``(radius, coord, other_array, other_array_name)`` with
        ``other_array_name = ["MEAN", "VALUE"]``.
    """
    global map_array_spherical_voidfinder
    map_name, map_mpc_per_pixel, map_array_spherical_voidfinder = parameters
    self.log.add(
        f"Beginning of the Simple spherical finding for the map {map_name}"
    )
    maximal_radius = np.around(
        self.params_void_finder["maximal_radius"] / map_mpc_per_pixel, decimals=0
    ).astype(int)
    global indice_spherical_voidfinder
    indice_spherical_voidfinder = np.transpose(
        np.indices(map_array_spherical_voidfinder.shape), axes=(1, 2, 3, 0)
    )

    if self.find_cluster:
        mask = map_array_spherical_voidfinder < self.params_void_finder["threshold"]
    else:
        mask = map_array_spherical_voidfinder > self.params_void_finder["threshold"]
    coord = np.argwhere(mask)
    del mask
    self.log.add(f"Number of pixels for the map {map_name} = {coord.shape[0]}")

    if self.number_core > 1:
        self.log.add(
            f"{self.number_core} processes used, start of multiprocessing routines"
        )
        self.log.add(f"Start of pool for the map {map_name}")
        func = partial(self.find_the_sphere, map_mpc_per_pixel, maximal_radius)
        with mp.Pool(self.number_core) as pool:
            pool_results = np.array(pool.map(func, coord))
        if pool_results.shape[0] != 0:
            radius, mean_value = pool_results[:, 0], pool_results[:, 1]
        else:
            radius, mean_value = np.array([]), np.array([])
        self.log.add(f"End of pool for the map {map_name}")
    else:
        self.log.add("Start of serial calculation")
        radius = np.zeros(coord.shape[0])
        mean_value = np.zeros(coord.shape[0])
        for i in range(len(coord)):
            radius[i], mean_value[i] = self.find_the_sphere(
                map_mpc_per_pixel, maximal_radius, coord[i]
            )
    del indice_spherical_voidfinder, maximal_radius
    mask = radius == 0
    coord = coord[~mask]
    radius = radius[~mask]
    mean_value = mean_value[~mask]
    coord_clean, radius_clean, other_array_clean = self.delete_voids(
        map_mpc_per_pixel, radius, coord, other_array=[mean_value]
    )
    nearest_coord = np.round(coord_clean, 0).astype(int)
    other_array_clean.append(
        map_array_spherical_voidfinder[
            nearest_coord[:, 0], nearest_coord[:, 1], nearest_coord[:, 2]
        ]
    )
    coord_clean = self.convert_to_Mpc(map_mpc_per_pixel, coord_clean)
    del (
        map_array_spherical_voidfinder,
        coord,
        mask,
        radius,
        nearest_coord,
        map_mpc_per_pixel,
    )
    other_array_name = ["MEAN", "VALUE"]
    self.log.add(f"End of the Simple spherical finding for the map {map_name}")
    if self.save_temporary_file:
        self.save_temporary_catalog(
            map_name, radius_clean, coord_clean, other_array_clean, other_array_name
        )
    return (radius_clean, coord_clean, other_array_clean, other_array_name)

find_the_sphere

find_the_sphere(mpc_per_pixel, maximal_radius, coord)

Grow the largest under-density sphere around a single seed pixel.

Parameters:

Name	Type	Description	Default
mpc_per_pixel	array - like	Pixel scale per axis (Mpc.h^-1).	required
maximal_radius	array - like	Search half-window in pixels per axis.	required
coord	array - like	Seed pixel index (i, j, k).	required

Returns:

Name	Type	Description
tuple		(radius, mean_value) in Mpc.h^-1 and enclosed mean delta;
		(0, 0) if no valid void is found.

Source code in lelantos/voidfinder.py

def find_the_sphere(self, mpc_per_pixel, maximal_radius, coord):
    """Grow the largest under-density sphere around a single seed pixel.

    Args:
        mpc_per_pixel (array-like): Pixel scale per axis (Mpc.h^-1).
        maximal_radius (array-like): Search half-window in pixels per axis.
        coord (array-like): Seed pixel index ``(i, j, k)``.

    Returns:
        tuple: ``(radius, mean_value)`` in Mpc.h^-1 and enclosed mean delta;
        ``(0, 0)`` if no valid void is found.
    """
    map_local = map_array_spherical_voidfinder[
        max(coord[0] - maximal_radius[0], 0) : min(
            map_array_spherical_voidfinder.shape[0], coord[0] + maximal_radius[0]
        ),
        max(coord[1] - maximal_radius[1], 0) : min(
            map_array_spherical_voidfinder.shape[1], coord[1] + maximal_radius[1]
        ),
        max(coord[2] - maximal_radius[2], 0) : min(
            map_array_spherical_voidfinder.shape[2], coord[2] + maximal_radius[2]
        ),
    ]
    distance_map = (
        indice_spherical_voidfinder[
            max(coord[0] - maximal_radius[0], 0) : min(
                map_array_spherical_voidfinder.shape[0],
                coord[0] + maximal_radius[0],
            ),
            max(coord[1] - maximal_radius[1], 0) : min(
                map_array_spherical_voidfinder.shape[1],
                coord[1] + maximal_radius[1],
            ),
            max(coord[2] - maximal_radius[2], 0) : min(
                map_array_spherical_voidfinder.shape[2],
                coord[2] + maximal_radius[2],
            ),
        ]
        - coord
    ) * mpc_per_pixel
    distance_map = np.sqrt(
        distance_map[:, :, :, 0] ** 2
        + distance_map[:, :, :, 1] ** 2
        + distance_map[:, :, :, 2] ** 2
    )
    radius = self.params_void_finder["minimal_radius"]
    if self.find_cluster:
        boolean = (
            np.mean(map_local[distance_map < radius])
            < self.params_void_finder["average"]
        )
    else:
        boolean = (
            np.mean(map_local[distance_map < radius])
            > self.params_void_finder["average"]
        )
    while boolean & (radius < self.params_void_finder["maximal_radius"]):
        radius = radius + self.params_void_finder["radius_step"]
        mean_value = np.mean(map_local[distance_map < radius])
        if self.find_cluster:
            boolean = mean_value < self.params_void_finder["average"]
        else:
            boolean = mean_value > self.params_void_finder["average"]
    del distance_map, map_local, boolean
    if (radius >= self.params_void_finder["maximal_radius"]) | (
        radius <= self.params_void_finder["minimal_radius"]
    ):
        radius, mean_value = 0, 0
    return (radius, mean_value)

delete_voids

delete_voids(mpc_per_pixel, radius, coord, other_array=None, mpc=False)

Remove overlapping voids using the configured delete_option.

Parameters:

Name	Type	Description	Default
mpc_per_pixel	array - like	Pixel scale per axis (Mpc.h^-1).	required
radius	ndarray	Void radii.	required
coord	ndarray	Void coordinates.	required
other_array	list[ndarray]	Extra per-void arrays.	None
mpc	bool	True if coord is already in Mpc.h^-1.	False

Returns:

Name	Type	Description
tuple		Cleaned (coord, radius[, other_array]) (the third element
		is omitted when other_array is None).

Raises:

Type	Description
ValueError	If delete_option is unknown.

Source code in lelantos/voidfinder.py

def delete_voids(self, mpc_per_pixel, radius, coord, other_array=None, mpc=False):
    """Remove overlapping voids using the configured ``delete_option``.

    Args:
        mpc_per_pixel (array-like): Pixel scale per axis (Mpc.h^-1).
        radius (numpy.ndarray): Void radii.
        coord (numpy.ndarray): Void coordinates.
        other_array (list[numpy.ndarray], optional): Extra per-void arrays.
        mpc (bool, optional): True if ``coord`` is already in Mpc.h^-1.

    Returns:
        tuple: Cleaned ``(coord, radius[, other_array])`` (the third element
        is omitted when ``other_array`` is None).

    Raises:
        ValueError: If ``delete_option`` is unknown.
    """
    if self.delete_option == "CLUSTERS":
        new_coord, new_radius, new_other_array = self.delete_overlapers_clusters(
            mpc_per_pixel, radius, coord, other_array=other_array, mpc=mpc
        )
    elif self.delete_option == "ITERATION":
        new_coord, new_radius, new_other_array = self.iterate_overlap_deletion(
            mpc_per_pixel, radius, coord, other_array=other_array, mpc=mpc
        )
    elif self.delete_option == "NONE":
        new_coord, new_radius, new_other_array = coord, radius, other_array
    else:
        raise ValueError(
            "The delete_option chosen is not implemented, try : CLUSTERS, ITERATION or NONE"
        )
    if other_array is not None:
        return (new_coord, new_radius, new_other_array)
    else:
        return (new_coord, new_radius)

iterate_overlap_deletion

iterate_overlap_deletion(mpc_per_pixel, radius, coord, other_array=None, mpc=False)

Repeatedly delete overlappers until no void overlaps remain.

Parameters:

Name	Type	Description	Default
mpc_per_pixel	array - like	Pixel scale per axis (Mpc.h^-1).	required
radius	ndarray	Void radii.	required
coord	ndarray	Void coordinates.	required
other_array	list[ndarray]	Extra per-void arrays.	None
mpc	bool	True if coord is already in Mpc.h^-1.	False

Returns:

Name	Type	Description
tuple		Cleaned (coord, radius, other_array).

Source code in lelantos/voidfinder.py

def iterate_overlap_deletion(
    self, mpc_per_pixel, radius, coord, other_array=None, mpc=False
):
    """Repeatedly delete overlappers until no void overlaps remain.

    Args:
        mpc_per_pixel (array-like): Pixel scale per axis (Mpc.h^-1).
        radius (numpy.ndarray): Void radii.
        coord (numpy.ndarray): Void coordinates.
        other_array (list[numpy.ndarray], optional): Extra per-void arrays.
        mpc (bool, optional): True if ``coord`` is already in Mpc.h^-1.

    Returns:
        tuple: Cleaned ``(coord, radius, other_array)``.
    """
    if other_array is not None:
        others_arrays_copies = []
        for i in range(len(other_array)):
            others_arrays_copies.append(other_array[i].copy())
    else:
        others_arrays_copies = None
    radius_copy = radius.copy()
    coord_copy = coord.copy()
    nb_voids_delete = len(coord)
    new_coord, new_radius, new_others_arrays = coord, radius, other_array
    while nb_voids_delete > 0:
        (
            new_coord,
            new_radius,
            nb_voids_delete,
            new_others_arrays,
        ) = self.delete_overlapers(
            mpc_per_pixel,
            radius_copy,
            coord_copy,
            other_array=others_arrays_copies,
            mpc=mpc,
        )
        coord_copy = new_coord
        radius_copy = new_radius
        others_arrays_copies = new_others_arrays
    if other_array is not None:
        return (new_coord, new_radius, new_others_arrays)
    else:
        return (new_coord, new_radius, None)

delete_overlapers

delete_overlapers(mpc_per_pixel, radius, coord, other_array=None, mpc=False)

One pass of overlap removal, keeping the largest of each overlap set.

Parameters:

Name	Type	Description	Default
mpc_per_pixel	array - like	Pixel scale per axis (Mpc.h^-1).	required
radius	ndarray	Void radii.	required
coord	ndarray	Void coordinates.	required
other_array	list[ndarray]	Extra per-void arrays.	None
mpc	bool	True if coord is already in Mpc.h^-1.	False

Returns:

Name	Type	Description
tuple		(coord, radius, nb_deleted, other_array) after one pass.

Source code in lelantos/voidfinder.py

def delete_overlapers(
    self, mpc_per_pixel, radius, coord, other_array=None, mpc=False
):
    """One pass of overlap removal, keeping the largest of each overlap set.

    Args:
        mpc_per_pixel (array-like): Pixel scale per axis (Mpc.h^-1).
        radius (numpy.ndarray): Void radii.
        coord (numpy.ndarray): Void coordinates.
        other_array (list[numpy.ndarray], optional): Extra per-void arrays.
        mpc (bool, optional): True if ``coord`` is already in Mpc.h^-1.

    Returns:
        tuple: ``(coord, radius, nb_deleted, other_array)`` after one pass.
    """
    if other_array is not None:
        other_array_clean = [[] for i in range(len(other_array))]
    else:
        other_array_clean = None
    coord_cleaned = []
    radius_cleaned = []
    radius_delete = radius.copy()
    mask = radius_delete > 0
    while len(radius_delete[mask]) != 0:
        coord_left = coord[mask]
        radius_left = radius_delete[mask]
        if other_array_clean is not None:
            other_array_left = []
            for i in range(len(other_array)):
                other_array_left.append(other_array[i][mask])
        else:
            other_array_left = None
        rad = radius_left[0]
        index = coord_left[0]
        sum_rad = radius_left + rad
        if mpc:
            indice_normalized = coord_left - index
        else:
            indice_normalized = (coord_left - index) * mpc_per_pixel
        dist_rad = np.sqrt(
            indice_normalized[:, 0] ** 2
            + indice_normalized[:, 1] ** 2
            + indice_normalized[:, 2] ** 2
        )
        mask2 = sum_rad >= dist_rad
        maxi = (
            np.argwhere(radius_left[mask2] == np.amax(radius_left[mask2]))
            .flatten()
            .tolist()
        )
        coord_cleaned.append(np.mean(coord_left[mask2][maxi], axis=0))
        radius_cleaned.append(np.mean(radius_left[mask2][maxi], axis=0))
        if other_array_clean is not None:
            for i in range(len(other_array)):
                other_array_clean[i].append(
                    np.mean(other_array_left[i][mask2][maxi])
                )
        radius_left[mask2] = 0
        radius_delete[mask] = radius_left
        mask = radius_delete > 0
    del (
        mask,
        radius_delete,
        radius_left,
        coord_left,
        indice_normalized,
        sum_rad,
        dist_rad,
    )
    if other_array_clean is not None:
        for i in range(len(other_array_clean)):
            other_array_clean[i] = np.array(other_array_clean[i])
    return (
        np.array(coord_cleaned),
        np.array(radius_cleaned),
        len(coord) - len(coord_cleaned),
        other_array_clean,
    )

find_overlapers

find_overlapers(mpc_per_pixel, index, rad, radius, coord, mpc=False)

Mask the voids that overlap a reference void.

Two voids overlap when their centre distance is below the sum of radii.

Parameters:

Name	Type	Description	Default
mpc_per_pixel	array - like	Pixel scale per axis (Mpc.h^-1).	required
index	array - like	Reference void coordinate.	required
rad	float	Reference void radius.	required
radius	ndarray	Candidate void radii.	required
coord	ndarray	Candidate void coordinates.	required
mpc	bool	True if coordinates are already in Mpc.h^-1.	False

Returns:

Type	Description
	numpy.ndarray: Boolean mask of overlapping voids.

Source code in lelantos/voidfinder.py

def find_overlapers(self, mpc_per_pixel, index, rad, radius, coord, mpc=False):
    """Mask the voids that overlap a reference void.

    Two voids overlap when their centre distance is below the sum of radii.

    Args:
        mpc_per_pixel (array-like): Pixel scale per axis (Mpc.h^-1).
        index (array-like): Reference void coordinate.
        rad (float): Reference void radius.
        radius (numpy.ndarray): Candidate void radii.
        coord (numpy.ndarray): Candidate void coordinates.
        mpc (bool, optional): True if coordinates are already in Mpc.h^-1.

    Returns:
        numpy.ndarray: Boolean mask of overlapping voids.
    """
    sum_rad = radius + rad
    if mpc:
        indice_normalized = coord - index
    else:
        indice_normalized = (coord - index) * mpc_per_pixel
    dist_rad = np.sqrt(
        indice_normalized[:, 0] ** 2
        + indice_normalized[:, 1] ** 2
        + indice_normalized[:, 2] ** 2
    )
    overlapers_mask = sum_rad >= dist_rad
    del sum_rad, dist_rad, indice_normalized
    return overlapers_mask

create_overlaper_map

create_overlaper_map(mpc_per_pixel, radius, coord, mpc=False)

Group overlapping voids into clusters.

Parameters:

Name	Type	Description	Default
mpc_per_pixel	array - like	Pixel scale per axis (Mpc.h^-1).	required
radius	ndarray	Void radii.	required
coord	ndarray	Void coordinates.	required
mpc	bool	True if coordinates are already in Mpc.h^-1.	False

Returns:

Name	Type	Description
tuple		(cluster_map, clusters) — per-void cluster labels and the
		unique cluster ids.

Source code in lelantos/voidfinder.py

def create_overlaper_map(self, mpc_per_pixel, radius, coord, mpc=False):
    """Group overlapping voids into clusters.

    Args:
        mpc_per_pixel (array-like): Pixel scale per axis (Mpc.h^-1).
        radius (numpy.ndarray): Void radii.
        coord (numpy.ndarray): Void coordinates.
        mpc (bool, optional): True if coordinates are already in Mpc.h^-1.

    Returns:
        tuple: ``(cluster_map, clusters)`` — per-void cluster labels and the
        unique cluster ids.
    """
    cluster_map = np.zeros(radius.shape, dtype=np.int64)
    mask_clusters = cluster_map == 0
    cluster_number = 0
    while len(cluster_map[mask_clusters]) != 0:
        arg = np.argwhere(mask_clusters)[0][0]
        rad = radius[arg]
        index = coord[arg]
        overlapers_mask = self.find_overlapers(
            mpc_per_pixel, index, rad, radius, coord, mpc=mpc
        )
        clust = np.array(list(set(cluster_map[overlapers_mask])))
        clust = clust[clust != 0]
        if len(clust) == 0:
            cluster_number += 1
            cluster_map[overlapers_mask] = cluster_number
        else:
            if len(clust) == 1:
                mask_clust = overlapers_mask & (cluster_map == 0)
                cluster_map[mask_clust] = clust[0]
            else:
                cluster_number += 1
                for c in clust:
                    mask_clust = cluster_map == c
                    cluster_map[mask_clust] = cluster_number
        mask_clusters = cluster_map == 0
    clusters = np.unique(cluster_map)
    return (cluster_map, clusters)

delete_overlapers_clusters

delete_overlapers_clusters(mpc_per_pixel, radius, coord, other_array=None, mpc=False)

Reduce each overlap cluster to its largest void (CLUSTERS option).

Parameters:

Name	Type	Description	Default
mpc_per_pixel	array - like	Pixel scale per axis (Mpc.h^-1).	required
radius	ndarray	Void radii.	required
coord	ndarray	Void coordinates.	required
other_array	list[ndarray]	Extra per-void arrays.	None
mpc	bool	True if coordinates are already in Mpc.h^-1.	False

Returns:

Name	Type	Description
tuple		Cleaned (coord, radius, other_array).

Source code in lelantos/voidfinder.py

def delete_overlapers_clusters(
    self, mpc_per_pixel, radius, coord, other_array=None, mpc=False
):
    """Reduce each overlap cluster to its largest void (CLUSTERS option).

    Args:
        mpc_per_pixel (array-like): Pixel scale per axis (Mpc.h^-1).
        radius (numpy.ndarray): Void radii.
        coord (numpy.ndarray): Void coordinates.
        other_array (list[numpy.ndarray], optional): Extra per-void arrays.
        mpc (bool, optional): True if coordinates are already in Mpc.h^-1.

    Returns:
        tuple: Cleaned ``(coord, radius, other_array)``.
    """
    if other_array is not None:
        other_array_clean = [[] for i in range(len(other_array))]
    else:
        other_array_clean = None
    coord_cleaned = []
    radius_cleaned = []
    (cluster_map, clusters) = self.create_overlaper_map(
        mpc_per_pixel, radius, coord, mpc=mpc
    )
    for c in clusters:
        mask = cluster_map == c
        radius_cluster = radius[mask]
        coord_cluster = coord[mask]
        if other_array is not None:
            other_array_cluster = []
            for i in range(len(other_array)):
                other_array_cluster.append(other_array[i][mask])
        maxi = (
            np.argwhere(radius_cluster == np.amax(radius_cluster))
            .flatten()
            .tolist()
        )
        coord_cleaned.append(np.mean(coord_cluster[maxi], axis=0))
        radius_cleaned.append(np.mean(radius_cluster[maxi], axis=0))
        if other_array is not None:
            for i in range(len(other_array)):
                other_array_clean[i].append(np.mean(other_array_cluster[i][maxi]))
    if other_array is not None:
        return (
            np.array(coord_cleaned),
            np.array(radius_cleaned),
            other_array_clean,
        )
    else:
        return (np.array(coord_cleaned), np.array(radius_cleaned), None)

convert_to_Mpc

convert_to_Mpc(mpc_per_pixel, coord)

Convert pixel coordinates to Mpc.h^-1.

Parameters:

Name	Type	Description	Default
mpc_per_pixel	array - like	Pixel scale per axis (Mpc.h^-1).	required
coord	ndarray	Coordinates in pixels.	required

Returns:

Type	Description
	numpy.ndarray: Coordinates in Mpc.h^-1.

Source code in lelantos/voidfinder.py

def convert_to_Mpc(self, mpc_per_pixel, coord):
    """Convert pixel coordinates to Mpc.h^-1.

    Args:
        mpc_per_pixel (array-like): Pixel scale per axis (Mpc.h^-1).
        coord (numpy.ndarray): Coordinates in pixels.

    Returns:
        numpy.ndarray: Coordinates in Mpc.h^-1.
    """
    if coord.shape[0] != 0:
        coord = coord * np.array(mpc_per_pixel)
    return coord

save_temporary_catalog

save_temporary_catalog(map_name, radius, coord, other_array, other_array_name)

Write a per-chunk temporary void catalog (used for restart).

Parameters:

Name	Type	Description	Default
map_name	str	Chunk map name (used to build the file name).	required
radius	ndarray	Void radii.	required
coord	ndarray	Void coordinates.	required
other_array	list[ndarray]	Extra per-void arrays.	required
other_array_name	list[str]	Names of the extra arrays.	required

Source code in lelantos/voidfinder.py

def save_temporary_catalog(
    self, map_name, radius, coord, other_array, other_array_name
):
    """Write a per-chunk temporary void catalog (used for restart).

    Args:
        map_name (str): Chunk map name (used to build the file name).
        radius (numpy.ndarray): Void radii.
        coord (numpy.ndarray): Void coordinates.
        other_array (list[numpy.ndarray]): Extra per-void arrays.
        other_array_name (list[str]): Names of the extra arrays.
    """
    dict_void = {"R": radius, "COORD": coord}
    for i in range(len(other_array)):
        dict_void[other_array_name[i]] = other_array[i]
    name = self.temporary_file_name.format(map_name)
    void = tomographic_objects.VoidCatalog.init_from_dictionary(
        name,
        radius,
        coord,
        "cartesian",
        self.map_coordinate_transform,
        self.map_Omega_m,
        self.map_boundary_cartesian_coord,
        self.map_boundary_sky_coord,
        other_array=other_array,
        other_array_name=other_array_name,
    )
    void.write()

delete_temporary_files

delete_temporary_files(map_chunks, list_index_map_chunks)

Delete the per-chunk temporary catalogs after a successful merge.

Parameters:

Name	Type	Description	Default
map_chunks	dict	Per-chunk results.	required
list_index_map_chunks	list[str]	Chunk keys.	required

Source code in lelantos/voidfinder.py

def delete_temporary_files(self, map_chunks, list_index_map_chunks):
    """Delete the per-chunk temporary catalogs after a successful merge.

    Args:
        map_chunks (dict): Per-chunk results.
        list_index_map_chunks (list[str]): Chunk keys.
    """
    for i in range(len(list_index_map_chunks)):
        map_name = map_chunks[list_index_map_chunks[i]]["map_name"]
        file_name = self.temporary_file_name.format(map_name)
        if os.path.isfile(file_name):
            os.remove(file_name)

restart_calculation

restart_calculation(map_chunks, list_index_map_chunks, other_array_name)

Load already-computed chunks from temporary catalogs (restart).

Parameters:

Name	Type	Description	Default
map_chunks	dict	Per-chunk results (filled in place for found ones).	required
list_index_map_chunks	list[str]	All chunk keys.	required
other_array_name	list[str]	Extra arrays to read from the temps.	required

Returns:

Name	Type	Description
tuple		(map_chunks, remaining_chunk_keys) — the chunks still to
		be computed.

Source code in lelantos/voidfinder.py

def restart_calculation(self, map_chunks, list_index_map_chunks, other_array_name):
    """Load already-computed chunks from temporary catalogs (restart).

    Args:
        map_chunks (dict): Per-chunk results (filled in place for found ones).
        list_index_map_chunks (list[str]): All chunk keys.
        other_array_name (list[str]): Extra arrays to read from the temps.

    Returns:
        tuple: ``(map_chunks, remaining_chunk_keys)`` — the chunks still to
        be computed.
    """
    list_index_map_chunks_restart = []
    for i in range(len(list_index_map_chunks)):
        map_name = map_chunks[list_index_map_chunks[i]]["map_name"]
        tmp_file_name = self.temporary_file_name.format(map_name)
        if os.path.isfile(tmp_file_name):
            self.log.add(
                f"Temporary file {tmp_file_name} found and added to the calculation"
            )
            tmp_catalog = tomographic_objects.VoidCatalog.init_from_fits(
                tmp_file_name
            )
            map_chunks[list_index_map_chunks[i]]["radius"] = tmp_catalog.radius
            map_chunks[list_index_map_chunks[i]]["coord"] = tmp_catalog.coord
            map_chunks[list_index_map_chunks[i]][
                "other_array"
            ] = tmp_catalog.return_array_list(other_array_name)
            map_chunks[list_index_map_chunks[i]][
                "other_array_name"
            ] = other_array_name
        else:
            list_index_map_chunks_restart.append(list_index_map_chunks[i])
    return (map_chunks, list_index_map_chunks_restart)

save_voids

save_voids(radius, coord, other_array, other_array_name, coordinate_transform, Omega_m, boundary_cartesian_coord, boundary_sky_coord)

Build and write the final void catalog from finder outputs.

Parameters:

Name	Type	Description	Default
radius	ndarray	Void radii.	required
coord	ndarray	Void coordinates (cartesian, Mpc.h^-1).	required
other_array	list[ndarray]	Extra per-void arrays.	required
other_array_name	list[str]	Names of the extra arrays.	required
coordinate_transform	str	Map sky<->cartesian transform mode.	required
Omega_m	float	Fiducial matter density used for the map.	required
boundary_cartesian_coord		Cartesian bounding box of the map.	required
boundary_sky_coord		Sky bounding box of the map.	required

Returns:

Name	Type	Description
str		Path of the written catalog.

Source code in lelantos/voidfinder.py

def save_voids(
    self,
    radius,
    coord,
    other_array,
    other_array_name,
    coordinate_transform,
    Omega_m,
    boundary_cartesian_coord,
    boundary_sky_coord,
):
    """Build and write the final void catalog from finder outputs.

    Args:
        radius (numpy.ndarray): Void radii.
        coord (numpy.ndarray): Void coordinates (cartesian, Mpc.h^-1).
        other_array (list[numpy.ndarray]): Extra per-void arrays.
        other_array_name (list[str]): Names of the extra arrays.
        coordinate_transform (str): Map sky<->cartesian transform mode.
        Omega_m (float): Fiducial matter density used for the map.
        boundary_cartesian_coord: Cartesian bounding box of the map.
        boundary_sky_coord: Sky bounding box of the map.

    Returns:
        str: Path of the written catalog.
    """
    dict_void = {"R": radius, "COORD": coord}
    for i in range(len(other_array)):
        dict_void[other_array_name[i]] = other_array[i]
    name = os.path.join(self.pwd, f"Catalog_{self.get_name_catalog()}.fits")
    void = tomographic_objects.VoidCatalog.init_from_dictionary(
        name,
        radius,
        coord,
        "cartesian",
        coordinate_transform,
        Omega_m,
        boundary_cartesian_coord,
        boundary_sky_coord,
        other_array=other_array,
        other_array_name=other_array_name,
    )
    void.write()
    return name

get_name_catalog

get_name_catalog()

Build the catalog base name encoding the finder parameters.

Returns:

Name	Type	Description
str		A name embedding the method, threshold, average/dist_clusters,
		minimal radius and delete option.

Raises:

Type	Description
ValueError	If the finder method is neither SPHERICAL nor WATERSHED.

Source code in lelantos/voidfinder.py

def get_name_catalog(self):
    """Build the catalog base name encoding the finder parameters.

    Returns:
        str: A name embedding the method, threshold, average/dist_clusters,
        minimal radius and delete option.

    Raises:
        ValueError: If the finder method is neither SPHERICAL nor WATERSHED.
    """
    if self.find_cluster:
        name_out = "Clusters"
    else:
        name_out = "Voids"
    if self.params_void_finder["method"] == "SPHERICAL":
        name = f"""{name_out}_{self.params_void_finder["method"]}_{self.params_void_finder["threshold"]}threshold_{self.params_void_finder["average"]}average_{self.params_void_finder["minimal_radius"]}rmin_{self.delete_option}_deletion"""
    elif self.params_void_finder["method"] == "WATERSHED":
        name = f"""{name_out}_{self.params_void_finder["method"]}_{self.params_void_finder["threshold"]}threshold_{self.params_void_finder["dist_clusters"]}dist_clusters_{self.params_void_finder["minimal_radius"]}rmin_{self.delete_option}_deletion"""
    else:
        raise ValueError(
            "The method_void chosen is not implemented, try : WATERSHED or SPHERICAL"
        )
    return name

PlotVoid

PlotVoid(pwd, void_catalog)

Bases: object

Diagnostic plots (histograms, redshift trends) for a void catalog.

Load the void catalog to plot.

Parameters:

Name	Type	Description	Default
pwd	str	Output directory for the figures.	required
void_catalog	str	Void catalog FITS file.	required

Source code in lelantos/voidfinder.py

def __init__(self, pwd, void_catalog):
    """Load the void catalog to plot.

    Args:
        pwd (str): Output directory for the figures.
        void_catalog (str): Void catalog FITS file.
    """
    self.pwd = pwd
    self.void = tomographic_objects.Catalog.init_catalog_from_fits(
        void_catalog, "void"
    )

load_catalog

load_catalog(comparison, value_name)

Read a quantity from the main catalog and any comparison catalogs.

Parameters:

Name	Type	Description	Default
comparison	list[str] | None	Comparison catalog files.	required
value_name	str	Attribute to read (e.g. radius).	required

Returns:

Name	Type	Description
tuple		(value, comparison_value, comparison_redshift).

Source code in lelantos/voidfinder.py

def load_catalog(self, comparison, value_name):
    """Read a quantity from the main catalog and any comparison catalogs.

    Args:
        comparison (list[str] | None): Comparison catalog files.
        value_name (str): Attribute to read (e.g. ``radius``).

    Returns:
        tuple: ``(value, comparison_value, comparison_redshift)``.
    """
    comparison_redshift, comparison_value = None, None
    if comparison is not None:
        comparison_value = []
        comparison_redshift = []
        for i in range(len(comparison)):
            catalog = tomographic_objects.Catalog.init_catalog_from_fits(
                comparison[i], "void"
            )
            comparison_value.append(getattr(catalog, value_name))
            comparison_redshift.append(catalog.redshift)
    value = getattr(self.void, value_name)
    return (value, comparison_value, comparison_redshift)

plot_histo

plot_histo(value_name, name, comparison=None, comparison_legend=None, loaded_value=None, **kwargs)

Plot the histogram of a void quantity (with optional comparisons).

Parameters:

Name	Type	Description	Default
value_name	str	Quantity to histogram (e.g. radius).	required
name	str	Base output name.	required
comparison	list[str] | list[array]	Comparison data.	None
comparison_legend	list[str]	Legend labels.	None
loaded_value	array - like	Pre-loaded main values.	None
**kwargs		Styling options.	{}

Source code in lelantos/voidfinder.py

def plot_histo(
    self,
    value_name,
    name,
    comparison=None,
    comparison_legend=None,
    loaded_value=None,
    **kwargs,
):
    """Plot the histogram of a void quantity (with optional comparisons).

    Args:
        value_name (str): Quantity to histogram (e.g. ``radius``).
        name (str): Base output name.
        comparison (list[str] | list[array], optional): Comparison data.
        comparison_legend (list[str], optional): Legend labels.
        loaded_value (array-like, optional): Pre-loaded main values.
        **kwargs: Styling options.
    """
    if loaded_value is None:
        (value, comparison_value, comparison_redshift) = self.load_catalog(
            comparison, value_name
        )
    else:
        value, comparison_value = loaded_value, comparison
    utils.save_histo(
        self.pwd,
        value,
        value_name,
        name,
        comparison=comparison_value,
        comparison_legend=comparison_legend,
        **kwargs,
    )

plot_mean_redshift_dependence

plot_mean_redshift_dependence(value_name, name, comparison=None, comparison_redshift=None, comparison_legend=None, loaded_value=None, **kwargs)

Plot the mean of a void quantity versus redshift.

Parameters:

Name	Type	Description	Default
value_name	str	Quantity to average (e.g. radius).	required
name	str	Base output name.	required
comparison	list[str] | list[array]	Comparison data.	None
comparison_redshift	list[array]	Comparison redshifts.	None
comparison_legend	list[str]	Legend labels.	None
loaded_value	array - like	Pre-loaded main values.	None
**kwargs		Binning/styling options.	{}

Source code in lelantos/voidfinder.py

def plot_mean_redshift_dependence(
    self,
    value_name,
    name,
    comparison=None,
    comparison_redshift=None,
    comparison_legend=None,
    loaded_value=None,
    **kwargs,
):
    """Plot the mean of a void quantity versus redshift.

    Args:
        value_name (str): Quantity to average (e.g. ``radius``).
        name (str): Base output name.
        comparison (list[str] | list[array], optional): Comparison data.
        comparison_redshift (list[array], optional): Comparison redshifts.
        comparison_legend (list[str], optional): Legend labels.
        loaded_value (array-like, optional): Pre-loaded main values.
        **kwargs: Binning/styling options.
    """
    if loaded_value is None:
        (value, comparison_value, comparison_redshift) = self.load_catalog(
            comparison, value_name
        )
    else:
        value, comparison_value = loaded_value, comparison
    redshift = self.void.redshift
    utils.save_mean_redshift_dependence(
        self.pwd,
        value,
        redshift,
        value_name,
        name,
        comparison=comparison_value,
        comparison_redshift=comparison_redshift,
        comparison_legend=None,
        **kwargs,
    )

plot_redshift_dependence

plot_redshift_dependence(value_name, name, comparison=None, comparison_redshift=None, comparison_legend=None, loaded_value=None, **kwargs)

Scatter a void quantity versus redshift.

Parameters:

Name	Type	Description	Default
value_name	str	Quantity to plot (e.g. radius).	required
name	str	Base output name.	required
comparison	list[str] | list[array]	Comparison data.	None
comparison_redshift	list[array]	Comparison redshifts.	None
comparison_legend	list[str]	Legend labels.	None
loaded_value	array - like	Pre-loaded main values.	None
**kwargs		Styling options.	{}

Source code in lelantos/voidfinder.py

def plot_redshift_dependence(
    self,
    value_name,
    name,
    comparison=None,
    comparison_redshift=None,
    comparison_legend=None,
    loaded_value=None,
    **kwargs,
):
    """Scatter a void quantity versus redshift.

    Args:
        value_name (str): Quantity to plot (e.g. ``radius``).
        name (str): Base output name.
        comparison (list[str] | list[array], optional): Comparison data.
        comparison_redshift (list[array], optional): Comparison redshifts.
        comparison_legend (list[str], optional): Legend labels.
        loaded_value (array-like, optional): Pre-loaded main values.
        **kwargs: Styling options.
    """
    if loaded_value is None:
        (value, comparison_value, comparison_redshift) = self.load_catalog(
            comparison, value_name
        )
    else:
        value, comparison_value = loaded_value, comparison
    redshift = self.void.redshift
    utils.save_redshift_dependence(
        self.pwd,
        value,
        redshift,
        value_name,
        name,
        comparison=comparison_value,
        comparison_redshift=comparison_redshift,
        comparison_legend=None,
        **kwargs,
    )

plot

plot(value_names, name, comparison=None, comparison_legend=None, histo=True, mean_z_dependence=True, z_dependence=True, **kwargs)

Produce all requested diagnostic plots for several void quantities.

Parameters:

Name	Type	Description	Default
value_names	list[str]	Quantities to plot.	required
name	str	Base output name.	required
comparison	list[str]	Comparison catalog files.	None
comparison_legend	list[str]	Legend labels.	None
histo	bool	Draw histograms.	True
mean_z_dependence	bool	Draw mean-vs-redshift plots.	True
z_dependence	bool	Draw value-vs-redshift plots.	True
**kwargs		Styling options (style selects a matplotlib style).	{}

Source code in lelantos/voidfinder.py

def plot(
    self,
    value_names,
    name,
    comparison=None,
    comparison_legend=None,
    histo=True,
    mean_z_dependence=True,
    z_dependence=True,
    **kwargs,
):
    """Produce all requested diagnostic plots for several void quantities.

    Args:
        value_names (list[str]): Quantities to plot.
        name (str): Base output name.
        comparison (list[str], optional): Comparison catalog files.
        comparison_legend (list[str], optional): Legend labels.
        histo (bool, optional): Draw histograms.
        mean_z_dependence (bool, optional): Draw mean-vs-redshift plots.
        z_dependence (bool, optional): Draw value-vs-redshift plots.
        **kwargs: Styling options (``style`` selects a matplotlib style).
    """
    style = utils.return_key(kwargs, "style", None)
    if style is not None:
        plt.style.use(style)

    for value_name in value_names:
        (value, comparison_value, comparison_redshift) = self.load_catalog(
            comparison, value_name
        )
        if histo:
            self.plot_histo(
                value_name,
                name,
                comparison=comparison_value,
                comparison_legend=comparison_legend,
                loaded_value=value,
                **kwargs,
            )
        if (mean_z_dependence) & (value_name != "redshift"):
            self.plot_mean_redshift_dependence(
                value_name,
                name,
                comparison=comparison_value,
                comparison_redshift=comparison_redshift,
                comparison_legend=comparison_legend,
                loaded_value=value,
                **kwargs,
            )
        if (z_dependence) & (value_name != "redshift"):
            self.plot_redshift_dependence(
                value_name,
                name,
                comparison=comparison_value,
                comparison_redshift=comparison_redshift,
                comparison_legend=comparison_legend,
                loaded_value=value,
                **kwargs,
            )

compute_ks_stat

compute_ks_stat(comparison)

Two-sample KS test between this catalog's radii and a comparison.

Parameters:

Name	Type	Description	Default
comparison	list[str]	Comparison catalog file(s).	required

Returns:

Name	Type	Description
tuple		(KS_stat, p_value).

Source code in lelantos/voidfinder.py

def compute_ks_stat(self, comparison):
    """Two-sample KS test between this catalog's radii and a comparison.

    Args:
        comparison (list[str]): Comparison catalog file(s).

    Returns:
        tuple: ``(KS_stat, p_value)``.
    """
    (radius, comparison, comparison_redshift) = self.load_catalog(
        comparison, "radius"
    )
    KS_stat, p_value = ks_2samp(radius, comparison[0])
    return (KS_stat, p_value)

plot_radius_histo_fit_expo

plot_radius_histo_fit_expo(expo_fit_rmin=0, comparison=None, **kwargs)

Plot the void-radius histogram with an exponential tail fit.

Parameters:

Name	Type	Description	Default
expo_fit_rmin	float	Minimum radius included in the fit.	0
comparison	list[str]	Comparison catalog files.	None
**kwargs		Histogram styling options.	{}

Returns:

Name	Type	Description
tuple		(perr, perr_comparison) — fit-parameter uncertainties for
		the main catalog and each comparison.

Source code in lelantos/voidfinder.py

def plot_radius_histo_fit_expo(self, expo_fit_rmin=0, comparison=None, **kwargs):
    """Plot the void-radius histogram with an exponential tail fit.

    Args:
        expo_fit_rmin (float, optional): Minimum radius included in the fit.
        comparison (list[str], optional): Comparison catalog files.
        **kwargs: Histogram styling options.

    Returns:
        tuple: ``(perr, perr_comparison)`` — fit-parameter uncertainties for
        the main catalog and each comparison.
    """
    fit_function = lambda x, a, b: np.exp(a * x + b)

    (radius, comparison, comparison_redshift) = self.load_catalog(
        comparison, "radius"
    )
    (name, n, bins, patches) = utils.plot_histo(radius, "radius", "", **kwargs)
    bin_center = (bins[1:] + bins[0:-1]) / 2
    mask = bin_center > expo_fit_rmin
    n, bins_fit = n[mask], bin_center[mask]
    fit = curve_fit(fit_function, bins_fit, n)
    plt.plot(bins_fit, fit_function(bins_fit, *fit[0]), "r")
    perr = np.sqrt(np.diag(fit[1]))

    perr_comparison = []
    if comparison is not None:
        for i in range(len(comparison)):
            (name, n, bins, patches) = utils.plot_histo(
                comparison[i], "radius", "", **kwargs
            )
            bin_center = (bins[1:] + bins[0:-1]) / 2
            mask = bin_center > expo_fit_rmin
            n, bins_fit = n[mask], bin_center[mask]
            fit = curve_fit(fit_function, bins_fit, n)
            plt.plot(bins_fit, fit_function(bins_fit, *fit[0]), "b")
            perr_comparison.append(np.sqrt(np.diag(fit[1])))
    return (perr, perr_comparison)

create_merged_catalog

create_merged_catalog(pwd, list_catalog_name, merged_catalog_name)

Merge several void catalogs into one and write it out.

Parameters:

Name	Type	Description	Default
pwd	str	Output directory.	required
list_catalog_name	list[str]	Void catalog files to merge.	required
merged_catalog_name	str	Output catalog file name.	required

Source code in lelantos/voidfinder.py

def create_merged_catalog(pwd, list_catalog_name, merged_catalog_name):
    """Merge several void catalogs into one and write it out.

    Args:
        pwd (str): Output directory.
        list_catalog_name (list[str]): Void catalog files to merge.
        merged_catalog_name (str): Output catalog file name.
    """
    void_merged = tomographic_objects.VoidCatalog.init_by_merging(
        list_catalog_name, name=os.path.join(pwd, merged_catalog_name)
    )
    void_merged.write()

compute_additional_stats

compute_additional_stats(catalog_name, pixel_name)

Add filling factor, QSO-crossing and LOS-distance stats to a catalog.

Loads the void catalog, computes the extra per-void quantities from the pixel (line-of-sight) file and writes the catalog back.

Parameters:

Name	Type	Description	Default
catalog_name	str	Void catalog FITS file.	required
pixel_name	str	Pixel (line-of-sight) file.	required

Source code in lelantos/voidfinder.py

def compute_additional_stats(catalog_name, pixel_name):
    """Add filling factor, QSO-crossing and LOS-distance stats to a catalog.

    Loads the void catalog, computes the extra per-void quantities from the
    pixel (line-of-sight) file and writes the catalog back.

    Args:
        catalog_name (str): Void catalog FITS file.
        pixel_name (str): Pixel (line-of-sight) file.
    """
    void = tomographic_objects.VoidCatalog.init_from_fits(catalog_name)
    void.compute_filling_factor()
    void.compute_crossing_criteria(pixel_name)
    void.compute_los_distance(pixel_name)
    void.write()

cut_catalog

cut_catalog(pwd, catalog_name, method_cut, coord_min=None, coord_max=None, cut_crossing_param=None, cut_radius=None, distance_map_name=None, distance_map_prop=None, distance_map_param=None, distance_map_percent=None)

Apply cuts to a void catalog and write the cut catalog.

Parameters:

Name	Type	Description	Default
pwd	str	Output directory.	required
catalog_name	str	Input void catalog FITS file.	required
method_cut	tuple[str]	Cuts to apply, any of CROSSING, RADIUS, BORDER, DIST.	required
coord_min, coord_max	sequence	Cartesian bounds (BORDER cut).	required
cut_crossing_param	float	Threshold for the CROSSING cut.	None
cut_radius	sequence	(rmin, rmax) for the RADIUS cut.	None
distance_map_name	str	Distance-map file (DIST cut).	None
distance_map_prop	str	Distance-map property file.	None
distance_map_param	float	Distance-map radius (DIST cut).	None
distance_map_percent	float	Required well-sampled fraction.	None

Returns:

Name	Type	Description
str		Path of the written cut catalog.

Source code in lelantos/voidfinder.py

def cut_catalog(
    pwd,
    catalog_name,
    method_cut,
    coord_min=None,
    coord_max=None,
    cut_crossing_param=None,
    cut_radius=None,
    distance_map_name=None,
    distance_map_prop=None,
    distance_map_param=None,
    distance_map_percent=None,
):
    """Apply cuts to a void catalog and write the cut catalog.

    Args:
        pwd (str): Output directory.
        catalog_name (str): Input void catalog FITS file.
        method_cut (tuple[str]): Cuts to apply, any of ``CROSSING``, ``RADIUS``,
            ``BORDER``, ``DIST``.
        coord_min, coord_max (sequence, optional): Cartesian bounds (BORDER cut).
        cut_crossing_param (float, optional): Threshold for the CROSSING cut.
        cut_radius (sequence, optional): ``(rmin, rmax)`` for the RADIUS cut.
        distance_map_name (str, optional): Distance-map file (DIST cut).
        distance_map_prop (str, optional): Distance-map property file.
        distance_map_param (float, optional): Distance-map radius (DIST cut).
        distance_map_percent (float, optional): Required well-sampled fraction.

    Returns:
        str: Path of the written cut catalog.
    """
    void_cut = tomographic_objects.VoidCatalog.init_from_fits(catalog_name)
    cut_catalog_name = void_cut.cut_catalog_void(
        method_cut,
        coord_min=coord_min,
        coord_max=coord_max,
        cut_crossing_param=cut_crossing_param,
        cut_radius=cut_radius,
        distance_map_name=distance_map_name,
        distance_map_prop=distance_map_prop,
        distance_map_param=distance_map_param,
        distance_map_percent=distance_map_percent,
    )
    void_cut.name = os.path.join(pwd, cut_catalog_name)
    void_cut.write()
    return void_cut.name

correct_void_coordinates

correct_void_coordinates(pwd, catalog_name, method, name_out, inv_g_function, pixel_name, **kwargs)

Apply a coordinate correction to a void catalog and write it out.

Parameters:

Name	Type	Description	Default
pwd	str	Output directory.	required
catalog_name	str	Input void catalog FITS file.	required
method	str	Correction method name.	required
name_out	str	Output catalog file name.	required
inv_g_function		Inverse growth/mapping function used by the correction.	required
pixel_name	str	Pixel (line-of-sight) file.	required
**kwargs		Extra correction options.	{}

Source code in lelantos/voidfinder.py

def correct_void_coordinates(
    pwd, catalog_name, method, name_out, inv_g_function, pixel_name, **kwargs
):
    """Apply a coordinate correction to a void catalog and write it out.

    Args:
        pwd (str): Output directory.
        catalog_name (str): Input void catalog FITS file.
        method (str): Correction method name.
        name_out (str): Output catalog file name.
        inv_g_function: Inverse growth/mapping function used by the correction.
        pixel_name (str): Pixel (line-of-sight) file.
        **kwargs: Extra correction options.
    """
    void = tomographic_objects.VoidCatalog.init_from_fits(catalog_name)
    void.correct_coordinates(method, name_out, inv_g_function, pixel_name, **kwargs)
    void.name = os.path.join(pwd, name_out)
    void.write()

get_crossing_qso

get_crossing_qso(catalog_name, qso_name)

Return the QSOs whose lines of sight cross the catalog voids.

Parameters:

Name	Type	Description	Default
catalog_name	str	Void catalog FITS file.	required
qso_name	str	QSO catalog file.	required

Returns:

Type	Description
	The crossing-QSO catalog object.

Source code in lelantos/voidfinder.py

def get_crossing_qso(catalog_name, qso_name):
    """Return the QSOs whose lines of sight cross the catalog voids.

    Args:
        catalog_name (str): Void catalog FITS file.
        qso_name (str): QSO catalog file.

    Returns:
        The crossing-QSO catalog object.
    """
    void = tomographic_objects.VoidCatalog.init_from_fits(catalog_name)
    qso = void.get_crossing_qso(qso_name)
    return qso

create_qso_like_catalog

create_qso_like_catalog(catalog_name)

Write a QSO-like (RA, Dec, z) void catalog for cross-correlation tools.

Parameters:

Name	Type	Description	Default
catalog_name	str	Input void catalog FITS file. The output is written next to it with a _qso_like suffix.	required

Source code in lelantos/voidfinder.py

def create_qso_like_catalog(catalog_name):
    """Write a QSO-like (RA, Dec, z) void catalog for cross-correlation tools.

    Args:
        catalog_name (str): Input void catalog FITS file. The output is written
            next to it with a ``_qso_like`` suffix.
    """
    void = tomographic_objects.VoidCatalog.init_from_fits(catalog_name)
    void.name = f"""{void.name.split(".fits")[0]}_qso_like.fits"""
    void.convert_to_cross_corr_radec()
    void.write(qso_like=True)

qso_to_3d

qso_to_3d(catalog_name, new_name, moveaxis=None)

Export a QSO catalog to a text file for 3D visualisation.

Parameters:

Name	Type	Description	Default
catalog_name	str	Input QSO catalog FITS file.	required
new_name	str	Output text file.	required
moveaxis	optional	Axis permutation passed to the writer.	None

Source code in lelantos/voidfinder.py

def qso_to_3d(catalog_name, new_name, moveaxis=None):
    """Export a QSO catalog to a text file for 3D visualisation.

    Args:
        catalog_name (str): Input QSO catalog FITS file.
        new_name (str): Output text file.
        moveaxis (optional): Axis permutation passed to the writer.
    """
    qso = tomographic_objects.QSOCatalog.init_from_fits(catalog_name)
    qso.writetxt(new_name, moveaxis=moveaxis)

void_to_3d

void_to_3d(catalog_name, new_name, moveaxis=None)

Export a void catalog to a text file for 3D visualisation.

Parameters:

Name	Type	Description	Default
catalog_name	str	Input void catalog FITS file.	required
new_name	str	Output text file.	required
moveaxis	optional	Axis permutation passed to the writer.	None

Source code in lelantos/voidfinder.py

def void_to_3d(catalog_name, new_name, moveaxis=None):
    """Export a void catalog to a text file for 3D visualisation.

    Args:
        catalog_name (str): Input void catalog FITS file.
        new_name (str): Output text file.
        moveaxis (optional): Axis permutation passed to the writer.
    """
    void = tomographic_objects.VoidCatalog.init_from_fits(catalog_name)
    void.writetxt(new_name, moveaxis=moveaxis)