Celldega is a spatial analysis and visualization library that is being developed by the Spatial Technology Platform at the Broad Institute of MIT and Harvard. This project enables researchers to easily visualize large ST datasets (e.g., datasets with >100M transcripts) alongside single-cell and spatial analysis notebook workflows (e.g., integrating with sverse tools and utilizing novel spatial analysis approaches).
Celldega is a spatial analysis and visualization library that is being developed by the Spatial Technology Platform at the Broad Institute of MIT and Harvard. This project enables researchers to easily visualize large ST datasets (e.g., datasets with >100M transcripts) alongside single-cell and spatial analysis notebook workflows (e.g., sverse tools and novel spatial analysis approaches).
Celldega enables researchers to easily visualize large ST datasets (e.g., datasets with >100M transcripts) alongside single-cell and spatial analysis notebook workflows (e.g., integrating with sverse tools and utilizing novel spatial analysis approaches).
The Celldega library is being developed to help researchers easily visualize and analyze high-dimensional spatial-omics data in the context of a notebook workflow. Initial development has been focused on spatial transcriptomics visualization.
Celldega can be used as a Python library in a Jupyter notebook environment or as a stand-alone JavaScript library for creating visualizations.
The pre module contains methods for pre-processing LandscapeFiles.
pre
The viz module contains functions and classes for data visualization.
viz
widget
Celldega is named after a Bodega, a small shop with all the essentials, that is part of the fabric of a neighborhood.
Landscape View Xenium
# %load_ext autoreload\n# %autoreload 2\n# %env ANYWIDGET_HMR=1\n
import celldega as dega\ndega.__version__\n
'0.0.0'
from observable_jupyter import embed\n
base_url = 'https://raw.githubusercontent.com/broadinstitute/celldega_Xenium_Prime_Mouse_Brain_Coronal_FF_outs/main/Xenium_Prime_Mouse_Brain_Coronal_FF_outs'\n
embed('@cornhundred/celldega-landscape-ist', inputs={'base_url': base_url}, cells=['landscape_container'], display_logo=False)\n
# base_url = 'https://raw.githubusercontent.com/broadinstitute/celldega_Xenium_Prime_Human_Skin_FFPE_outs/main/Xenium_Prime_Human_Skin_FFPE_outs'\n\n# landscape_ist = dega.viz.Landscape(\n# technology='Xenium',\n# ini_zoom = -4.5,\n# ini_x=6000,\n# ini_y=8000,\n# base_url = base_url,\n\n# )\n\n# landscape_ist\n
\n
This page includes links to visualizations that are made with the stand-alone Celldega JavaScript library.
Celldega's visualization methods can be used as a stand-alone JavaScript library outside the context of a Jupyter notebook. This can be used to create showcase visualizations with publicly hosted data.
The JavaScript component of Celldega is used within the Jupyter Widgets framework to provide interactive visualization in the context of a Jupyter notebook but can also be used as a standalone JavaScript library.
landscape_ist
The landscape_ist function initializes and renders an interactive spatial transcriptomics (IST) landscape visualization. This API is designed to work with Deck.gl and includes customizable visualization options, dynamic data updates, and UI interactions.
el
HTMLElement
ini_model
Object
token
string
ini_x
ini_y
ini_z
number
ini_zoom
base_url
dataset_name
trx_radius
0.25
width
number|string
100%
height
800
view_change_custom_callback
Function
The landscape_ist function returns an object (landscape) with several methods for interacting with the visualization.
landscape
update_matrix_gene
Updates the visualization to highlight data for a specific gene.
inst_gene
update_matrix_col
Updates the visualization to highlight data for a specific column (e.g., cluster).
inst_col
update_matrix_dendro_col
Updates the visualization based on a dendrogram selection of columns.
selected_cols
Array<string>
update_view_state
Updates the view state of the Deck.gl visualization.
new_view_state
close_up
boolean
trx_layer
update_layers
Updates all visualization layers.
finalize
Finalizes the Deck.gl instance and cleans up resources.
\njavascript\nimport { landscape_ist } from 'path/to/landscape_ist';\n\nconst rootElement = document.getElementById('visualization-container');\nconst model = { /* Model containing visualization data */ };\n\nconst visualization = await landscape_ist(\n rootElement,\n model,\n 'example-token',\n 100,\n 200,\n 0,\n -5,\n 'https://example.com/data',\n 'Example Dataset'\n);\n\n// Update the visualization with a specific gene.\nvisualization.update_matrix_gene('TP53');\n\n// Update the visualization with a specific column.\nvisualization.update_matrix_col('Cluster 1');\n\n// Finalize the visualization when done.\nvisualization.finalize();\n\n
matrix_viz
The matrix_viz function initializes and renders a matrix visualization. This API is built using approaches and code adaptations from the Clustergrammer-GL library, and it integrates tightly with Deck.gl to provide interactive and dynamic visualizations.
model
network
string|number
'800'
row_label_callback
col_label_callback
col_dendro_callback
The function performs the following setup: 1. Deck.gl Integration: - Initializes a Deck.gl instance for the matrix visualization. - Sets properties for interactivity, including tooltips, view state changes, and layer filtering.
Configures labels, categories, and dendrograms for both rows and columns.
Layer Initialization:
Attaches interactions (e.g., click events) to these layers.
UI Setup:
import { matrix_viz } from 'path/to/matrix_viz';\n\nconst rootElement = document.getElementById('matrix-container');\nconst model = { /* Model containing visualization data */ };\nconst network = { /* Network object representing the matrix data */ };\n\n// Callback functions\nconst rowLabelCallback = (row) => {\n console.log('Row label clicked:', row);\n};\n\nconst colLabelCallback = (col) => {\n console.log('Column label clicked:', col);\n};\n\nconst colDendroCallback = (dendro) => {\n console.log('Column dendrogram clicked:', dendro);\n};\n\n// Initialize the matrix visualization\nawait matrix_viz(\n model,\n rootElement,\n network,\n 800,\n 800,\n rowLabelCallback,\n colLabelCallback,\n colDendroCallback\n);\n
Celldega is a spatial analysis and visualization library that is being developed by the Spatial Technology Platform at the Broad Institute of MIT and Harvard. Celldega can be used as a Jupyter Widget in Python as well as a stand-alone JavaScript library.
Please see examples notebooks below to try out Celldega in a Jupyter notebook or ObservableHQ JavaScript notebook:
The Celldega library can be installed using pip
# install Celldega (without vips for visualization pre-processing)\npip install celldega\n\n# install Celldega with optional pre-processing requirements\npip install celldega[pre]\n\n
Celldega can be used in a JavaScript environment such as ObservableHQ by importing it as a module
celldega = import('https://unpkg.com/celldega@' + version + '/src/celldega/static/widget.js?module')\n
** Coming soon **
Module for pre-processing to generate LandscapeFiles from ST data.
Module for visualization
convert_long_id_to_short(df)
Converts a column of long integer cell IDs in a DataFrame to a shorter, hash-based representation.
Parameters:
df
DataFrame
The DataFrame containing the EntityID.
Returns: pd.DataFrame: The original DataFrame with an additional column named cell_id containing the shortened cell IDs.
cell_id
The function applies a SHA-256 hash to each cell ID, encodes the hash using base64, and truncates it to create a shorter identifier that is added as a new column to the DataFrame.
src/celldega/pre/__init__.py
def convert_long_id_to_short(df):\n \"\"\"\n Converts a column of long integer cell IDs in a DataFrame to a shorter, hash-based representation.\n\n Args:\n df (pd.DataFrame): The DataFrame containing the EntityID.\n Returns:\n pd.DataFrame: The original DataFrame with an additional column named `cell_id`\n containing the shortened cell IDs.\n\n The function applies a SHA-256 hash to each cell ID, encodes the hash using base64, and truncates\n it to create a shorter identifier that is added as a new column to the DataFrame.\n \"\"\"\n # Function to hash and encode the cell ID\n def hash_and_shorten_id(cell_id):\n # Create a hash of the cell ID\n cell_id_bytes = str(cell_id).encode('utf-8')\n hash_object = hashlib.sha256(cell_id_bytes)\n hash_digest = hash_object.digest()\n\n # Encode the hash to a base64 string to mix letters and numbers, truncate to 9 characters\n short_id = base64.urlsafe_b64encode(hash_digest).decode('utf-8')[:9]\n return short_id\n\n # Apply the hash_and_shorten_id function to each cell ID in the specified column\n df['cell_id'] = df['EntityID'].apply(hash_and_shorten_id)\n\n return df\n
convert_to_jpeg(image_path, quality=80)
Convert a TIFF image to a JPEG image with a quality of score
image_path : str Path to the image file quality : int (default=80) Quality score for the JPEG image
new_image_path : str Path to the JPEG image file
def convert_to_jpeg(image_path, quality=80):\n \"\"\"\n Convert a TIFF image to a JPEG image with a quality of score\n\n Parameters\n ----------\n image_path : str\n Path to the image file\n quality : int (default=80)\n Quality score for the JPEG image\n\n Returns\n -------\n new_image_path : str\n Path to the JPEG image file\n\n \"\"\"\n\n # Load the TIFF image\n image = pyvips.Image.new_from_file(image_path, access=\"sequential\")\n\n # Save the image as a JPEG with a quality of 80\n new_image_path = image_path.replace(\".tif\", \".jpeg\")\n image.jpegsave(new_image_path, Q=quality)\n\n return new_image_path\n
convert_to_png(image_path)
def convert_to_png(image_path):\n \"\"\"\n Convert a TIFF image to a JPEG image with a quality of score\n\n Parameters\n ----------\n image_path : str\n Path to the image file\n quality : int (default=80)\n Quality score for the JPEG image\n\n Returns\n -------\n new_image_path : str\n Path to the JPEG image file\n\n \"\"\"\n\n # Load the TIFF image\n image = pyvips.Image.new_from_file(image_path, access=\"sequential\")\n\n # Save the image as a JPEG with a quality of 80\n new_image_path = image_path.replace(\".tif\", \".png\")\n image.pngsave(new_image_path)\n\n return new_image_path\n
convert_to_webp(image_path, quality=100)
Convert a TIFF image to a WEBP image with a specified quality score.
image_path : str Path to the image file quality : int (default=100) Quality score for the WEBP image (higher is better quality)
new_image_path : str Path to the WEBP image file
def convert_to_webp(image_path, quality=100):\n \"\"\"\n Convert a TIFF image to a WEBP image with a specified quality score.\n\n Parameters\n ----------\n image_path : str\n Path to the image file\n quality : int (default=100)\n Quality score for the WEBP image (higher is better quality)\n\n Returns\n -------\n new_image_path : str\n Path to the WEBP image file\n \"\"\"\n # Load the TIFF image\n image = pyvips.Image.new_from_file(image_path, access=\"sequential\")\n\n # Save the image as a WEBP with specified quality\n new_image_path = image_path.replace(\".tif\", \".webp\")\n image.webpsave(new_image_path, Q=quality)\n\n return new_image_path\n
get_max_zoom_level(path_image_pyramid)
Returns the maximum zoom level based on the highest-numbered directory in the specified path_image_pyramid.
path_image_pyramid
str
The path to the directory containing zoom level directories.
Returns:
max_pyramid_zoom
int
The maximum zoom level.
def get_max_zoom_level(path_image_pyramid):\n \"\"\"\n Returns the maximum zoom level based on the highest-numbered directory\n in the specified path_image_pyramid.\n\n Parameters:\n path_image_pyramid (str): The path to the directory containing zoom level directories.\n\n Returns:\n max_pyramid_zoom (int): The maximum zoom level.\n \"\"\"\n # List all entries in the path_image_pyramid that are directories and can be converted to integers\n zoom_levels = [\n entry\n for entry in os.listdir(path_image_pyramid)\n if os.path.isdir(os.path.join(path_image_pyramid, entry)) and entry.isdigit()\n ]\n\n # Convert to integer and find the maximum value\n max_pyramid_zoom = max(map(int, zoom_levels)) if zoom_levels else None\n\n return max_pyramid_zoom\n
make_cell_boundary_tiles(technology, path_cell_boundaries, path_meta_cell_micron, path_transformation_matrix, path_output, coarse_tile_factor=20, tile_size=250, tile_bounds=None, image_scale=1, max_workers=8)
Processes cell boundary data and divides it into spatial tiles based on the provided technology. Reads cell boundary data, applies affine transformations, and divides the data into coarse and fine tiles. The resulting tiles are saved as Parquet files, each containing the geometries of cells in that tile.
technology : str The technology used to generate the cell boundary data, e.g., \"MERSCOPE\", \"Xenium\", or \"custom\". path_cell_boundaries : str Path to the file containing the cell boundaries (Parquet format). path_meta_cell_micron : str Path to the file containing cell metadata (CSV format). path_transformation_matrix : str Path to the file containing the transformation matrix (CSV format). path_output : str Directory path where the output files (Parquet files) for each tile will be saved. coarse_tile_factor : int, optional, default=20. scaling factor of each coarse-grain tile comparing to the fine tile size. tile_size : int, optional, default=500 Size of each fine-grain tile in microns. tile_bounds : dict, optional Dictionary containing the minimum and maximum bounds for x and y coordinates. image_scale : float, optional, default=1 Scale factor to apply to the geometry data. max_workers : int, optional, default=8 Maximum number of parallel workers for processing tiles.
None
src/celldega/pre/boundary_tile.py
def make_cell_boundary_tiles(\n technology,\n path_cell_boundaries,\n path_meta_cell_micron,\n path_transformation_matrix,\n path_output,\n coarse_tile_factor=20,\n tile_size=250,\n tile_bounds=None,\n image_scale=1,\n max_workers=8\n):\n\n\n \"\"\"\n Processes cell boundary data and divides it into spatial tiles based on the provided technology.\n Reads cell boundary data, applies affine transformations, and divides the data into coarse and fine tiles.\n The resulting tiles are saved as Parquet files, each containing the geometries of cells in that tile.\n\n Parameters\n ----------\n technology : str\n The technology used to generate the cell boundary data, e.g., \"MERSCOPE\", \"Xenium\", or \"custom\".\n path_cell_boundaries : str\n Path to the file containing the cell boundaries (Parquet format).\n path_meta_cell_micron : str\n Path to the file containing cell metadata (CSV format).\n path_transformation_matrix : str\n Path to the file containing the transformation matrix (CSV format).\n path_output : str\n Directory path where the output files (Parquet files) for each tile will be saved.\n coarse_tile_factor : int, optional, default=20.\n scaling factor of each coarse-grain tile comparing to the fine tile size.\n tile_size : int, optional, default=500\n Size of each fine-grain tile in microns.\n tile_bounds : dict, optional\n Dictionary containing the minimum and maximum bounds for x and y coordinates.\n image_scale : float, optional, default=1\n Scale factor to apply to the geometry data.\n max_workers : int, optional, default=8\n Maximum number of parallel workers for processing tiles.\n\n Returns\n -------\n None\n \"\"\"\n\n def numpy_affine_transform(coords, matrix):\n \"\"\"Apply affine transformation to numpy coordinates.\"\"\"\n # Homogeneous coordinates for affine transformation\n coords = np.hstack([coords, np.ones((coords.shape[0], 1))])\n transformed_coords = coords @ matrix.T\n return transformed_coords[:, :2] # Drop the homogeneous coordinate\n\n def batch_transform_geometries(geometries, transformation_matrix, scale):\n \"\"\"\n Batch transform geometries using numpy for optimized performance.\n \"\"\"\n # Extract affine transformation parameters into a 3x3 matrix for numpy\n affine_matrix = np.array([\n [transformation_matrix[0, 0], transformation_matrix[0, 1], transformation_matrix[0, 2]],\n [transformation_matrix[1, 0], transformation_matrix[1, 1], transformation_matrix[1, 2]],\n [0, 0, 1]\n ])\n\n transformed_geometries = []\n\n for polygon in geometries:\n # Extract coordinates and transform them\n if isinstance(polygon, MultiPolygon):\n polygon = next(polygon.geoms) # Use the first geometry\n\n # Transform the exterior of the polygon\n exterior_coords = np.array(polygon.exterior.coords)\n\n # Apply the affine transformation and scale\n transformed_coords = numpy_affine_transform(exterior_coords, affine_matrix) / scale\n\n # Append the result to the transformed_geometries list\n transformed_geometries.append([transformed_coords.tolist()])\n\n return transformed_geometries\n\n\n def filter_and_save_fine_boundary(coarse_tile, fine_i, fine_j, fine_tile_x_min, fine_tile_x_max, fine_tile_y_min, fine_tile_y_max, path_output):\n cell_ids = coarse_tile.index.values\n\n tile_filter = (\n (coarse_tile[\"center_x\"] >= fine_tile_x_min) & (coarse_tile[\"center_x\"] < fine_tile_x_max) &\n (coarse_tile[\"center_y\"] >= fine_tile_y_min) & (coarse_tile[\"center_y\"] < fine_tile_y_max)\n )\n filtered_indices = np.where(tile_filter)[0]\n\n keep_cells = cell_ids[filtered_indices]\n fine_tile_cells = coarse_tile.loc[keep_cells, [\"GEOMETRY\"]]\n fine_tile_cells = fine_tile_cells.assign(name=fine_tile_cells.index)\n\n if not fine_tile_cells.empty:\n filename = f\"{path_output}/cell_tile_{fine_i}_{fine_j}.parquet\"\n fine_tile_cells.to_parquet(filename)\n\n def process_fine_boundaries(coarse_tile, i, j, coarse_tile_x_min, coarse_tile_x_max, coarse_tile_y_min, coarse_tile_y_max, tile_size, path_output, x_min, y_min, n_fine_tiles_x, n_fine_tiles_y):\n with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:\n futures = []\n for fine_i in range(n_fine_tiles_x):\n fine_tile_x_min = x_min + fine_i * tile_size\n fine_tile_x_max = fine_tile_x_min + tile_size\n\n if not (fine_tile_x_min >= coarse_tile_x_min and fine_tile_x_max <= coarse_tile_x_max):\n continue\n\n for fine_j in range(n_fine_tiles_y):\n fine_tile_y_min = y_min + fine_j * tile_size\n fine_tile_y_max = fine_tile_y_min + tile_size\n\n if not (fine_tile_y_min >= coarse_tile_y_min and fine_tile_y_max <= coarse_tile_y_max):\n continue\n\n futures.append(executor.submit(\n filter_and_save_fine_boundary, coarse_tile, fine_i, fine_j, fine_tile_x_min, fine_tile_x_max, fine_tile_y_min, fine_tile_y_max, path_output\n ))\n\n for future in futures:\n future.result()\n\n tile_size_x = tile_size\n tile_size_y = tile_size\n\n transformation_matrix = pd.read_csv(path_transformation_matrix, header=None, sep=\" \").values\n\n # Load cell boundary data based on the technology\n if technology == \"MERSCOPE\":\n df_meta = pd.read_parquet(f\"{path_output.replace('cell_segmentation','cell_metadata.parquet')}\")\n entity_to_cell_id_dict = pd.Series(df_meta.index.values, index=df_meta.EntityID).to_dict()\n cells_orig = gpd.read_parquet(path_cell_boundaries)\n cells_orig['cell_id'] = cells_orig['EntityID'].map(entity_to_cell_id_dict)\n cells_orig = cells_orig[cells_orig[\"ZIndex\"] == 1]\n\n # Correct cell_id issues with meta_cell\n meta_cell = pd.read_csv(path_meta_cell_micron)\n meta_cell['cell_id'] = meta_cell['EntityID'].map(entity_to_cell_id_dict)\n cells_orig.index = meta_cell[meta_cell[\"cell_id\"].isin(cells_orig['cell_id'])].index\n\n # Correct 'MultiPolygon' to 'Polygon'\n cells_orig[\"geometry\"] = cells_orig[\"Geometry\"].apply(\n lambda x: list(x.geoms)[0] if isinstance(x, MultiPolygon) else x\n )\n\n cells_orig.set_index('cell_id', inplace=True)\n\n elif technology == \"Xenium\":\n xenium_cells = pd.read_parquet(path_cell_boundaries)\n grouped = xenium_cells.groupby(\"cell_id\")[[\"vertex_x\", \"vertex_y\"]].agg(lambda x: x.tolist())\n grouped[\"geometry\"] = grouped.apply(lambda row: Polygon(zip(row[\"vertex_x\"], row[\"vertex_y\"])), axis=1)\n cells_orig = gpd.GeoDataFrame(grouped, geometry=\"geometry\")[[\"geometry\"]]\n\n elif technology == \"custom\":\n cells_orig = gpd.read_parquet(path_cell_boundaries)\n\n # Transform geometries\n cells_orig[\"GEOMETRY\"] = batch_transform_geometries(cells_orig[\"geometry\"], transformation_matrix, image_scale)\n\n # Convert transformed geometries to polygons and calculate centroids\n cells_orig[\"polygon\"] = cells_orig[\"GEOMETRY\"].apply(lambda x: Polygon(x[0]))\n gdf_cells = gpd.GeoDataFrame(geometry=cells_orig[\"polygon\"])\n gdf_cells[\"center_x\"] = gdf_cells.geometry.centroid.x\n gdf_cells[\"center_y\"] = gdf_cells.geometry.centroid.y\n gdf_cells[\"GEOMETRY\"] = cells_orig[\"GEOMETRY\"]\n\n # Ensure the output directory exists\n if not os.path.exists(path_output):\n os.makedirs(path_output)\n\n # Calculate tile bounds and fine/coarse tiles\n x_min, x_max = tile_bounds[\"x_min\"], tile_bounds[\"x_max\"]\n y_min, y_max = tile_bounds[\"y_min\"], tile_bounds[\"y_max\"]\n n_fine_tiles_x = int(np.ceil((x_max - x_min) / tile_size))\n n_fine_tiles_y = int(np.ceil((y_max - y_min) / tile_size))\n n_coarse_tiles_x = int(np.ceil((x_max - x_min) / (coarse_tile_factor * tile_size)))\n n_coarse_tiles_y = int(np.ceil((y_max - y_min) / (coarse_tile_factor * tile_size)))\n\n # Process coarse tiles in parallel\n for i in tqdm(range(n_coarse_tiles_x), desc=\"Processing coarse tiles\"):\n coarse_tile_x_min = x_min + i * (coarse_tile_factor * tile_size)\n coarse_tile_x_max = coarse_tile_x_min + (coarse_tile_factor * tile_size)\n\n for j in range(n_coarse_tiles_y):\n coarse_tile_y_min = y_min + j * (coarse_tile_factor * tile_size)\n coarse_tile_y_max = coarse_tile_y_min + (coarse_tile_factor * tile_size)\n\n coarse_tile = gdf_cells[\n (gdf_cells[\"center_x\"] >= coarse_tile_x_min) & (gdf_cells[\"center_x\"] < coarse_tile_x_max) &\n (gdf_cells[\"center_y\"] >= coarse_tile_y_min) & (gdf_cells[\"center_y\"] < coarse_tile_y_max)\n ]\n if not coarse_tile.empty:\n process_fine_boundaries(coarse_tile, i, j, coarse_tile_x_min, coarse_tile_x_max, coarse_tile_y_min, coarse_tile_y_max, tile_size, path_output, x_min, y_min, n_fine_tiles_x, n_fine_tiles_y)\n
make_deepzoom_pyramid(image_path, output_path, pyramid_name, tile_size=512, overlap=0, suffix='.jpeg')
Create a DeepZoom image pyramid from a JPEG image
image_path : str Path to the JPEG image file tile_size : int (default=512) Tile size for the DeepZoom pyramid overlap : int (default=0) Overlap size for the DeepZoom pyramid suffix : str (default='jpeg') Suffix for the DeepZoom pyramid tiles
def make_deepzoom_pyramid(\n image_path, output_path, pyramid_name, tile_size=512, overlap=0, suffix=\".jpeg\"\n):\n \"\"\"\n Create a DeepZoom image pyramid from a JPEG image\n\n Parameters\n ----------\n image_path : str\n Path to the JPEG image file\n tile_size : int (default=512)\n Tile size for the DeepZoom pyramid\n overlap : int (default=0)\n Overlap size for the DeepZoom pyramid\n suffix : str (default='jpeg')\n Suffix for the DeepZoom pyramid tiles\n\n Returns\n -------\n None\n\n \"\"\"\n\n # Define the output path\n output_path = Path(output_path)\n\n # Load the JPEG image\n image = pyvips.Image.new_from_file(image_path, access=\"sequential\")\n\n # check if the output path exists and create it if it does not\n output_path.mkdir(parents=True, exist_ok=True)\n\n # append the pyramid name to the output path\n output_path = output_path / pyramid_name\n\n # Save the image as a DeepZoom image pyramid\n image.dzsave(output_path, tile_size=tile_size, overlap=overlap, suffix=suffix)\n
make_meta_cell_image_coord(technology, path_transformation_matrix, path_meta_cell_micron, path_meta_cell_image, image_scale)
Apply an affine transformation to the cell coordinates in microns and save the transformed coordinates in pixels
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
make_meta_cell_image_coord( ... technology='Xenium', ... path_transformation_matrix='data/transformation_matrix.txt', ... path_meta_cell_micron='data/meta_cell_micron.csv', ... path_meta_cell_image='data/meta_cell_image.parquet' ... )
def make_meta_cell_image_coord(\n technology,\n path_transformation_matrix,\n path_meta_cell_micron,\n path_meta_cell_image,\n image_scale\n):\n \"\"\"\n Apply an affine transformation to the cell coordinates in microns and save\n the transformed coordinates in pixels\n\n Parameters\n ----------\n technology : str\n The technology used to generate the data, Xenium and MERSCOPE are supported.\n path_transformation_matrix : str\n Path to the transformation matrix file\n path_meta_cell_micron : str\n Path to the meta cell file with coordinates in microns\n path_meta_cell_image : str\n Path to save the meta cell file with coordinates in pixels\n\n Returns\n -------\n None\n\n Examples\n --------\n >>> make_meta_cell_image_coord(\n ... technology='Xenium',\n ... path_transformation_matrix='data/transformation_matrix.txt',\n ... path_meta_cell_micron='data/meta_cell_micron.csv',\n ... path_meta_cell_image='data/meta_cell_image.parquet'\n ... )\n\n \"\"\"\n\n transformation_matrix = pd.read_csv(\n path_transformation_matrix, header=None, sep=\" \"\n ).values\n\n if technology == \"MERSCOPE\":\n meta_cell = pd.read_csv(path_meta_cell_micron, usecols=[\"EntityID\", \"center_x\", \"center_y\"])\n meta_cell = convert_long_id_to_short(meta_cell)\n meta_cell[\"name\"] = meta_cell[\"cell_id\"]\n meta_cell = meta_cell.set_index('cell_id')\n elif technology == \"Xenium\":\n usecols = [\"cell_id\", \"x_centroid\", \"y_centroid\"]\n meta_cell = pd.read_csv(path_meta_cell_micron, index_col=0, usecols=usecols)\n meta_cell.columns = [\"center_x\", \"center_y\"]\n meta_cell[\"name\"] = pd.Series(meta_cell.index, index=meta_cell.index)\n\n # Adding a ones column to accommodate for affine transformation\n meta_cell[\"ones\"] = 1\n\n # Preparing the data for matrix multiplication\n points = meta_cell[[\"center_x\", \"center_y\", \"ones\"]].values\n\n # Applying the transformation matrix\n transformed_points = np.dot(transformation_matrix, points.T).T\n\n # Updating the DataFrame with transformed coordinates\n meta_cell[\"center_x\"] = transformed_points[:, 0]\n meta_cell[\"center_y\"] = transformed_points[:, 1]\n\n # Dropping the ones column as it's no longer needed\n meta_cell.drop(columns=[\"ones\"], inplace=True)\n\n meta_cell[\"center_x\"] = meta_cell[\"center_x\"] / image_scale\n meta_cell[\"center_y\"] = meta_cell[\"center_y\"] / image_scale\n\n meta_cell[\"geometry\"] = meta_cell.apply(\n lambda row: [row[\"center_x\"], row[\"center_y\"]], axis=1\n )\n\n if technology == \"MERSCOPE\":\n meta_cell = meta_cell[[\"name\", \"geometry\", \"EntityID\"]]\n else:\n meta_cell = meta_cell[[\"name\", \"geometry\"]]\n\n\n meta_cell.to_parquet(path_meta_cell_image)\n
make_meta_gene(technology, path_cbg, path_output)
Create a DataFrame with genes and their assigned colors
technology : str The technology used to generate the data, Xenium and MERSCOPE are supported. path_cbg : str Path to the cell-by-gene matrix data (the data format can vary based on technology) path_output : str Path to save the meta gene file
make_meta_gene( ... technology='Xenium', ... path_cbg='data/', ... path_output='data/meta_gene.parquet' ... )
def make_meta_gene(technology, path_cbg, path_output):\n \"\"\"\n Create a DataFrame with genes and their assigned colors\n\n Parameters\n ----------\n technology : str\n The technology used to generate the data, Xenium and MERSCOPE are supported.\n path_cbg : str\n Path to the cell-by-gene matrix data (the data format can vary based on technology)\n path_output : str\n Path to save the meta gene file\n\n Returns\n -------\n None\n\n Examples\n --------\n >>> make_meta_gene(\n ... technology='Xenium',\n ... path_cbg='data/',\n ... path_output='data/meta_gene.parquet'\n ... )\n \"\"\"\n\n if technology == \"MERSCOPE\":\n cbg = pd.read_csv(path_cbg, index_col=0)\n genes = cbg.columns.tolist()\n elif technology == \"Xenium\":\n # genes = pd.read_csv(path_cbg + 'features.tsv.gz', sep='\\t', header=None)[1].values.tolist()\n cbg = read_cbg_mtx(path_cbg)\n genes = cbg.columns.tolist()\n\n # Get all categorical color palettes from Matplotlib and flatten them into a single list of colors\n palettes = [plt.get_cmap(name).colors for name in plt.colormaps() if \"tab\" in name]\n flat_colors = [color for palette in palettes for color in palette]\n\n # Convert RGB tuples to hex codes\n flat_colors_hex = [to_hex(color) for color in flat_colors]\n\n # Use modular arithmetic to assign a color to each gene, white for genes with \"Blank\"\n colors = [\n flat_colors_hex[i % len(flat_colors_hex)] if \"Blank\" not in gene else \"#FFFFFF\"\n for i, gene in enumerate(genes)\n ]\n\n # Create a DataFrame with genes and their assigned colors\n ser_color = pd.Series(colors, index=genes)\n\n # calculate gene expression metadata\n meta_gene = calc_meta_gene_data(cbg)\n meta_gene['color'] = ser_color\n\n # Identify sparse columns\n sparse_cols = [col for col in meta_gene.columns if pd.api.types.is_sparse(meta_gene[col])]\n\n # Convert sparse columns to dense\n for col in sparse_cols:\n meta_gene[col] = meta_gene[col].sparse.to_dense()\n\n meta_gene.to_parquet(path_output)\n
make_trx_tiles(technology, path_trx, path_transformation_matrix, path_trx_tiles, coarse_tile_factor=10, tile_size=250, chunk_size=1000000, verbose=False, image_scale=1, max_workers=8)
Processes transcript data by dividing it into coarse-grain and fine-grain tiles, applying transformations, and saving the results in a parallelized manner.
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 8).
dict A dictionary containing the bounds of the processed data in both x and y directions.
src/celldega/pre/trx_tile.py
def make_trx_tiles(\n technology,\n path_trx,\n path_transformation_matrix,\n path_trx_tiles,\n coarse_tile_factor=10,\n tile_size=250,\n chunk_size=1000000,\n verbose=False,\n image_scale=1,\n max_workers=8\n):\n \"\"\"\n Processes transcript data by dividing it into coarse-grain and fine-grain tiles,\n applying transformations, and saving the results in a parallelized manner.\n\n Parameters\n ----------\n technology : str\n The technology used for generating the transcript data (e.g., \"MERSCOPE\" or \"Xenium\").\n path_trx : str\n Path to the file containing the transcript data.\n path_transformation_matrix : str\n Path to the file containing the transformation matrix (CSV file).\n path_trx_tiles : str\n Directory path where the output files (Parquet files) for each tile will be saved.\n coarse_tile_factor : int, optional\n Scaling factor of each coarse-grain tile comparing to the fine tile size.\n tile_size : int, optional\n Size of each fine-grain tile in microns (default is 250).\n chunk_size : int, optional\n Number of rows to process per chunk for memory efficiency (default is 1000000).\n verbose : bool, optional\n Flag to enable verbose output (default is False).\n image_scale : float, optional\n Scale factor to apply to the transcript coordinates (default is 0.5).\n max_workers : int, optional\n Maximum number of parallel workers for processing tiles (default is 8).\n\n Returns\n -------\n dict\n A dictionary containing the bounds of the processed data in both x and y directions.\n \"\"\"\n\n def process_coarse_tile(trx, i, j, coarse_tile_x_min, coarse_tile_x_max, coarse_tile_y_min, coarse_tile_y_max, tile_size, path_trx_tiles, x_min, y_min, n_fine_tiles_x, n_fine_tiles_y, max_workers):\n # Filter the entire dataset for the current coarse tile\n coarse_tile = trx.filter(\n (pl.col(\"transformed_x\") >= coarse_tile_x_min) & (pl.col(\"transformed_x\") < coarse_tile_x_max) &\n (pl.col(\"transformed_y\") >= coarse_tile_y_min) & (pl.col(\"transformed_y\") < coarse_tile_y_max)\n )\n\n if not coarse_tile.is_empty():\n # Now process fine tiles using global fine tile indices\n process_fine_tiles(coarse_tile, i, j, coarse_tile_x_min, coarse_tile_x_max, coarse_tile_y_min, coarse_tile_y_max, tile_size, path_trx_tiles, x_min, y_min, n_fine_tiles_x, n_fine_tiles_y, max_workers) \n\n\n def process_fine_tiles(coarse_tile, coarse_i, coarse_j, coarse_tile_x_min, coarse_tile_x_max, coarse_tile_y_min, coarse_tile_y_max, tile_size, path_trx_tiles, x_min, y_min, n_fine_tiles_x, n_fine_tiles_y, max_workers=8):\n\n # Use ThreadPoolExecutor for parallel processing of fine-grain tiles within the coarse tile\n with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:\n futures = []\n\n # Iterate over fine-grain tiles within the global bounds\n for fine_i in range(n_fine_tiles_x):\n fine_tile_x_min = x_min + fine_i * tile_size\n fine_tile_x_max = fine_tile_x_min + tile_size\n\n # Process only if the fine tile falls within the current coarse tile's bounds\n if not (fine_tile_x_min >= coarse_tile_x_min and fine_tile_x_max <= coarse_tile_x_max):\n continue\n\n for fine_j in range(n_fine_tiles_y):\n fine_tile_y_min = y_min + fine_j * tile_size\n fine_tile_y_max = fine_tile_y_min + tile_size\n\n # Process only if the fine tile falls within the current coarse tile's bounds\n if not (fine_tile_y_min >= coarse_tile_y_min and fine_tile_y_max <= coarse_tile_y_max):\n continue\n\n # Submit the task for each fine tile to process in parallel\n futures.append(executor.submit(\n filter_and_save_fine_tile, coarse_tile, coarse_i, coarse_j, fine_i, fine_j, \n fine_tile_x_min, fine_tile_x_max, fine_tile_y_min, fine_tile_y_max, path_trx_tiles\n ))\n\n # Wait for all futures to complete\n for future in concurrent.futures.as_completed(futures):\n future.result() # Raise exceptions if any occurred during execution\n\n\n def filter_and_save_fine_tile(coarse_tile, coarse_i, coarse_j, fine_i, fine_j, fine_tile_x_min, fine_tile_x_max, fine_tile_y_min, fine_tile_y_max, path_trx_tiles):\n\n # Filter the coarse tile for the current fine tile's boundaries\n fine_tile_trx = coarse_tile.filter(\n (pl.col(\"transformed_x\") >= fine_tile_x_min) & (pl.col(\"transformed_x\") < fine_tile_x_max) &\n (pl.col(\"transformed_y\") >= fine_tile_y_min) & (pl.col(\"transformed_y\") < fine_tile_y_max)\n )\n\n if not fine_tile_trx.is_empty():\n # Add geometry column as a list of [x, y] pairs\n fine_tile_trx = fine_tile_trx.with_columns(\n pl.concat_list([pl.col(\"transformed_x\"), pl.col(\"transformed_y\")]).alias(\"geometry\")\n ).drop(['transformed_x', 'transformed_y'])\n\n # Define the filename based on fine tile coordinates\n filename = f\"{path_trx_tiles}/transcripts_tile_{fine_i}_{fine_j}.parquet\"\n\n # Save the filtered DataFrame to a Parquet file\n fine_tile_trx.to_pandas().to_parquet(filename)\n\n\n # Load transformation matrix\n transformation_matrix = np.loadtxt(path_transformation_matrix)\n\n # Load the transcript data based on the technology using Polars\n if technology == \"MERSCOPE\":\n trx_ini = pl.read_csv(path_trx, columns=[\"gene\", \"global_x\", \"global_y\"])\n trx_ini = trx_ini.with_columns([\n pl.col(\"global_x\").alias(\"x\"),\n pl.col(\"global_y\").alias(\"y\"),\n pl.col(\"gene\").alias(\"name\")\n ]).select([\"name\", \"x\", \"y\"])\n\n elif technology == \"Xenium\":\n trx_ini = pl.read_parquet(path_trx).select([\n pl.col(\"feature_name\").alias(\"name\"),\n pl.col(\"x_location\").alias(\"x\"),\n pl.col(\"y_location\").alias(\"y\")\n ])\n\n # Process the data in chunks and apply transformations\n all_chunks = []\n\n for start_row in tqdm(range(0, trx_ini.height, chunk_size), desc=\"Processing chunks\"):\n chunk = trx_ini.slice(start_row, chunk_size)\n\n # Apply transformation matrix to the coordinates\n points = np.hstack([chunk.select([\"x\", \"y\"]).to_numpy(), np.ones((chunk.height, 1))])\n transformed_points = np.dot(points, transformation_matrix.T)[:, :2]\n\n # Create new transformed columns and drop original x, y columns\n transformed_chunk = chunk.with_columns([\n (pl.Series(transformed_points[:, 0]) * image_scale).round(2).alias(\"transformed_x\"),\n (pl.Series(transformed_points[:, 1]) * image_scale).round(2).alias(\"transformed_y\")\n ]).drop([\"x\", \"y\"])\n all_chunks.append(transformed_chunk)\n\n # Concatenate all chunks after processing\n trx = pl.concat(all_chunks)\n\n # Ensure the output directory exists\n if not os.path.exists(path_trx_tiles):\n os.makedirs(path_trx_tiles)\n\n # Get min and max x, y values\n x_min, x_max = trx.select([\n pl.col(\"transformed_x\").min().alias(\"x_min\"),\n pl.col(\"transformed_x\").max().alias(\"x_max\")\n ]).row(0)\n\n y_min, y_max = trx.select([\n pl.col(\"transformed_y\").min().alias(\"y_min\"),\n pl.col(\"transformed_y\").max().alias(\"y_max\")\n ]).row(0)\n\n # Calculate the number of fine-grain tiles globally\n n_fine_tiles_x = int(np.ceil((x_max - x_min) / tile_size))\n n_fine_tiles_y = int(np.ceil((y_max - y_min) / tile_size))\n\n # Calculate the number of coarse-grain tiles\n n_coarse_tiles_x = int(np.ceil((x_max - x_min) / (coarse_tile_factor * tile_size)))\n n_coarse_tiles_y = int(np.ceil((y_max - y_min) / (coarse_tile_factor * tile_size)))\n\n # Use ThreadPoolExecutor for parallel processing of coarse-grain tiles\n with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:\n futures = []\n for i in range(n_coarse_tiles_x):\n coarse_tile_x_min = x_min + i * (coarse_tile_factor * tile_size)\n coarse_tile_x_max = coarse_tile_x_min + (coarse_tile_factor * tile_size)\n\n for j in range(n_coarse_tiles_y):\n coarse_tile_y_min = y_min + j * (coarse_tile_factor * tile_size)\n coarse_tile_y_max = coarse_tile_y_min + (coarse_tile_factor * tile_size)\n\n # Submit each coarse tile for parallel processing\n futures.append(executor.submit(\n process_coarse_tile, trx, i, j, coarse_tile_x_min, coarse_tile_x_max, coarse_tile_y_min, coarse_tile_y_max, tile_size, path_trx_tiles, x_min, y_min, n_fine_tiles_x, n_fine_tiles_y, max_workers\n ))\n\n # Wait for all coarse tiles to complete\n for future in tqdm(concurrent.futures.as_completed(futures), desc=\"Processing coarse tiles\", unit=\"tile\"):\n future.result() # Raise exceptions if any occurred during execution\n\n # Return the tile bounds\n tile_bounds = {\n \"x_min\": x_min,\n \"x_max\": x_max,\n \"y_min\": y_min,\n \"y_max\": y_max,\n }\n\n return tile_bounds\n
reduce_image_size(image_path, scale_image=0.5, path_landscape_files='')
image_path : str Path to the image file scale_image : float (default=0.5) Scale factor for the image resize
new_image_path : str Path to the resized image file
def reduce_image_size(image_path, scale_image=0.5, path_landscape_files=\"\"):\n \"\"\"\n\n Parameters\n ----------\n image_path : str\n Path to the image file\n scale_image : float (default=0.5)\n Scale factor for the image resize\n\n Returns\n -------\n new_image_path : str\n Path to the resized image file\n \"\"\"\n\n image = pyvips.Image.new_from_file(image_path, access=\"sequential\")\n\n resized_image = image.resize(scale_image)\n\n new_image_name = image_path.split(\"/\")[-1].replace(\".tif\", \"_downsize.tif\")\n new_image_path = f\"{path_landscape_files}/{new_image_name}\"\n resized_image.write_to_file(new_image_path)\n\n return new_image_path\n
save_landscape_parameters(technology, path_landscape_files, image_name='dapi_files', tile_size=1000, image_info={}, image_format='.webp')
Save the landscape parameters to a JSON file.
def save_landscape_parameters(\n technology, path_landscape_files, image_name=\"dapi_files\", tile_size=1000, image_info={}, image_format='.webp'\n):\n \"\"\"\n Save the landscape parameters to a JSON file.\n \"\"\"\n\n path_image_pyramid = f\"{path_landscape_files}/pyramid_images/{image_name}\"\n\n print(path_image_pyramid)\n\n max_pyramid_zoom = get_max_zoom_level(path_image_pyramid)\n\n landscape_parameters = {\n \"technology\": technology,\n \"max_pyramid_zoom\": max_pyramid_zoom,\n \"tile_size\": tile_size,\n \"image_info\": image_info,\n \"image_format\": image_format\n }\n\n path_landscape_parameters = f\"{path_landscape_files}/landscape_parameters.json\"\n\n with open(path_landscape_parameters, \"w\") as file:\n json.dump(landscape_parameters, file, indent=4)\n
Landscape
Bases: AnyWidget
AnyWidget
A widget for interactive visualization of spatial omics data. This widget currently supports iST (Xenium and MERSCOPE) and sST (Visium HD data)
float
The initial x-coordinate of the view.
The initial y-coordinate of the view.
The initial zoom level of the view.
The token traitlet.
The base URL for the widget.
The name of the dataset to visualize. This will show up in the user interface bar.
Attributes:
component
The name of the component.
technology
The technology used.
The initial z-coordinate of the view.
The name of the dataset to visualize.
update_trigger
dict
The dictionary to trigger updates.
cell_clusters
The dictionary containing cell cluster information.
A widget for visualizing a 'landscape' view of spatial omics data.
src/celldega/viz/widget.py
class Landscape(anywidget.AnyWidget):\n \"\"\"\n A widget for interactive visualization of spatial omics data. This widget\n currently supports iST (Xenium and MERSCOPE) and sST (Visium HD data)\n\n Args:\n ini_x (float): The initial x-coordinate of the view.\n ini_y (float): The initial y-coordinate of the view.\n ini_zoom (float): The initial zoom level of the view.\n token (str): The token traitlet.\n base_url (str): The base URL for the widget.\n dataset_name (str, optional): The name of the dataset to visualize. This will show up in the user interface bar.\n\n Attributes:\n component (str): The name of the component.\n technology (str): The technology used.\n base_url (str): The base URL for the widget.\n token (str): The token traitlet.\n ini_x (float): The initial x-coordinate of the view.\n ini_y (float): The initial y-coordinate of the view.\n ini_z (float): The initial z-coordinate of the view.\n ini_zoom (float): The initial zoom level of the view.\n dataset_name (str): The name of the dataset to visualize.\n update_trigger (dict): The dictionary to trigger updates.\n cell_clusters (dict): The dictionary containing cell cluster information.\n\n Returns:\n Landscape: A widget for visualizing a 'landscape' view of spatial omics data.\n \"\"\"\n _esm = pathlib.Path(__file__).parent / \"../static\" / \"widget.js\"\n _css = pathlib.Path(__file__).parent / \"../static\" / \"widget.css\"\n component = traitlets.Unicode(\"Landscape\").tag(sync=True)\n\n technology = traitlets.Unicode(\"sst\").tag(sync=True)\n base_url = traitlets.Unicode(\"\").tag(sync=True)\n token = traitlets.Unicode(\"\").tag(sync=True)\n ini_x = traitlets.Float(1000).tag(sync=True)\n ini_y = traitlets.Float(1000).tag(sync=True)\n ini_z = traitlets.Float(0).tag(sync=True)\n ini_zoom = traitlets.Float(0).tag(sync=True)\n square_tile_size = traitlets.Float(1.4).tag(sync=True)\n dataset_name = traitlets.Unicode(\"\").tag(sync=True)\n region = traitlets.Dict({}).tag(sync=True)\n\n update_trigger = traitlets.Dict().tag(sync=True)\n cell_clusters = traitlets.Dict().tag(sync=True)\n\n width = traitlets.Int(0).tag(sync=True)\n height = traitlets.Int(800).tag(sync=True)\n\n def trigger_update(self, new_value):\n # This method updates the update_trigger traitlet with a new value\n # You can pass any information necessary for the update, or just a timestamp\n self.update_trigger = new_value\n\n def update_cell_clusters(self, new_clusters):\n # Convert the new_clusters to a JSON serializable format if necessary\n self.cell_clusters = new_clusters\n
Matrix
A widget for interactive visualization of a hierarchically clustered matrix.
value
The value traitlet.
The component traitlet.
The network traitlet.
click_info
The click_info traitlet.
The network dictionary.
The click_info dictionary.
A widget for visualizing a hierarchically clustered matrix.
class Matrix(anywidget.AnyWidget):\n \"\"\"\n A widget for interactive visualization of a hierarchically clustered matrix.\n\n Args:\n value (int): The value traitlet.\n component (str): The component traitlet.\n network (dict): The network traitlet.\n click_info (dict): The click_info traitlet.\n\n Attributes:\n component (str): The name of the component.\n network (dict): The network dictionary.\n click_info (dict): The click_info dictionary.\n\n Returns:\n Matrix: A widget for visualizing a hierarchically clustered matrix.\n \"\"\"\n _esm = pathlib.Path(__file__).parent / \"../static\" / \"widget.js\"\n _css = pathlib.Path(__file__).parent / \"../static\" / \"widget.css\"\n value = traitlets.Int(0).tag(sync=True)\n component = traitlets.Unicode(\"Matrix\").tag(sync=True)\n\n network = traitlets.Dict({}).tag(sync=True)\n click_info = traitlets.Dict({}).tag(sync=True)\n