dsa

Build DSA annotation json from stardist geojson classification results

Parameters:

Name	Type	Description	Default
slide_manifest	DataFrame[SlideSchema]	slide manifest from slide_etl	required
output_urlpath	string	URL/path prefix to save annotations	required
annotation_name	string	name of the annotation to be displayed in DSA	required
line_colors	dict	user-provided line color map with {feature name:rgb values}	None
fill_colors	dict	user-provided fill color map with {feature name:rgba values}	None
storage_options	dict	storage options to pass to read functions	{}
output_storage_options	dict	storage options to pass to write functions	{}
annotation_column	string	column containing url to stardist polygon geojson	''
output_column	string	column with result url to add to slide_manifest	''

Returns:

Type	Description
	DataFrame[SlideSchema]: slide manifest

Source code in src/luna/pathology/cli/dsa_viz.py

def stardist_polygon(
    slide_manifest: DataFrame[SlideSchema],
    output_urlpath: str,
    annotation_name: str,
    line_colors: Optional[Dict[str, str]] = None,
    fill_colors: Optional[Dict[str, str]] = None,
    storage_options: Dict = {},
    output_storage_options: Dict = {},
    annotation_column: str = "",
    output_column: str = "",
):
    """Build DSA annotation json from stardist geojson classification results

    Args:
        slide_manifest (DataFrame[SlideSchema]): slide manifest from slide_etl
        output_urlpath (string): URL/path prefix to save annotations
        annotation_name (string): name of the annotation to be displayed in DSA
        line_colors (dict): user-provided line color map with {feature name:rgb values}
        fill_colors (dict): user-provided fill color map with {feature name:rgba values}
        storage_options (dict): storage options to pass to read functions
        output_storage_options (dict): storage options to pass to write functions
        annotation_column (string): column containing url to stardist polygon geojson
        output_column (string): column with result url to add to slide_manifest

    Returns:
        DataFrame[SlideSchema]: slide manifest
    """
    if not annotation_column:
        annotation_column = f"{annotation_name}_geojson_url"
    if not output_column:
        output_column = f"{annotation_name}_dsa_url"

    if annotation_column not in slide_manifest.columns:
        raise ValueError(f"{annotation_column} not found in slide manifest")
    client = get_or_create_dask_client()
    futures = []
    for _, row in slide_manifest.iterrows():
        image_filename = os.path.basename(row["url"])
        future = client.submit(
            __stardist_polygon,
            row[annotation_column],
            output_urlpath,
            image_filename,
            annotation_name,
            line_colors,
            fill_colors,
            storage_options,
            output_storage_options,
        )

        futures.append(future)
    progress(futures)
    dsa_annotation_urls = client.gather(futures)
    for idx, dsa_annotation_url in enumerate(dsa_annotation_urls):
        slide_manifest.at[idx, output_column] = dsa_annotation_url

    return slide_manifest

Build DSA annotation json from TSV classification data generated by stardist

Processes a cell classification data generated by Qupath/stardist and adds the center coordinates of the cells as annotation elements.

Parameters:

Name	Type	Description	Default
input_urlpath	string	URL/path to TSV classification data generated by stardist	required
output_urlpath	string	URL/path prefix for saving dsa annotation json	required
annotation_name	string	name of the annotation to be displayed in DSA	required
line_colors	dict	line color map with {feature name:rgb values}	None
fill_colors	dict	fill color map with {feature name:rgba values}	None
storage_options	dict	storage options to pass to read functions	{}
output_storage_options	dict	storage options to pass to write functions	{}
annotation_column	string	column containing url to stardist polygon geojson	''
output_column_suffix	string	column suffix with result url to add to slide_manifest	required

Returns:

Type	Description
	DataFrame[SlideSchema]: slide manifest

Source code in src/luna/pathology/cli/dsa_viz.py

def stardist_cell(
    slide_manifest: DataFrame[SlideSchema],
    output_urlpath: str,
    annotation_name: str,
    line_colors: Optional[Dict[str, str]] = None,
    fill_colors: Optional[Dict[str, str]] = None,
    storage_options: Dict = {},
    output_storage_options: Dict = {},
    annotation_column: str = "",
    output_column: str = "",
):
    """Build DSA annotation json from TSV classification data generated by
    stardist

    Processes a cell classification data generated by Qupath/stardist and
    adds the center coordinates of the cells
    as annotation elements.

    Args:
        input_urlpath (string): URL/path to TSV classification data generated by stardist
        output_urlpath (string): URL/path prefix for saving dsa annotation json
        annotation_name (string): name of the annotation to be displayed in DSA
        line_colors (dict, optional): line color map with {feature name:rgb values}
        fill_colors (dict, optional): fill color map with {feature name:rgba values}
        storage_options (dict): storage options to pass to read functions
        output_storage_options (dict): storage options to pass to write functions
        annotation_column (string): column containing url to stardist polygon geojson
        output_column_suffix (string): column suffix with result url to add to slide_manifest

    Returns:
        DataFrame[SlideSchema]: slide manifest
    """
    if not annotation_column:
        annotation_column = f"{annotation_name}_tsv_url"
    if not output_column:
        output_column = f"{annotation_name}_dsa_url"
    if annotation_column not in slide_manifest.columns:
        raise ValueError(f"{annotation_column} not found in slide manifest")
    client = get_or_create_dask_client()
    futures = []
    for _, row in slide_manifest.iterrows():
        image_filename = os.path.basename(row["url"])
        future = client.submit(
            __stardist_cell,
            row[annotation_column],
            output_urlpath,
            image_filename,
            annotation_name,
            line_colors,
            fill_colors,
            storage_options,
            output_storage_options,
        )

        futures.append(future)
    progress(futures)
    dsa_annotation_urls = client.gather(futures)
    return slide_manifest.assign(**{output_column: dsa_annotation_urls})

Build DSA annotation json from regional annotation geojson

Parameters:

Name	Type	Description	Default
slide_manifest	DataFrame[SlideSchema]	slide manifest	required
output_urlpath	string	URL/path prefix for saving dsa annotation json	required
annotation_name	string	name of the annotation to be displayed in DSA	required
line_colors	dict	line color map with {feature name:rgb values}	None
fill_colors	dict	fill color map with {feature name:rgba values}	None
storage_options	dict	storage options to pass to read functions	{}
output_storage_options	dict	storage options to pass to write functions	{}
annotation_column	string	column containing url to regional geojson	''
output_column_suffix	string	column suffix with result url to add to slide_manifest	required

Returns:

Type	Description
	DataFrame[SlideSchema]: slide schema

Source code in src/luna/pathology/cli/dsa_viz.py

def regional_polygon(
    slide_manifest: DataFrame[SlideSchema],
    output_urlpath: str,
    annotation_name: str,
    line_colors: Optional[Dict[str, str]] = None,
    fill_colors: Optional[Dict[str, str]] = None,
    storage_options: Dict = {},
    output_storage_options: Dict = {},
    annotation_column: str = "",
    output_column: str = "",
):
    """Build DSA annotation json from regional annotation geojson

    Args:
        slide_manifest (DataFrame[SlideSchema]): slide manifest
        output_urlpath (string): URL/path prefix for saving dsa annotation json
        annotation_name (string): name of the annotation to be displayed in DSA
        line_colors (dict, optional): line color map with {feature name:rgb values}
        fill_colors (dict, optional): fill color map with {feature name:rgba values}
        storage_options (dict): storage options to pass to read functions
        output_storage_options (dict): storage options to pass to write functions
        annotation_column (string): column containing url to regional geojson
        output_column_suffix (string): column suffix with result url to add to slide_manifest

    Returns:
        DataFrame[SlideSchema]: slide schema
    """

    if not annotation_column:
        annotation_column = f"{annotation_name}_geojson_url"
    if not output_column:
        output_column = f"{annotation_name}_dsa_url"
    if annotation_column not in slide_manifest.columns:
        raise ValueError(f"{annotation_column} not found in slide manifest")
    client = get_or_create_dask_client()
    futures = []
    for _, row in slide_manifest.iterrows():
        image_filename = os.path.basename(row["url"])
        future = client.submit(
            __regional_polygon,
            row[annotation_column],
            output_urlpath,
            image_filename,
            annotation_name,
            fill_colors,
            line_colors,
            storage_options,
            output_storage_options,
        )

        futures.append(future)
    progress(futures)
    dsa_annotation_urls = client.gather(futures)
    return slide_manifest.assign(**{output_column: dsa_annotation_urls})

Build DSA annotation json from Qupath polygon geojson

Parameters:

Name	Type	Description	Default
slide_manifest	DataFrame[SlideSchema]	slide manifest from slide_etl	required
output_urlpath	string	URL/path prefix for saving the DSA compatible annotation	required
image_filename	string	name of the image file in DSA e.g. 123.svs	required
annotation_name	string	name of the annotation to be displayed in DSA	required
classes_to_include	list	list of classification labels to visualize	required
line_colors	dict	line color map with {feature name:rgb values}	None
fill_colors	dict	fill color map with {feature name:rgba values}	None
storage_options	dict	storage options to pass to read functions	{}
output_storage_options	dict	storage options to pass to write functions	{}
annotation_column	string	column containing url to qupath geojson	''
output_column_suffix	string	column suffix with result url to add to slide_manifest	required

Returns:

Type	Description
	DataFrame[SlideSchema]: slide manifest

Source code in src/luna/pathology/cli/dsa_viz.py

def qupath_polygon(
    slide_manifest: DataFrame[SlideSchema],
    output_urlpath: str,
    image_filename: str,
    annotation_name: str,
    classes_to_include: List,
    line_colors: Optional[Dict[str, str]] = None,
    fill_colors: Optional[Dict[str, str]] = None,
    storage_options: Dict = {},
    output_storage_options: Dict = {},
    annotation_column: str = "",
    output_column: str = "",
):
    """Build DSA annotation json from Qupath polygon geojson

    Args:
        slide_manifest (DataFrame[SlideSchema]): slide manifest from slide_etl
        output_urlpath (string): URL/path prefix for saving the DSA compatible annotation
        json
        image_filename (string): name of the image file in DSA e.g. 123.svs
        annotation_name (string): name of the annotation to be displayed in DSA
        classes_to_include (list): list of classification labels to visualize
        e.g. ["Tumor", "Stroma", ...]
        line_colors (dict, optional): line color map with {feature name:rgb values}
        fill_colors (dict, optional): fill color map with {feature name:rgba values}
        storage_options (dict): storage options to pass to read functions
        output_storage_options (dict): storage options to pass to write functions
        annotation_column (string): column containing url to qupath geojson
        output_column_suffix (string): column suffix with result url to add to slide_manifest

    Returns:
        DataFrame[SlideSchema]: slide manifest
    """
    if not annotation_column:
        annotation_column = f"{annotation_name}_geojson_url"
    if not output_column:
        output_column = f"{annotation_name}_dsa_url"
    if annotation_column not in slide_manifest.columns:
        raise ValueError(f"{annotation_column} not found in slide manifest")
    client = get_or_create_dask_client()
    futures = []
    for _, row in slide_manifest.iterrows():
        image_filename = os.path.basename(row["url"])
        future = client.submit(
            __qupath_polygon,
            row[annotation_column],
            output_urlpath,
            image_filename,
            annotation_name,
            classes_to_include,
            line_colors,
            fill_colors,
            storage_options,
            output_storage_options,
        )

        futures.append(future)
    progress(futures)
    dsa_annotation_urls = client.gather(futures)
    return slide_manifest.assign(**{output_column: dsa_annotation_urls})

Build DSA annotation json from bitmask PNGs

Vectorizes and simplifies contours from the bitmask.

Parameters:

Name	Type	Description	Default
input	map	map of {label:urlpath_to_bitmask_png}	required
annotation_name	string	name of the annotation to be displayed in DSA	required
line_colors	dict	line color map with {feature name:rgb values}	None
fill_colors	dict	fill color map with {feature name:rgba values}	None
scale_factor	int	scale to match the image on DSA.	1
storage_options	dict	storage options to pass to read/write functions	{}

Returns:

Name	Type	Description
dict		DSA annotation

Source code in src/luna/pathology/cli/dsa_viz.py

def bitmask_polygon(
    input_map: Dict[str, str],
    output_urlpath: str,
    image_filename: str,
    annotation_name: str,
    line_colors: Optional[Dict[str, str]] = None,
    fill_colors: Optional[Dict[str, str]] = None,
    scale_factor: Optional[int] = 1,
    storage_options: Dict = {},
    output_storage_options: Dict = {},
):
    """Build DSA annotation json from bitmask PNGs

    Vectorizes and simplifies contours from the bitmask.

    Args:
        input (map): map of {label:urlpath_to_bitmask_png}
        annotation_name (string): name of the annotation to be displayed in DSA
        line_colors (dict, optional): line color map with {feature name:rgb values}
        fill_colors (dict, optional): fill color map with {feature name:rgba values}
        scale_factor (int, optional): scale to match the image on DSA.
        storage_options (dict): storage options to pass to read/write functions

    Returns:
        dict: DSA annotation
    """
    if not check_filepaths_valid(input_map.values(), storage_options):
        raise ValueError("No valid PNG masks found. Exiting..")

    elements = []
    for bitmask_label, bitmask_filepath in input_map.items():
        Image.MAX_IMAGE_PIXELS = 5000000000
        with open(bitmask_filepath, "rb", **storage_options).open() as of:
            annotation = Image.open(of)
            bitmask_np = np.array(annotation)
        simplified_contours = vectorize_np_array_bitmask_by_pixel_value(
            bitmask_np, scale_factor=scale_factor
        )

        for n, contour in enumerate(simplified_contours):
            element = copy.deepcopy(base_dsa_polygon_element)
            label_name = bitmask_label
            element["label"]["value"] = label_name
            if fill_colors and label_name in fill_colors:
                element["fillColor"] = fill_colors[label_name]
            if line_colors and label_name in line_colors:
                element["lineColor"] = line_colors[label_name]

            coords = contour.tolist()
            for c in coords:
                c.append(0)
            element["points"] = coords
            elements.append(element)

    dsa_annotation = get_dsa_annotation(elements, annotation_name)
    return save_dsa_annotation(
        dsa_annotation,
        output_urlpath,
        image_filename,
        output_storage_options,
    )

Generate heatmap based on the tile scores

Creates a heatmap for the given column, using the color palette viridis to set a fill value - the color ranges from purple to yellow, for scores from 0 to 1.

Parameters:

Name	Type	Description	Default
slide_manifest	DataFrame[SlideSchema]	slide manifest from slide_etl	required
output_urlpath	string	URL/path prefix to save the DSA compatible annotation	required
annotation_name	string	name of the annotation to be displayed in DSA	required
column	string	column to visualize e.g. tile_score	required
tile_size	int	size of tiles	required
scale_factor	int	scale to match the image on DSA.	None
line_colors	dict	line color map with {feature name:rgb values}	None
fill_colors	dict	fill color map with {feature name:rgba values}	None
storage_options	dict	storage options to pass to read functions	{}
output_storage_options	dict	storage options to pass to write functions	{}

Returns:

Name	Type	Description
dict		annotation file path. None if error in writing the file.

Source code in src/luna/pathology/cli/dsa_viz.py

def heatmap(
    slide_manifest: DataFrame[SlideSchema],
    output_urlpath: str,
    annotation_name: str,
    column: List[str],
    tile_size: int,
    scale_factor: Optional[int] = None,
    fill_colors: Optional[Dict[str, str]] = None,
    line_colors: Optional[Dict[str, str]] = None,
    output_column: str = "",
    storage_options: Dict = {},
    output_storage_options: Dict = {},
):
    """Generate heatmap based on the tile scores

    Creates a heatmap for the given column, using the color palette `viridis`
    to set a fill value
    - the color ranges from purple to yellow, for scores from 0 to 1.

    Args:
        slide_manifest (DataFrame[SlideSchema]): slide manifest from slide_etl
        output_urlpath (string): URL/path prefix to save the DSA compatible annotation
        json
        annotation_name (string): name of the annotation to be displayed in DSA
        column (string): column to visualize e.g. tile_score
        tile_size (int): size of tiles
        scale_factor (int, optional): scale to match the image on DSA.
        line_colors (dict, optional): line color map with {feature name:rgb values}
        fill_colors (dict, optional): fill color map with {feature name:rgba values}
        storage_options (dict): storage options to pass to read functions
        output_storage_options (dict): storage options to pass to write functions

    Returns:
        dict: annotation file path. None if error in writing the file.
    """
    if not output_column:
        output_column = f"{annotation_name}_dsa_url"
    if "tiles_url" not in slide_manifest.columns:
        raise ValueError("tiles_url not found in slide manifest")
    client = get_or_create_dask_client()
    futures = []
    for _, row in slide_manifest.iterrows():
        image_filename = os.path.basename(row["url"])
        future = client.submit(
            __heatmap,
            row["tiles_url"],
            output_urlpath,
            image_filename,
            annotation_name,
            column,
            tile_size,
            scale_factor,
            fill_colors,
            line_colors,
            storage_options,
            output_storage_options,
        )

        futures.append(future)
    progress(futures)
    dsa_annotation_urls = client.gather(futures)
    return slide_manifest.assign(**{output_column: dsa_annotation_urls})

Build DSA annotation json from a BMP with multiple labels.

Vectorizes and simplifies contours per label.

Parameters:

Name	Type	Description	Default
slide_manifest	DataFrame[SlideSchema]	slide manifest from slide_etl	required
output_urlpath	string	url/path prefix to save the DSA compatible annotation	required
label_map	dict[int, str]	map of label number to label name	required
annotation_name	string	name of the annotation to be displayed in DSA	required
line_colors	dict[str, str]	line color map with {feature name:rgb values}	None
fill_colors	dict[str, str]	fill color map with {feature name:rgba values}	None
scale_factor	int	scale to match image DSA.	1
storage_options	dict	storage options to pass to read functions	{}
output_storage_options	dict	storage options to pass to write functions	{}
annotation_column	string	column containing url to BMP polygon	'bmp_polygon_url'
output_column_suffix	string	column suffix with result url to add to slide_manifest	required

Returns:

Name	Type	Description
dict		annotation file path

Source code in src/luna/pathology/cli/dsa_viz.py

def bmp_polygon(
    slide_manifest: DataFrame[SlideSchema],
    output_urlpath: str,
    label_map: Dict[int, str],
    annotation_name: str,
    line_colors: Optional[Dict[str, str]] = None,
    fill_colors: Optional[Dict[str, str]] = None,
    scale_factor: Optional[int] = 1,
    storage_options: Dict = {},
    output_storage_options: Dict = {},
    annotation_column: str = "bmp_polygon_url",
    output_column: str = "bmp_polygon_dsa_url",
):
    """Build DSA annotation json from a BMP with multiple labels.

    Vectorizes and simplifies contours per label.

    Args:
        slide_manifest (DataFrame[SlideSchema]): slide manifest from slide_etl
        output_urlpath (string): url/path prefix to save the DSA compatible annotation
        json
        label_map (dict[int,str]): map of label number to label name
        annotation_name (string): name of the annotation to be displayed in DSA
        line_colors (dict[str,str], optional): line color map with {feature name:rgb values}
        fill_colors (dict[str,str], optional): fill color map with {feature name:rgba values}
        scale_factor (int, optional): scale to match image DSA.
        storage_options (dict): storage options to pass to read functions
        output_storage_options (dict): storage options to pass to write functions
        annotation_column (string): column containing url to BMP polygon
        output_column_suffix (string): column suffix with result url to add to slide_manifest

    Returns:
        dict: annotation file path
    """
    if annotation_column not in slide_manifest.columns:
        raise ValueError(f"{annotation_column} not found in slide manifest")
    client = get_or_create_dask_client()
    futures = []
    for _, row in slide_manifest.iterrows():
        image_filename = os.path.basename(row["url"])
        future = client.submit(
            __bmp_polygon,
            row[annotation_column],
            output_urlpath,
            image_filename,
            label_map,
            annotation_name,
            line_colors,
            fill_colors,
            scale_factor,
            storage_options,
            output_storage_options,
        )
        futures.append(future)
    progress(futures)
    dsa_annotation_urls = client.gather(futures)
    return slide_manifest.assign(**{output_column: dsa_annotation_urls})