Post processing

ResultType

Bases: Enum

Enumeration of the possible result types.

Source code in src/pheval/post_processing/post_processing.py

class ResultType(Enum):
    """Enumeration of the possible result types."""

    GENE = "gene"
    DISEASE = "disease"
    VARIANT = "variant"

SortOrder

Bases: Enum

Enumeration representing sorting orders.

Source code in src/pheval/post_processing/post_processing.py

class SortOrder(Enum):
    """Enumeration representing sorting orders."""

    ASCENDING = 1
    """Ascending sort order."""
    DESCENDING = 2
    """Descending sort order."""

ASCENDING = 1 class-attribute instance-attribute

Ascending sort order.

DESCENDING = 2 class-attribute instance-attribute

Descending sort order.

create_empty_pheval_result(phenopacket_dir, output_dir, result_type)

Create an empty PhEval result for a given result type (gene, variant, or disease).

Notes

This is necessary because some tools may not generate a result output for certain cases. By explicitly creating an empty result, which will contain the known entity with a rank and score of 0, we can track and identify false negatives during benchmarking, ensuring that missing predictions are accounted for in the evaluation.

Parameters:

Name	Type	Description	Default
phenopacket_dir	Path	The directory containing the phenopackets.	required
output_dir	Path	The output directory.	required
result_type	ResultType	The result type.	required

Source code in src/pheval/post_processing/post_processing.py

def create_empty_pheval_result(phenopacket_dir: Path, output_dir: Path, result_type: ResultType) -> None:
    """
    Create an empty PhEval result for a given result type (gene, variant, or disease).

    Notes:
        This is necessary because some tools may not generate a result output for certain cases.
        By explicitly creating an empty result, which will contain the known entity with a rank and score of 0,
        we can track and identify false negatives  during benchmarking,
        ensuring that missing predictions are accounted for in the evaluation.

    Args:
        phenopacket_dir (Path): The directory containing the phenopackets.
        output_dir (Path): The output directory.
        result_type (ResultType): The result type.

    """
    if result_type in executed_results:
        return
    logger.info(f"Writing classified results for {len(all_files(phenopacket_dir))} phenopackets to {output_dir}")
    executed_results.add(result_type)
    phenopacket_truth_set = PhenopacketTruthSet(phenopacket_dir)
    classify_method, write_method = _get_result_type(result_type, phenopacket_truth_set)
    for file in all_files(phenopacket_dir):
        classified_results = classify_method(file.stem)
        write_method(
            classified_results,
            output_dir.joinpath(f"{file.stem}-{result_type.value}_result.parquet"),
        )

generate_disease_result(results, sort_order, output_dir, result_path, phenopacket_dir)

Generate PhEval disease results to a compressed Parquet output. Args: results (pl.DataFrame): The disease results. sort_order (SortOrder): The sort order to use. output_dir (Path): Path to the output directory result_path (Path): Path to the tool-specific result file. phenopacket_dir (Path): Path to the Phenopacket directory

Source code in src/pheval/post_processing/post_processing.py

@validate_dataframe(ResultSchema.DISEASE_RESULT_SCHEMA)
def generate_disease_result(
    results: pl.DataFrame,
    sort_order: SortOrder,
    output_dir: Path,
    result_path: Path,
    phenopacket_dir: Path,
) -> None:
    """
    Generate PhEval disease results to a compressed Parquet output.
    Args:
        results (pl.DataFrame): The disease results.
        sort_order (SortOrder): The sort order to use.
        output_dir (Path): Path to the output directory
        result_path (Path): Path to the tool-specific result file.
        phenopacket_dir (Path): Path to the Phenopacket directory
    """
    output_file = output_dir.joinpath(f"pheval_disease_results/{result_path.stem}-disease_result.parquet")
    create_empty_pheval_result(
        phenopacket_dir,
        output_dir.joinpath("pheval_disease_results"),
        ResultType.DISEASE,
    )
    ranked_results = _rank_results(results, sort_order)
    classified_results = PhenopacketTruthSet(phenopacket_dir).merge_disease_results(
        ranked_results, output_file, mondo_mapping_table
    )

    _write_disease_result(classified_results, output_file)

generate_gene_result(results, sort_order, output_dir, result_path, phenopacket_dir)

Generate PhEval gene results to a compressed Parquet output. Args: results (pl.DataFrame): The gene results. sort_order (SortOrder): The sort order to use. output_dir (Path): Path to the output directory result_path (Path): Path to the tool-specific result file. phenopacket_dir (Path): Path to the Phenopacket directory

Source code in src/pheval/post_processing/post_processing.py

@validate_dataframe(ResultSchema.GENE_RESULT_SCHEMA)
def generate_gene_result(
    results: pl.DataFrame,
    sort_order: SortOrder,
    output_dir: Path,
    result_path: Path,
    phenopacket_dir: Path,
) -> None:
    """
    Generate PhEval gene results to a compressed Parquet output.
    Args:
        results (pl.DataFrame): The gene results.
        sort_order (SortOrder): The sort order to use.
        output_dir (Path): Path to the output directory
        result_path (Path): Path to the tool-specific result file.
        phenopacket_dir (Path): Path to the Phenopacket directory
    """
    output_file = output_dir.joinpath(f"pheval_gene_results/{result_path.stem}-gene_result.parquet")
    create_empty_pheval_result(phenopacket_dir, output_dir.joinpath("pheval_gene_results"), ResultType.GENE)
    ranked_results = _rank_results(results, sort_order)
    classified_results = PhenopacketTruthSet(phenopacket_dir).merge_gene_results(ranked_results, output_file)
    _write_gene_result(classified_results, output_file)

generate_variant_result(results, sort_order, output_dir, result_path, phenopacket_dir)

Generate PhEval variant results to a compressed Parquet output. Args: results (pl.DataFrame): The variant results. sort_order (SortOrder): The sort order to use. output_dir (Path): Path to the output directory result_path (Path): Path to the tool-specific result file. phenopacket_dir (Path): Path to the Phenopacket directory

Source code in src/pheval/post_processing/post_processing.py

@validate_dataframe(ResultSchema.VARIANT_RESULT_SCHEMA)
def generate_variant_result(
    results: pl.DataFrame,
    sort_order: SortOrder,
    output_dir: Path,
    result_path: Path,
    phenopacket_dir: Path,
) -> None:
    """
    Generate PhEval variant results to a compressed Parquet output.
    Args:
        results (pl.DataFrame): The variant results.
        sort_order (SortOrder): The sort order to use.
        output_dir (Path): Path to the output directory
        result_path (Path): Path to the tool-specific result file.
        phenopacket_dir (Path): Path to the Phenopacket directory
    """
    output_file = output_dir.joinpath(f"pheval_variant_results/{result_path.stem}-variant_result.parquet")
    create_empty_pheval_result(
        phenopacket_dir,
        output_dir.joinpath("pheval_variant_results"),
        ResultType.VARIANT,
    )
    ranked_results = _rank_results(results, sort_order).with_columns(
        pl.concat_str(["chrom", "start", "ref", "alt"], separator="-").alias("variant_id")
    )
    classified_results = PhenopacketTruthSet(phenopacket_dir).merge_variant_results(ranked_results, output_file)
    _write_variant_result(classified_results, output_file)