Pre Module API Reference

Module for pre-processing to generate LandscapeFiles from ST data.

add_custom_segmentation(path_landscape_files, path_segmentation_files, image_scale=1, tile_size=250)

Add custom segmentation to existing landscape files.

Parameters: - path_landscape_files: Path to landscape files - path_segmentation_files: Path to segmentation files - image_scale: Image scale factor - tile_size: Tile size for processing

Source code in src/celldega/pre/__init__.py

def add_custom_segmentation(
    path_landscape_files, path_segmentation_files, image_scale=1, tile_size=250
):
    """
    Add custom segmentation to existing landscape files.

    Parameters:
    - path_landscape_files: Path to landscape files
    - path_segmentation_files: Path to segmentation files
    - image_scale: Image scale factor
    - tile_size: Tile size for processing
    """
    with (Path(path_segmentation_files) / "segmentation_parameters.json").open() as file:
        segmentation_parameters = json.load(file)

    cbg_custom = pd.read_parquet(Path(path_segmentation_files) / "cell_by_gene_matrix.parquet")

    # make sure all genes are present in cbg_custom
    meta_gene = pd.read_parquet(Path(path_landscape_files) / "meta_gene.parquet")
    missing_cols = meta_gene.index.difference(cbg_custom.columns)
    for col in missing_cols:
        cbg_custom[col] = 0

    make_meta_gene(
        cbg=cbg_custom,
        path_output=Path(path_landscape_files)
        / f"meta_gene_{segmentation_parameters['segmentation_approach']}.parquet",
    )

    make_meta_cell_image_coord(
        technology=segmentation_parameters["technology"],
        path_transformation_matrix=str(
            Path(path_landscape_files) / "micron_to_image_transform.csv"
        ),
        path_meta_cell_micron=str(
            Path(path_segmentation_files) / "cell_metadata_micron_space.parquet"
        ),
        path_meta_cell_image=str(
            Path(path_landscape_files)
            / f"cell_metadata_{segmentation_parameters['segmentation_approach']}.parquet"
        ),
        image_scale=image_scale,
    )

    save_cbg_gene_parquets(
        base_path=path_landscape_files,
        cbg=cbg_custom,
        verbose=True,
        segmentation_approach=segmentation_parameters["segmentation_approach"],
    )

    create_cluster_and_meta_cluster(
        technology=segmentation_parameters["technology"],
        path_landscape_files=path_landscape_files,
        segmentation_approach=segmentation_parameters["segmentation_approach"],
    )

    # Get the first .dzi file in sorted order
    dzi_files = sorted((Path(path_landscape_files) / "pyramid_images").glob("*.dzi"))
    if not dzi_files:
        raise FileNotFoundError("No .dzi files found in pyramid_images.")

    # Use the first .dzi file
    tree = ET.parse(dzi_files[0])
    root = tree.getroot()
    width = int(root[0].attrib["Width"])
    height = int(root[0].attrib["Height"])

    tile_bounds = {"x_min": 0, "x_max": width, "y_min": 0, "y_max": height}

    make_cell_boundary_tiles(
        technology=segmentation_parameters["technology"],
        path_cell_boundaries=str(Path(path_segmentation_files) / "cell_polygons.parquet"),
        path_output=str(
            Path(path_landscape_files)
            / f"cell_segmentation_{segmentation_parameters['segmentation_approach']}"
        ),
        tile_size=tile_size,
        tile_bounds=tile_bounds,
        image_scale=image_scale,
    )

    cluster_gene_expression(
        technology=segmentation_parameters["technology"],
        path_landscape_files=path_landscape_files,
        cbg=cbg_custom,
        segmentation_approach=segmentation_parameters["segmentation_approach"],
    )

    save_landscape_parameters(
        technology=segmentation_parameters["technology"],
        path_landscape_files=path_landscape_files,
        image_name="dapi_files",
        tile_size=tile_size,
        image_format=".webp",
        segmentation_approach=segmentation_parameters["segmentation_approach"],
    )

cluster_gene_expression(technology, path_landscape_files, cbg, data_dir=None, segmentation_approach='default')

Calculates cluster-specific gene expression signatures for Xenium data.

Parameters:

Name	Type	Description	Default
technology	str	The technology used (e.g., "Xenium" or "MERSCOPE"). Currently, only "Xenium" is supported.	required
data_dir	str	Path to the directory containing the Xenium data.	None
path_landscape_files	str	Path to the directory where the gene expression signature file will be saved.	required
cbg	DataFrame	A cell-by-gene matrix where rows represent cells and columns represent genes. The index of the DataFrame should match the cell IDs in the Xenium metadata.	required

Raises:

Type	Description
ValueError	If the specified technology is not supported.
FileNotFoundError	If the required input files are not found.

Source code in src/celldega/pre/__init__.py

def cluster_gene_expression(
    technology, path_landscape_files, cbg, data_dir=None, segmentation_approach="default"
):
    """
    Calculates cluster-specific gene expression signatures for Xenium data.

    Args:
        technology (str): The technology used (e.g., "Xenium" or "MERSCOPE"). Currently, only "Xenium" is supported.
        data_dir (str): Path to the directory containing the Xenium data.
        path_landscape_files (str): Path to the directory where the gene expression signature file will be saved.
        cbg (pd.DataFrame): A cell-by-gene matrix where rows represent cells and columns represent genes.
                            The index of the DataFrame should match the cell IDs in the Xenium metadata.

    Raises:
        ValueError: If the specified technology is not supported.
        FileNotFoundError: If the required input files are not found.
    """
    print("\n========Create cluster gene expression (df_sig)========")
    if technology not in ["Xenium", "custom"]:
        raise ValueError(
            f"Unsupported technology: {technology}. Currently, only 'Xenium' and 'Custom' is supported."
        )

    if technology == "Xenium":
        cells_csv_path = Path(data_dir) / "cells.csv.gz"
        clusters_csv_path = (
            Path(data_dir)
            / "analysis"
            / "clustering"
            / "gene_expression_graphclust"
            / "clusters.csv"
        )

        # Load the cell metadata
        usecols = ["cell_id", "x_centroid", "y_centroid"]
        meta_cell = pd.read_csv(cells_csv_path, index_col=0, usecols=usecols)
        meta_cell.columns = ["center_x", "center_y"]

        # Load the clustering data
        df_meta = pd.read_csv(clusters_csv_path, index_col=0)
        df_meta["Cluster"] = df_meta["Cluster"].astype("string")
        df_meta.columns = ["cluster"]

        # Add cluster information to the cell metadata
        meta_cell["cluster"] = df_meta["cluster"]
        clusters = meta_cell["cluster"].unique().tolist()

        # Calculate cluster-specific gene expression signatures
        list_ser = []
        for inst_cat in meta_cell["cluster"].unique().tolist():
            if inst_cat is not None:
                inst_cells = meta_cell[meta_cell["cluster"] == inst_cat].index.tolist()
                inst_ser = cbg.loc[inst_cells].sum() / len(inst_cells)
                inst_ser.name = inst_cat
                list_ser.append(inst_ser)

    elif technology == "custom":
        df_cluster = pd.read_parquet(
            Path(path_landscape_files)
            / f"cell_clusters_{segmentation_approach}"
            / "cluster.parquet"
        )
        clusters = df_cluster["cluster"].unique().tolist()

        list_ser = []
        for inst_cat in df_cluster["cluster"].unique():
            if inst_cat is not None:
                inst_cells = df_cluster[df_cluster["cluster"] == inst_cat].index.tolist()

                if set(inst_cells) & set(cbg.index):
                    common_cells = list(set(inst_cells) & set(cbg.index))
                    inst_ser = cbg.loc[common_cells].sum() / len(common_cells)
                else:
                    genes = cbg.columns
                    inst_ser = pd.Series(0.0, index=genes)

                inst_ser.name = inst_cat
                list_ser.append(inst_ser)

    # Combine the signatures into a DataFrame
    df_sig = pd.concat(list_ser, axis=1)

    # Handle potential multiindex issues
    df_sig.columns = df_sig.columns.tolist()
    df_sig.index = df_sig.index.tolist()

    # Filter out unwanted genes
    keep_genes = df_sig.index.tolist()
    keep_genes = [x for x in keep_genes if "Unassigned" not in x]
    keep_genes = [x for x in keep_genes if "NegControl" not in x]
    keep_genes = [x for x in keep_genes if "DeprecatedCodeword" not in x]

    # Subset the DataFrame to keep only relevant genes and clusters
    df_sig = df_sig.loc[keep_genes, clusters]

    # drop columns with Nan values
    df_sig = df_sig.dropna(axis=1, how="all")

    df_sig = df_sig.loc[sorted(df_sig.index), sorted(df_sig.columns)]

    # Save the gene expression signatures
    segmentation_suffix = f"_{segmentation_approach}" if segmentation_approach != "default" else ""
    output_path = Path(path_landscape_files) / f"df_sig{segmentation_suffix}.parquet"

    if any(isinstance(dtype, pd.SparseDtype) for dtype in df_sig.dtypes):
        df_sig.sparse.to_dense().to_parquet(output_path)
    else:
        df_sig.to_parquet(output_path)

    print("Cluster-specific gene expression signatures saved successfully.")

    return df_sig

create_cluster_and_meta_cluster(technology, path_landscape_files, data_dir=None, segmentation_approach='default')

Creates cell clusters and meta cluster files for visualization. Currently supports only Xenium.

Parameters:

Name	Type	Description	Default
technology	str	The technology used (e.g., "Xenium" or "MERSCOPE"). Currently, only "Xenium" is supported.	required
data_dir	str	Path to the directory containing the Xenium data.	None
path_landscape_files	str	Path to the directory where the cluster and meta cluster files will be saved.	required

Raises:

Type	Description
ValueError	If the specified technology is not supported.
FileNotFoundError	If the required input files are not found.

Source code in src/celldega/pre/__init__.py

def create_cluster_and_meta_cluster(
    technology, path_landscape_files, data_dir=None, segmentation_approach="default"
):
    """
    Creates cell clusters and meta cluster files for visualization.
    Currently supports only Xenium.

    Args:
        technology (str): The technology used (e.g., "Xenium" or "MERSCOPE"). Currently, only "Xenium" is supported.
        data_dir (str): Path to the directory containing the Xenium data.
        path_landscape_files (str): Path to the directory where the cluster and meta cluster files will be saved.

    Raises:
        ValueError: If the specified technology is not supported.
        FileNotFoundError: If the required input files are not found.
    """
    print("\n========Create clusters and meta clusters files========")

    if technology not in ["Xenium", "custom"]:
        raise ValueError(
            f"Unsupported technology: {technology}. Currently, only 'Xenium' and 'Custom' is supported."
        )

    # Check if the cell metadata file exists
    segmentation_suffix = f"_{segmentation_approach}" if segmentation_approach != "default" else ""
    cell_metadata_path = Path(path_landscape_files) / f"cell_metadata{segmentation_suffix}.parquet"

    if not cell_metadata_path.exists():
        raise FileNotFoundError(
            f"The file '{cell_metadata_path.name}' does not exist in directory '{path_landscape_files}'."
        )

    # Create the cell_clusters directory if it doesn't exist
    cell_clusters_dir = Path(path_landscape_files) / f"cell_clusters{segmentation_suffix}"
    cell_clusters_dir.mkdir(exist_ok=True)

    # Load the cell metadata
    meta_cell = pd.read_parquet(cell_metadata_path)

    if technology == "Xenium":
        default_clustering, clusters, ser_counts = _load_xenium_cluster_data(data_dir, meta_cell)
        _save_cluster_data(cell_clusters_dir, default_clustering, clusters, ser_counts)

    elif technology == "custom":
        df_cluster = pd.DataFrame(index=meta_cell["name"].tolist())
        df_cluster["cluster"] = "0"
        df_cluster["cluster"] = df_cluster["cluster"].astype("string")
        df_cluster.to_parquet(cell_clusters_dir / "cluster.parquet")

        meta_cluster = pd.DataFrame(index=["0"])
        meta_cluster.loc["0", "color"] = "#1f77b4"
        meta_cluster.loc["0", "count"] = len(meta_cell["name"].tolist())
        meta_cluster.to_parquet(cell_clusters_dir / "meta_cluster.parquet")

        ser_counts = df_cluster["cluster"].value_counts()
        clusters = ser_counts.index.tolist()

    print("Cell clusters and meta cluster files created successfully.")

    return clusters

create_image_tiles(technology, data_dir, path_landscape_files, image_tile_layer='dapi')

Creates image tiles for visualization from the Xenium morphology image.

Parameters:

Name	Type	Description	Default
technology	str	The technology used (e.g., "Xenium", "MERSCOPE", "VisiumHD", "H&E").	required
data_dir	str	Path to the directory containing the data (e.g., morphology_focus_0000.ome.tif).	required
path_landscape_files	str	Path to the directory where the image tiles and pyramid will be saved.	required
image_tile_layer	str	Specifies which image layers to process. Options for Xenium are	'dapi'

Raises:

Type	Description
ValueError	If the specified technology is not supported or if the image_tile_layer is invalid.
FileNotFoundError	If the required input image file is not found.

Source code in src/celldega/pre/__init__.py

def create_image_tiles(technology, data_dir, path_landscape_files, image_tile_layer="dapi"):
    """
    Creates image tiles for visualization from the Xenium morphology image.

    Args:
        technology (str): The technology used (e.g., "Xenium", "MERSCOPE", "VisiumHD", "H&E").
        data_dir (str): Path to the directory containing the data (e.g., morphology_focus_0000.ome.tif).
        path_landscape_files (str): Path to the directory where the image tiles and pyramid will be saved.
        image_tile_layer (str, optional): Specifies which image layers to process. Options for Xenium are
        'dapi' (default) or 'all'. Use the filename of the .scn file for h&e Landscapes.

    Raises:
        ValueError: If the specified technology is not supported or if the image_tile_layer is invalid.
        FileNotFoundError: If the required input image file is not found.
    """
    print("\n========Generating image tiles========")
    if technology == "Xenium":
        print("------ xenium")
        create_image_tiles_xenium(data_dir, path_landscape_files, image_tile_layer=image_tile_layer)
    elif technology == "MERSCOPE":
        print("------ merscope")
        create_image_tiles_merscope(
            data_dir, path_landscape_files, image_tile_layer=image_tile_layer
        )
    elif technology == "h&e":
        print("------ h&e")
        create_image_tiles_h_and_e(
            data_dir, path_landscape_files, image_tile_layer=image_tile_layer
        )

    print("Image tiles created successfully.")

create_image_tiles_h_and_e(data_dir, path_landscape_files, image_tile_layer)

Creates image tiles for visualization from the H&E image.

Parameters:

Name	Type	Description	Default
data_dir	str	Path to the directory containing the data (e.g., morphology_focus_0000.ome.tif).	required
path_landscape_files	str	Path to the directory where the image tiles and pyramid will be saved.	required
image_tile_layer	str	Specifies the name of the h&e image to process.	required

Raises: FileNotFoundError: If the required input image file is not found.

Source code in src/celldega/pre/__init__.py

def create_image_tiles_h_and_e(data_dir, path_landscape_files, image_tile_layer):
    """
    Creates image tiles for visualization from the H&E image.

    Args:
        data_dir (str): Path to the directory containing the data (e.g., morphology_focus_0000.ome.tif).
        path_landscape_files (str): Path to the directory where the image tiles and pyramid will be saved.
        image_tile_layer (str, optional): Specifies the name of the h&e image to process.
    Raises:
        FileNotFoundError: If the required input image file is not found.
    """
    with tifffile.TiffFile(Path(data_dir) / image_tile_layer) as tif:
        print(tif.pages)  # Show available pages
        image = tif.pages[0].asarray()

        # make this directory, path_landscape_files, if it does not exist
        landscape_path = Path(path_landscape_files)
        landscape_path.mkdir(exist_ok=True)

        temp_tiff_path = landscape_path / image_tile_layer.replace(".scn", "_output_regular.tif")
        tifffile.imwrite(temp_tiff_path, image)

        # Convert the image to PNG format
        image_png = _convert_to_png(str(temp_tiff_path))

        # Create a DeepZoom pyramid for the DAPI channel
        make_deepzoom_pyramid(
            image_png,
            str(landscape_path / "pyramid_images"),
            "h_and_e",
            suffix=".webp[Q=100]",
        )

        remove_intermediate_files(path_landscape_files)

create_image_tiles_merscope(data_dir, path_landscape_files, image_tile_layer='dapi')

Creates image tiles for visualization from the Xenium morphology image.

Parameters:

Name	Type	Description	Default
data_dir	str	Path to the directory containing the data (e.g., morphology_focus_0000.ome.tif).	required
path_landscape_files	str	Path to the directory where the image tiles and pyramid will be saved.	required
image_tile_layer	str	Specifies which image layers to process. Options are 'dapi' (default) or 'all'.	'dapi'

Raises: FileNotFoundError: If the required input image file is not found.

Source code in src/celldega/pre/__init__.py

def create_image_tiles_merscope(data_dir, path_landscape_files, image_tile_layer="dapi"):
    """
    Creates image tiles for visualization from the Xenium morphology image.

    Args:
        data_dir (str): Path to the directory containing the data (e.g., morphology_focus_0000.ome.tif).
        path_landscape_files (str): Path to the directory where the image tiles and pyramid will be saved.
        image_tile_layer (str, optional): Specifies which image layers to process. Options are 'dapi' (default) or 'all'.
    Raises:
        FileNotFoundError: If the required input image file is not found.
    """
    if image_tile_layer not in ["dapi", "all"]:
        raise ValueError(f"Invalid image_tile_layer: {image_tile_layer}. Must be 'dapi' or 'all'.")

    # Define the path to the DAPI image
    dapi_file_path = Path(data_dir) / "images" / "mosaic_DAPI_z3.tif"

    # Check if the DAPI image exists
    if not dapi_file_path.exists():
        raise FileNotFoundError(
            f"The file 'mosaic_DAPI_z3.tif' does not exist in directory '{data_dir}'."
        )

    # Load the DAPI image once if processing multiple channels
    img_dapi = imread(dapi_file_path)

    # Process the DAPI channel
    _process_image_channel(path_landscape_files, {"name": "dapi", "index": 0}, img_dapi)

    # Process additional channels if image_tile_layer is 'all'
    if image_tile_layer == "all":
        # Define the path to the boundary image
        bounda_file_path = Path(data_dir) / "images" / "mosaic_Cellbound1_z3.tif"

        # Check if the boundary image exists
        if not bounda_file_path.exists():
            raise FileNotFoundError(
                f"The file 'mosaic_Cellbound1_z3.tif' does not exist in directory '{data_dir}'."
            )

        # Load the boundary image once if processing multiple channels
        img_bound = imread(bounda_file_path)

        # Process the boundary channel
        _process_image_channel(path_landscape_files, {"name": "bound", "index": 0}, img_bound)

    remove_intermediate_files(path_landscape_files)

create_image_tiles_xenium(data_dir, path_landscape_files, image_tile_layer='dapi')

Creates image tiles for visualization from the Xenium morphology image.

Parameters:

Name	Type	Description	Default
data_dir	str	Path to the directory containing the data (e.g., morphology_focus_0000.ome.tif).	required
path_landscape_files	str	Path to the directory where the image tiles and pyramid will be saved.	required
image_tile_layer	str	Specifies which image layers to process. Options are 'dapi' (default) or 'all'.	'dapi'

Raises: FileNotFoundError: If the required input image file is not found.

Source code in src/celldega/pre/__init__.py

def create_image_tiles_xenium(data_dir, path_landscape_files, image_tile_layer="dapi"):
    """
    Creates image tiles for visualization from the Xenium morphology image.

    Args:
        data_dir (str): Path to the directory containing the data (e.g., morphology_focus_0000.ome.tif).
        path_landscape_files (str): Path to the directory where the image tiles and pyramid will be saved.
        image_tile_layer (str, optional): Specifies which image layers to process. Options are 'dapi' (default) or 'all'.
    Raises:
        FileNotFoundError: If the required input image file is not found.
    """
    if image_tile_layer not in ["dapi", "all"]:
        raise ValueError(f"Invalid image_tile_layer: {image_tile_layer}. Must be 'dapi' or 'all'.")

    # Define the path to the morphology image
    file_path = Path(data_dir) / "morphology_focus" / "morphology_focus_0000.ome.tif"

    # Check if the morphology image exists
    if not file_path.exists():
        raise FileNotFoundError(
            f"The file 'morphology_focus_0000.ome.tif' does not exist in directory '{data_dir}'."
        )

    # Load the morphology image once if processing multiple channels
    img = imread(file_path)

    # Process the DAPI channel
    if image_tile_layer in ["dapi", "all"]:
        _process_image_channel(path_landscape_files, {"name": "dapi", "index": 0}, img)

    # Process additional channels if image_tile_layer is 'all'
    if image_tile_layer == "all":
        for idx, channel in enumerate(["bound", "rna", "prot"]):
            _process_image_channel(path_landscape_files, {"name": channel, "index": idx + 1}, img)

    remove_intermediate_files(path_landscape_files)

get_image_info(technology, image_tile_layer='dapi')

Retrieve image information for a given technology and image tile layer.

Parameters:

Name	Type	Description	Default
technology	str	The technology for which image information is requested. Currently supports 'Xenium' and 'MERSCOPE'.	required
image_tile_layer	str	The type of image tile layer to retrieve information for. Options are 'dapi' or 'all'. Defaults to 'dapi'.	'dapi'

Returns:

Type	Description
list[dict]	A list of dictionaries containing image information, including name,
list[dict]	button name, and color.

Raises:

Type	Description
ValueError	If the technology is not supported or the image_tile_layer is invalid.

Source code in src/celldega/pre/image_info.py

def get_image_info(technology: str, image_tile_layer: str = "dapi") -> list[dict]:
    """
    Retrieve image information for a given technology and image tile layer.

    Args:
        technology: The technology for which image information is requested.
                   Currently supports 'Xenium' and 'MERSCOPE'.
        image_tile_layer: The type of image tile layer to retrieve information for.
                         Options are 'dapi' or 'all'. Defaults to 'dapi'.

    Returns:
        A list of dictionaries containing image information, including name,
        button name, and color.

    Raises:
        ValueError: If the technology is not supported or the image_tile_layer
                   is invalid.
    """
    # Validate technology
    supported_technologies = ["Xenium", "MERSCOPE"]
    if technology not in supported_technologies:
        raise ValueError(
            f"Unsupported technology: {technology}. Supported technologies are: {supported_technologies}."
        )

    # Validate image_tile_layer
    if image_tile_layer not in ["dapi", "all"]:
        raise ValueError(f"Invalid image_tile_layer: {image_tile_layer}. Must be 'dapi' or 'all'.")

    # Handle 'dapi' case for both Xenium and MERSCOPE
    if image_tile_layer == "dapi":
        return [{"name": "dapi", "button_name": "DAPI", "color": [0, 0, 255]}]

    # Handle 'all' case (only for Xenium)
    if technology != "Xenium":
        raise ValueError(
            f"image_tile_layer='all' is only supported for 'Xenium'. "
            f"Received technology: {technology}."
        )

    return [
        {"name": "dapi", "button_name": "DAPI", "color": [0, 0, 255]},
        {"name": "bound", "button_name": "BOUND", "color": [0, 255, 0]},
        {"name": "rna", "button_name": "RNA", "color": [255, 0, 0]},
        {"name": "prot", "button_name": "PROT", "color": [255, 255, 255]},
    ]

get_max_zoom_level(path_image_pyramid)

Returns the maximum zoom level based on the highest-numbered directory in the specified path.

Parameters:

Name	Type	Description	Default
path_image_pyramid	str	Path to the directory containing zoom level directories.	required

Returns:

Name	Type	Description
int		The maximum zoom level.

Source code in src/celldega/pre/__init__.py

def get_max_zoom_level(path_image_pyramid):
    """Returns the maximum zoom level based on the highest-numbered directory in the specified path.

    Args:
        path_image_pyramid (str): Path to the directory containing zoom level directories.

    Returns:
        int: The maximum zoom level.
    """
    path_pyramid = Path(path_image_pyramid)
    zoom_levels = [
        entry.name for entry in path_pyramid.iterdir() if entry.is_dir() and entry.name.isdigit()
    ]
    return max(map(int, zoom_levels)) if zoom_levels else None

main(sample, data_root_dir, tile_size, image_tile_layer='all', path_landscape_files='', use_int_index=True, max_workers=1)

Main function to preprocess Xenium or MERSCOPE data and generate landscape files.

Parameters:

Name	Type	Description	Default
sample	str	Name of the sample (e.g., 'Xenium_V1_human_Pancreas_FFPE_outs').	required
data_root_dir	str	Root directory containing all sample data. The sample name will be appended to this path to locate the specific dataset.	required
tile_size	int	Size of the tiles for transcript and boundary tiles.	required
image_tile_layer	str	Image layers to be tiled. 'dapi' or 'all'.	'all'
path_landscape_files	str	Directory to save the landscape files.	''
use_int_index	bool	Use integer index for smaller files and faster rendering.	True

Example

change directory to celldega, and run:

python run_pre_processing.py --sample Xenium_V1_human_Pancreas_FFPE_outs --data_root_dir data --tile_size 250 --image_tile_layer 'dapi' --path_landscape_files notebooks/Xenium_V1_human_Pancreas_FFPE_outs

Source code in src/celldega/pre/run_pre_processing.py

def main(
    sample,
    data_root_dir,
    tile_size,
    image_tile_layer="all",
    path_landscape_files="",
    use_int_index=True,
    max_workers=1,
):
    """
    Main function to preprocess Xenium or MERSCOPE data and generate landscape files.

    Args:
        sample (str): Name of the sample (e.g., 'Xenium_V1_human_Pancreas_FFPE_outs').
        data_root_dir (str): Root directory containing all sample data. The
            ``sample`` name will be appended to this path to locate the
            specific dataset.
        tile_size (int): Size of the tiles for transcript and boundary tiles.
        image_tile_layer (str): Image layers to be tiled. 'dapi' or 'all'.
        path_landscape_files (str): Directory to save the landscape files.
        use_int_index (bool): Use integer index for smaller files and faster rendering.

    Example:
        change directory to celldega, and run:

        python run_pre_processing.py \
            --sample Xenium_V1_human_Pancreas_FFPE_outs \
            --data_root_dir data \
            --tile_size 250 \
            --image_tile_layer 'dapi' \
            --path_landscape_files notebooks/Xenium_V1_human_Pancreas_FFPE_outs

    """
    print(f"Starting preprocessing for sample: {sample}")

    # Construct data directory
    data_dir = Path(data_root_dir) / sample

    # Create necessary directories if they don't exist
    _create_directories([data_dir, path_landscape_files])

    # Determine technology
    technology = _determine_technology(data_dir)

    # Setup file paths
    paths = _setup_preprocessing_paths(technology, path_landscape_files, data_dir)

    # Save transformation matrices

    if technology == "Xenium":
        # Unzip compressed files in Xenium data folder
        dega.pre._xenium_unzipper(str(data_dir))
        # Write transform file
        dega.pre.write_xenium_transform(str(data_dir), path_landscape_files)

    elif technology == "MERSCOPE":
        source_path = Path(paths["transformation_matrix"])

        # Copy the file to the destination directory, keeping the same filename
        shutil.copy(source_path, Path(path_landscape_files) / "micron_to_image_transform.csv")

    # Check required files for preprocessing
    dega.pre._check_required_files(technology, str(data_dir))

    # Make cell image coordinates
    dega.pre.make_meta_cell_image_coord(
        technology,
        str(paths["transformation_matrix"]),
        str(paths["meta_cell_micron"]),
        str(paths["meta_cell_image"]),
        image_scale=1,
    )

    # Calculate CBG
    if technology == "Xenium":
        cbg = dega.pre.read_cbg_mtx(str(paths["cbg_matrix"]))
    elif technology == "MERSCOPE":
        cbg = pd.read_csv(str(paths["cbg_csv"]), index_col=0)

    if technology == "Xenium":
        # Create cluster-based gene expression
        dega.pre.cluster_gene_expression(technology, path_landscape_files, cbg, str(data_dir))
    elif technology == "MERSCOPE":
        create_dummy_clusters(path_landscape_files, cbg)

    # Make meta gene files
    dega.pre.make_meta_gene(cbg, str(paths["meta_gene"]))

    # Save CBG gene parquet files
    dega.pre.save_cbg_gene_parquets(path_landscape_files, cbg, verbose=True)

    if technology == "Xenium":
        # Create cluster and meta cluster files
        dega.pre.create_cluster_and_meta_cluster(technology, path_landscape_files, str(data_dir))

    # Generate image tiles
    dega.pre.create_image_tiles(
        technology, str(data_dir), path_landscape_files, image_tile_layer=image_tile_layer
    )

    # Generate transcript tiles
    print("\n========Generating transcript tiles========")
    tile_bounds = dega.pre.make_trx_tiles(
        technology,
        str(paths["transcripts"]),
        str(paths["transformation_matrix"]),
        str(paths["transcript_tiles"]),
        coarse_tile_factor=10,
        tile_size=tile_size,
        chunk_size=100000,
        verbose=False,
        image_scale=1,
        max_workers=max_workers,
    )
    print(f"tile bounds: {tile_bounds}")

    # Generate boundary tiles
    print("\n========Generating boundary tiles========")
    dega.pre.make_cell_boundary_tiles(
        technology,
        str(paths["cell_boundaries"]),
        str(paths["cell_segmentation"]),
        str(paths["meta_cell_micron"]),
        str(paths["transformation_matrix"]),
        coarse_tile_factor=10,
        tile_size=tile_size,
        tile_bounds=tile_bounds,
        max_workers=max_workers,
    )

    # Force name to be str for MERSCOPE
    if technology == "MERSCOPE":
        cell_meta = pd.read_parquet(str(paths["meta_cell_image"]))
        cell_meta["name"] = cell_meta["name"].astype(str)
        cell_meta.to_parquet(str(paths["meta_cell_image"]))

    # Save landscape parameters
    dega.pre.save_landscape_parameters(
        technology,
        path_landscape_files,
        "dapi_files",
        tile_size=tile_size,
        image_info=dega.pre.get_image_info(technology, image_tile_layer),
        image_format=".webp",
        use_int_index=use_int_index,
    )

    print("Preprocessing completed successfully.")

make_chromium_from_anndata(adata, path_landscape_files)

Generate minimal LandscapeFiles from a Chromium AnnData object.

Parameters

adata : anndata.AnnData AnnData object containing scRNA-seq count data. path_landscape_files : str or Path Directory where LandscapeFiles will be written.

Raises

ValueError If the expression matrix contains non-integer values.

Source code in src/celldega/pre/__init__.py

def make_chromium_from_anndata(adata, path_landscape_files):
    """Generate minimal LandscapeFiles from a Chromium AnnData object.

    Parameters
    ----------
    adata : anndata.AnnData
        AnnData object containing scRNA-seq count data.
    path_landscape_files : str or Path
        Directory where LandscapeFiles will be written.

    Raises
    ------
    ValueError
        If the expression matrix contains non-integer values.
    """

    print("\n========Process Chromium AnnData========")
    path_landscape_files = Path(path_landscape_files)
    path_landscape_files.mkdir(parents=True, exist_ok=True)

    X = adata.layers.get("counts", adata.X)

    data = X.data if issparse(X) else np.asarray(X)

    if not np.all(np.equal(np.mod(data, 1), 0)):
        raise ValueError("Chromium processing requires integer counts")

    if issparse(X):
        cbg = pd.DataFrame.sparse.from_spmatrix(X, index=adata.obs_names, columns=adata.var_names)
    else:
        cbg = pd.DataFrame(X, index=adata.obs_names, columns=adata.var_names)

    cell_meta = pd.DataFrame({"name": adata.obs_names, "geometry": [[0.0, 0.0]] * adata.n_obs})
    cell_meta.to_parquet(path_landscape_files / "cell_metadata.parquet", index=False)

    save_cbg_gene_parquets(path_landscape_files, cbg)

    make_meta_gene(cbg, path_landscape_files / "meta_gene.parquet")

    (path_landscape_files / "pyramid_images").mkdir(exist_ok=True)

    save_landscape_parameters(
        technology="Chromium",
        path_landscape_files=path_landscape_files,
        image_name="",
        tile_size=1,
        image_info=[],
    )

make_deepzoom_pyramid(image_path, output_path, pyramid_name, tile_size=512, overlap=0, suffix='.jpeg')

Creates a DeepZoom image pyramid from a JPEG image.

Parameters:

Name	Type	Description	Default
image_path	str	Path to the JPEG image file.	required
output_path	str	Directory to save the DeepZoom pyramid.	required
pyramid_name	str	Name of the pyramid directory.	required
tile_size	int	Tile size for the DeepZoom pyramid. Defaults to 512.	512
overlap	int	Overlap size for the DeepZoom pyramid. Defaults to 0.	0
suffix	str	Suffix for the DeepZoom pyramid tiles. Defaults to ".jpeg".	'.jpeg'

Returns:

Type	Description
	None

Source code in src/celldega/pre/__init__.py

def make_deepzoom_pyramid(
    image_path, output_path, pyramid_name, tile_size=512, overlap=0, suffix=".jpeg"
):
    """Creates a DeepZoom image pyramid from a JPEG image.

    Args:
        image_path (str): Path to the JPEG image file.
        output_path (str): Directory to save the DeepZoom pyramid.
        pyramid_name (str): Name of the pyramid directory.
        tile_size (int, optional): Tile size for the DeepZoom pyramid. Defaults to 512.
        overlap (int, optional): Overlap size for the DeepZoom pyramid. Defaults to 0.
        suffix (str, optional): Suffix for the DeepZoom pyramid tiles. Defaults to ".jpeg".

    Returns:
        None
    """
    output_path = Path(output_path)
    image = pyvips.Image.new_from_file(image_path, access="sequential")
    output_path.mkdir(parents=True, exist_ok=True)
    output_path = output_path / pyramid_name
    image.dzsave(str(output_path), tile_size=tile_size, overlap=overlap, suffix=suffix)

make_meta_cell_image_coord(technology, path_transformation_matrix, path_meta_cell_micron, path_meta_cell_image, image_scale=1)

Applies an affine transformation to cell coordinates in microns and saves the transformed coordinates in pixels.

Parameters

technology : str The technology used to generate the data, Xenium and MERSCOPE are supported. path_transformation_matrix : str Path to the transformation matrix file path_meta_cell_micron : str Path to the meta cell file with coordinates in microns path_meta_cell_image : str Path to save the meta cell file with coordinates in pixels

Returns

None

Examples

make_meta_cell_image_coord( ... technology='Xenium', ... path_transformation_matrix='data/transformation_matrix.csv', ... path_meta_cell_micron='data/meta_cell_micron.csv', ... path_meta_cell_image='data/meta_cell_image.parquet' ... ) Args: technology (str): The technology used to generate the data (e.g., "Xenium" or "MERSCOPE"). path_transformation_matrix (str): Path to the transformation matrix file. path_meta_cell_micron (str): Path to the meta cell file with coordinates in microns. path_meta_cell_image (str): Path to save the meta cell file with coordinates in pixels. image_scale (float): Scaling factor to convert micron coordinates to pixel coordinates.

Returns:

Type	Description
	None

Source code in src/celldega/pre/__init__.py

def make_meta_cell_image_coord(
    technology,
    path_transformation_matrix,
    path_meta_cell_micron,
    path_meta_cell_image,
    image_scale=1,
):
    """Applies an affine transformation to cell coordinates in microns and saves the transformed coordinates in pixels.

    Parameters
    ----------
    technology : str
        The technology used to generate the data, Xenium and MERSCOPE are supported.
    path_transformation_matrix : str
        Path to the transformation matrix file
    path_meta_cell_micron : str
        Path to the meta cell file with coordinates in microns
    path_meta_cell_image : str
        Path to save the meta cell file with coordinates in pixels

    Returns
    -------
    None

    Examples
    --------
    >>> make_meta_cell_image_coord(
    ...     technology='Xenium',
    ...     path_transformation_matrix='data/transformation_matrix.csv',
    ...     path_meta_cell_micron='data/meta_cell_micron.csv',
    ...     path_meta_cell_image='data/meta_cell_image.parquet'
    ... )
    Args:
        technology (str): The technology used to generate the data (e.g., "Xenium" or "MERSCOPE").
        path_transformation_matrix (str): Path to the transformation matrix file.
        path_meta_cell_micron (str): Path to the meta cell file with coordinates in microns.
        path_meta_cell_image (str): Path to save the meta cell file with coordinates in pixels.
        image_scale (float): Scaling factor to convert micron coordinates to pixel coordinates.

    Returns:
        None
    """
    print("\n========Make meta cells in pixel space========")
    transformation_matrix = pd.read_csv(path_transformation_matrix, header=None, sep=" ").values
    sparse_matrix = csr_matrix(transformation_matrix)

    meta_cell = _load_meta_cell_by_technology(technology, path_meta_cell_micron)

    # Adding a ones column to accommodate for affine transformation
    meta_cell["ones"] = 1
    points = meta_cell[["center_x", "center_y", "ones"]].values

    # Applying the transformation matrix
    transformed_points = sparse_matrix.dot(points.T).T[:, :2]

    meta_cell["center_x"] = transformed_points[:, 0]
    meta_cell["center_y"] = transformed_points[:, 1]
    meta_cell.drop(columns=["ones"], inplace=True)

    meta_cell["center_x"] = meta_cell["center_x"] / image_scale
    meta_cell["center_y"] = meta_cell["center_y"] / image_scale

    meta_cell["geometry"] = meta_cell.apply(lambda row: [row["center_x"], row["center_y"]], axis=1)

    if technology == "MERSCOPE":
        meta_cell = meta_cell[["name", "geometry", "EntityID"]]
    else:
        meta_cell = meta_cell[["name", "geometry"]]

    # Check if the 'name' column is unique
    if not meta_cell["name"].is_unique:
        warnings.warn("Duplicate cell names found in meta_cell!", UserWarning, stacklevel=2)

    # Apply rounding to the GEOMETRY column
    meta_cell["geometry"] = meta_cell["geometry"].apply(_round_nested_coord_list)

    # Force alphabetically sort by 'name'
    meta_cell = meta_cell.sort_values(by=["name"]).reset_index(drop=True)
    meta_cell.to_parquet(path_meta_cell_image, index=False)
    print("Done.")

make_meta_gene(cbg, path_output)

Creates a DataFrame with genes and their assigned colors.

Parameters:

Name	Type	Description	Default
cbg	DataFrame	A sparse DataFrame with genes as columns and barcodes as rows..	required
path_output	str	Path to save the meta gene file.	required

Returns:

Type	Description
	None

Source code in src/celldega/pre/__init__.py

def make_meta_gene(cbg, path_output):
    """Creates a DataFrame with genes and their assigned colors.

    Args:
        cbg (pandas.DataFrame): A sparse DataFrame with genes as columns and barcodes as rows..
        path_output (str): Path to save the meta gene file.

    Returns:
        None
    """
    print("\n========Write meta gene files========")
    genes = cbg.columns.tolist()

    palettes = [plt.get_cmap(name).colors for name in plt.colormaps() if "tab" in name]
    flat_colors = [color for palette in palettes for color in palette]
    flat_colors_hex = [to_hex(color) for color in flat_colors]

    colors = [
        flat_colors_hex[i % len(flat_colors_hex)] if "Blank" not in gene else "#FFFFFF"
        for i, gene in enumerate(genes)
    ]

    ser_color = pd.Series(colors, index=genes)
    meta_gene = calc_meta_gene_data(cbg)
    meta_gene["color"] = ser_color

    sparse_cols = [col for col in meta_gene.columns if pd.api.types.is_sparse(meta_gene[col])]
    for col in sparse_cols:
        meta_gene[col] = meta_gene[col].sparse.to_dense()

    # Force alphabetically sort by index
    meta_gene.sort_index(inplace=True)
    meta_gene.to_parquet(path_output)
    print("All meta gene files are succesfully saved.")

make_trx_tiles(technology, path_trx, path_transformation_matrix=None, path_trx_tiles=None, coarse_tile_factor=10, tile_size=250, chunk_size=1000000, verbose=False, image_scale=1, max_workers=1)

Processes transcript data by dividing it into coarse-grain and fine-grain tiles, applying transformations, and saving the results in a parallelized manner.

Parameters

technology : str The technology used for generating the transcript data (e.g., "MERSCOPE" or "Xenium"). path_trx : str Path to the file containing the transcript data. path_transformation_matrix : str Path to the file containing the transformation matrix (CSV file). path_trx_tiles : str Directory path where the output files (Parquet files) for each tile will be saved. coarse_tile_factor : int, optional Scaling factor of each coarse-grain tile comparing to the fine tile size. tile_size : int, optional Size of each fine-grain tile in microns (default is 250). chunk_size : int, optional Number of rows to process per chunk for memory efficiency (default is 1000000). verbose : bool, optional Flag to enable verbose output (default is False). image_scale : float, optional Scale factor to apply to the transcript coordinates (default is 0.5). max_workers : int, optional Maximum number of parallel workers for processing tiles (default is 1).

Returns

dict A dictionary containing the bounds of the processed data in both x and y directions.

Source code in src/celldega/pre/trx_tile.py

def make_trx_tiles(
    technology,
    path_trx,
    path_transformation_matrix=None,
    path_trx_tiles=None,
    coarse_tile_factor=10,
    tile_size=250,
    chunk_size=1000000,
    verbose=False,
    image_scale=1,
    max_workers=1,
):
    """
    Processes transcript data by dividing it into coarse-grain and fine-grain tiles,
    applying transformations, and saving the results in a parallelized manner.

    Parameters
    ----------
    technology : str
        The technology used for generating the transcript data (e.g., "MERSCOPE" or "Xenium").
    path_trx : str
        Path to the file containing the transcript data.
    path_transformation_matrix : str
        Path to the file containing the transformation matrix (CSV file).
    path_trx_tiles : str
        Directory path where the output files (Parquet files) for each tile will be saved.
    coarse_tile_factor : int, optional
        Scaling factor of each coarse-grain tile comparing to the fine tile size.
    tile_size : int, optional
        Size of each fine-grain tile in microns (default is 250).
    chunk_size : int, optional
        Number of rows to process per chunk for memory efficiency (default is 1000000).
    verbose : bool, optional
        Flag to enable verbose output (default is False).
    image_scale : float, optional
        Scale factor to apply to the transcript coordinates (default is 0.5).
    max_workers : int, optional
        Maximum number of parallel workers for processing tiles (default is 1).

    Returns
    -------
    dict
        A dictionary containing the bounds of the processed data in both x and y directions.
    """
    if technology == "custom":
        x_min, y_min, x_max, y_max = _get_custom_tile_bounds(path_trx)
    else:
        # Create output directory
        tiles_path = Path(path_trx_tiles)
        tiles_path.mkdir(exist_ok=True)

        transformation_matrix = np.loadtxt(path_transformation_matrix)

        gene_str_to_int_mapping = _get_name_mapping(
            path_trx_tiles.replace("/transcript_tiles", ""),
            layer="transcript",
        )

        trx = transform_transcript_coordinates(
            technology,
            path_trx,
            chunk_size,
            transformation_matrix,
            image_scale,
            gene_str_to_int_mapping=gene_str_to_int_mapping,
        )

        # Get min and max x, y values
        x_min, y_min, x_max, y_max = _get_transformed_tile_bounds(trx)

        # Process tiles in parallel
        _process_transcript_tiles_parallel(
            trx,
            x_min,
            y_min,
            x_max,
            y_max,
            tile_size,
            coarse_tile_factor,
            path_trx_tiles,
            max_workers,
        )

    # Return the tile bounds
    return {
        "x_min": x_min,
        "x_max": x_max,
        "y_min": y_min,
        "y_max": y_max,
    }

read_cbg_mtx(base_path)

Read the cell-by-gene matrix from the mtx files.

Parameters

base_path : str The base path to the directory containing the mtx files.

Returns

cbg : pandas.DataFrame A sparse DataFrame with genes as columns and barcodes as rows.

Source code in src/celldega/pre/landscape.py

def read_cbg_mtx(base_path):
    """
    Read the cell-by-gene matrix from the mtx files.

    Parameters
    ----------
    base_path : str
        The base path to the directory containing the mtx files.

    Returns
    -------
    cbg : pandas.DataFrame
        A sparse DataFrame with genes as columns and barcodes as rows.
    """
    base_path = Path(base_path)

    # File paths
    barcodes_path = base_path / "barcodes.tsv.gz"
    features_path = base_path / "features.tsv.gz"
    matrix_path = base_path / "matrix.mtx.gz"

    # Read barcodes and features
    barcodes = pd.read_csv(barcodes_path, header=None, compression="gzip")
    features = pd.read_csv(features_path, header=None, compression="gzip", sep="\t")

    # Read the gene expression matrix and transpose it
    # Transpose and convert to CSC format for fast column slicing
    matrix = mmread(matrix_path).transpose().tocsc()

    # Create a sparse DataFrame with genes as columns and barcodes as rows
    cbg = pd.DataFrame.sparse.from_spmatrix(matrix, index=barcodes[0], columns=features[1])

    return cbg.rename_axis("__index_level_0__", axis="columns")

remove_intermediate_files(path_landscape_files)

Remove intermediate image files.

Parameters: - path_landscape_files: Path to landscape files directory

Source code in src/celldega/pre/__init__.py

def remove_intermediate_files(path_landscape_files):
    """
    Remove intermediate image files.

    Parameters:
    - path_landscape_files: Path to landscape files directory
    """
    # Remove intermediate files
    intermediate_image_files = list(Path(path_landscape_files).glob("*output_regular*"))
    for file in intermediate_image_files:
        file.unlink()

save_landscape_parameters(technology, path_landscape_files, image_name='dapi_files', tile_size=1000, image_info=None, image_format='.webp', use_int_index=True, segmentation_approach='default')

Saves the landscape parameters to a JSON file.

Parameters:

Name	Type	Description	Default
technology	str	The technology used to generate the data.	required
path_landscape_files	str	Path to the directory where landscape files are stored.	required
image_name	str	Name of the image directory. Defaults to "dapi_files".	'dapi_files'
tile_size	int	Tile size for the image pyramid. Defaults to 1000.	1000
image_info	dict	Additional image metadata. Defaults to None.	None
image_format	str	Format of the image files. Defaults to ".webp".	'.webp'
use_int_index	bool	Use integer name for cell_tile and trx_tile.	True

Returns:

Type	Description
	None

Source code in src/celldega/pre/__init__.py

def save_landscape_parameters(
    technology,
    path_landscape_files,
    image_name="dapi_files",
    tile_size=1000,
    image_info=None,
    image_format=".webp",
    use_int_index=True,
    segmentation_approach="default",
):
    """Saves the landscape parameters to a JSON file.

    Args:
        technology (str): The technology used to generate the data.
        path_landscape_files (str): Path to the directory where landscape files are stored.
        image_name (str, optional): Name of the image directory. Defaults to "dapi_files".
        tile_size (int, optional): Tile size for the image pyramid. Defaults to 1000.
        image_info (dict, optional): Additional image metadata. Defaults to None.
        image_format (str, optional): Format of the image files. Defaults to ".webp".
        use_int_index (bool, optional): Use integer name for cell_tile and trx_tile.

    Returns:
        None
    """
    print("\n========Save landscape parameters========")

    if image_info is None:
        image_info = {}

    if technology == "h&e":
        image_name = "h_and_e_files"
        image_info = [{"name": "h&e", "button_name": "H&E", "color": [0, 0, 255]}]

    path_image_pyramid = Path(path_landscape_files) / "pyramid_images" / image_name
    max_pyramid_zoom = get_max_zoom_level(path_image_pyramid)

    path_landscape_parameters = Path(path_landscape_files) / "landscape_parameters.json"

    # if technology is 'h&e' set parameters
    if technology == "h&e":
        landscape_parameters = {
            "technology": technology,
            "segmentation_approach": ["N.A."],
            "max_pyramid_zoom": max_pyramid_zoom,
            "tile_size": "N.A.",
            "image_info": image_info,
            "image_format": image_format,
            "use_int_index": "N.A.",
        }
    elif technology != "custom":
        landscape_parameters = {
            "technology": technology,
            "segmentation_approach": [segmentation_approach],
            "max_pyramid_zoom": max_pyramid_zoom,
            "tile_size": tile_size,
            "image_info": image_info,
            "image_format": image_format,
            "use_int_index": use_int_index,
        }
    else:
        with path_landscape_parameters.open() as file:
            landscape_parameters = json.load(file)
        landscape_parameters["segmentation_approach"].append(segmentation_approach)

    with path_landscape_parameters.open("w") as file:
        json.dump(landscape_parameters, file, indent=4)

    print("Done.")

write_xenium_transform(data_dir, path_landscape_files, transform_fname='micron_to_image_transform.csv')

Extracts the transformation matrix from the Xenium cells.zarr.zip file and saves it as a CSV file.

Parameters:

Name	Type	Description	Default
data_dir	str	Path to the directory containing the Xenium data (e.g., cells.zarr.zip).	required
path_landscape_files	str	Path to the directory where the transformation matrix CSV will be saved.	required
transform_fname	str	Name of the output CSV file. Defaults to "micron_to_image_transform.csv".	'micron_to_image_transform.csv'

Returns:

Type	Description
	numpy.ndarray: The full transformation matrix extracted from the Xenium cells.zarr.zip file.

Raises:

Type	Description
FileNotFoundError	If the cells.zarr.zip file does not exist in the specified data_dir.
KeyError	If the transformation matrix is not found in the Zarr file under the expected path.
Exception	If an unexpected error occurs while processing the Zarr file.

Source code in src/celldega/pre/__init__.py

def write_xenium_transform(
    data_dir, path_landscape_files, transform_fname="micron_to_image_transform.csv"
):
    """
    Extracts the transformation matrix from the Xenium cells.zarr.zip file and saves it as a CSV file.

    Args:
        data_dir (str): Path to the directory containing the Xenium data (e.g., cells.zarr.zip).
        path_landscape_files (str): Path to the directory where the transformation matrix CSV will be saved.
        transform_fname (str, optional): Name of the output CSV file. Defaults to "micron_to_image_transform.csv".

    Returns:
        numpy.ndarray: The full transformation matrix extracted from the Xenium cells.zarr.zip file.

    Raises:
        FileNotFoundError: If the cells.zarr.zip file does not exist in the specified `data_dir`.
        KeyError: If the transformation matrix is not found in the Zarr file under the expected path.
        Exception: If an unexpected error occurs while processing the Zarr file.
    """
    print("\n========Write xenium transform file from the Zarr folder========")
    # Path to the cells.zarr.zip file
    cells_zarr_path = Path(data_dir) / "cells.zarr.zip"

    # Check if the cells.zarr.zip file exists
    if not cells_zarr_path.exists():
        raise FileNotFoundError(
            f"The file 'cells.zarr.zip' does not exist in directory '{data_dir}'."
        )

    # Function to open a Zarr file
    def open_zarr(path: str) -> zarr.Group:
        store = (
            zarr.ZipStore(path, mode="r") if path.endswith(".zip") else zarr.DirectoryStore(path)
        )
        return zarr.group(store=store)

    try:
        # Open the cells Zarr file
        root = open_zarr(str(cells_zarr_path))

        # Extract the transformation matrix
        transformation_matrix = root["masks"]["homogeneous_transform"][:]

        # Save the transformation matrix as a CSV file
        output_path = Path(path_landscape_files) / transform_fname
        pd.DataFrame(transformation_matrix[:3, :3]).to_csv(
            output_path, sep=" ", header=False, index=False
        )

        print(f"Transformation matrix saved to '{output_path}'.")
    except KeyError as e:
        raise KeyError(f"Could not find the transformation matrix in the Zarr file: {e}") from e
    except Exception as e:
        raise Exception(f"An error occurred while processing the Zarr file: {e}") from e

    return transformation_matrix