Developing a PhEval Plugin

Description

Plugin development allows PhEval to be extensible, as we have designed it. The plugin goal is to be flexible through custom runner implementations. This plugin development enhances the PhEval functionality. You can build one quickly using this step-by-step process.

All custom Runners implementations must implement all PhevalRunner methods

Bases: ABC

PhEvalRunner Class

Source code in src/pheval/runners/runner.py

@dataclass
class PhEvalRunner(ABC):
    """PhEvalRunner Class"""

    input_dir: Path
    testdata_dir: Path
    tmp_dir: Path
    output_dir: Path
    config_file: Path
    version: str
    directory_path = None
    input_dir_config = None
    _meta_data = None
    __raw_results_dir = "raw_results/"
    __pheval_gene_results_dir = "pheval_gene_results/"
    __pheval_variant_results_dir = "pheval_variant_results/"
    __pheval_disease_results_dir = "pheval_disease_results/"
    __tool_input_commands_dir = "tool_input_commands/"
    __run_meta_data_file = "results.yml"

    def __post_init__(self):
        self.input_dir_config = parse_input_dir_config(self.input_dir)

    def _get_tool(self):
        return self.input_dir_config.tool

    def _get_variant_analysis(self):
        return self.input_dir_config.variant_analysis

    def _get_gene_analysis(self):
        return self.input_dir_config.gene_analysis

    def _get_disease_analysis(self):
        return self.input_dir_config.disease_analysis

    @property
    def tool_input_commands_dir(self):
        return Path(self.output_dir).joinpath(self.__tool_input_commands_dir)

    @tool_input_commands_dir.setter
    def tool_input_commands_dir(self, directory_path):
        self.directory_path = Path(directory_path)

    @property
    def raw_results_dir(self):
        return Path(self.output_dir).joinpath(self.__raw_results_dir)

    @raw_results_dir.setter
    def raw_results_dir(self, directory_path):
        self.directory_path = Path(directory_path)

    @property
    def pheval_gene_results_dir(self):
        return Path(self.output_dir).joinpath(self.__pheval_gene_results_dir)

    @pheval_gene_results_dir.setter
    def pheval_gene_results_dir(self, directory_path):
        self.directory_path = Path(directory_path)

    @property
    def pheval_variant_results_dir(self):
        return Path(self.output_dir).joinpath(self.__pheval_variant_results_dir)

    @pheval_variant_results_dir.setter
    def pheval_variant_results_dir(self, directory_path):
        self.directory_path = Path(directory_path)

    @property
    def pheval_disease_results_dir(self):
        return Path(self.output_dir).joinpath(self.__pheval_disease_results_dir)

    @pheval_disease_results_dir.setter
    def pheval_disease_results_dir(self, directory_path):
        self.directory_path = Path(directory_path)

    def build_output_directory_structure(self):
        """build output directory structure"""
        logger.info(
            f"Building output directory structure for {self.input_dir_config.tool} "
            f"version {self.input_dir_config.tool_version}"
        )
        self.tool_input_commands_dir.mkdir(exist_ok=True)
        self.raw_results_dir.mkdir(exist_ok=True)
        if self._get_variant_analysis():
            self.pheval_variant_results_dir.mkdir(exist_ok=True)
        if self._get_gene_analysis():
            self.pheval_gene_results_dir.mkdir(exist_ok=True)
        if self._get_disease_analysis():
            self.pheval_disease_results_dir.mkdir(exist_ok=True)

    @property
    def meta_data(self):
        self._meta_data = BasicOutputRunMetaData(
            tool=self.input_dir_config.tool,
            tool_version=self.version,
            config=f"{Path(self.input_dir).parent.name}/{Path(self.input_dir).name}",
            run_timestamp=datetime.now().timestamp(),
            corpus=f"{Path(self.testdata_dir).parent.name}/{Path(self.testdata_dir).name}",
        )
        return self._meta_data

    @meta_data.setter
    def meta_data(self, meta_data):
        self._meta_data = meta_data

    @abstractmethod
    def prepare(self) -> str:
        """prepare"""

    @abstractmethod
    def run(self):
        """run"""

    @abstractmethod
    def post_process(self):
        """post_process"""

    def construct_meta_data(self):
        """Construct run output meta data"""
        return self.meta_data

---

Step-by-Step Plugin Development Process

The plugin structure is derived from a cookiecutter template, cookiecutter, and it uses MkDocs, tox and poetry as core dependencies. This allows PhEval extensibility to be standardised in terms of documentation and dependency management.

1. Cookiecutter scaffold

First, install the cruft package. Cruft enables keeping projects up-to-date with future updates made to this original template.

Install the latest release of cruft from pip

pip install cruft

NOTE: You may encounter an error with the naming of the project layout if using an older release of cruft. To avoid this, make sure you have installed the latest release version.

Next, create a project using the cookiecutter template.

cruft create https://github.com/monarch-initiative/pheval-runner-template

2. Further setup

Install poetry if you haven't already.

pip install poetry

Install dependencies

poetry install

Run tox to see if the setup works

poetry run tox

3. Implement PhEval Custom Runner

In the project structure generated by Cookiecutter, you'll find runner.py located in the src directory. This is where you'll define the methods required to develop the plugin. Specifically, you'll implement the prepare, run, and post-process methods, which are essential for executing the pheval run command.

"""Runner."""

from dataclasses import dataclass
from pathlib import Path

from pheval.runners.runner import PhEvalRunner


@dataclass
class CustomRunner(PhEvalRunner):
    """Runner class implementation."""

    input_dir: Path
    testdata_dir: Path
    tmp_dir: Path
    output_dir: Path
    config_file: Path
    version: str

    def prepare(self):
        """Prepare."""
        print("preparing")

    def run(self):
        """Run."""
        print("running")

    def post_process(self):
        """Post Process."""
        print("post processing")

The Cookiecutter will automatically populate the plugins section in the pyproject.toml file. If you decide to modify the path of runner.py or rename its class, be sure to update the corresponding entries in this section accordingly:

[tool.poetry.plugins."pheval.plugins"]
customrunner = "pheval_plugin_example.runner:CustomRunner"

Please Note that the path here and naming of the class is case-sensitive.

4. Implementing PhEval helper methods

Streamlining the creation of your custom PhEval runner can be facilitated by leveraging PhEval's versatile helper methods, where applicable.

Within PhEval, numerous public methods have been designed to assist in your runner methods. The utilisation of these helper methods is optional, yet they are crafted to enhance the overall implementation process.

Utility methods

The PhenopacketUtil class is designed to aid in the collection of specific data from a Phenopacket.

Class for retrieving data from a Phenopacket or Family object

Source code in src/pheval/utils/phenopacket_utils.py

class PhenopacketUtil:
    """Class for retrieving data from a Phenopacket or Family object"""

    def __init__(self, phenopacket_contents: Union[Phenopacket, Family]):
        """Initialise PhenopacketUtil

        Args:
            phenopacket_contents (Union[Phenopacket, Family]): Phenopacket or Family object
        """
        self.phenopacket_contents = phenopacket_contents

    def sample_id(self) -> str:
        """
        Retrieve the sample ID from a Phenopacket or proband of a Family

        Returns:
            str: Sample ID
        """
        if hasattr(self.phenopacket_contents, "proband"):
            return self.phenopacket_contents.proband.subject.id
        else:
            return self.phenopacket_contents.subject.id

    def phenotypic_features(self) -> List[PhenotypicFeature]:
        """
        Retrieve a list of all HPO terms

        Returns:
            List[PhenotypicFeature]: List of HPO terms
        """
        if hasattr(self.phenopacket_contents, "proband"):
            return self.phenopacket_contents.proband.phenotypic_features
        else:
            return self.phenopacket_contents.phenotypic_features

    def observed_phenotypic_features(self) -> List[PhenotypicFeature]:
        """
        Retrieve a list of all observed HPO terms

        Returns:
            List[PhenotypicFeature]: List of observed HPO terms
        """
        phenotypic_features = []
        all_phenotypic_features = self.phenotypic_features()
        for p in all_phenotypic_features:
            if p.excluded:
                continue
            phenotypic_features.append(p)
        return phenotypic_features

    def negated_phenotypic_features(self) -> List[PhenotypicFeature]:
        """
        Retrieve a list of all negated HPO terms

        Returns:
            List[PhenotypicFeature]: List of negated HPO terms
        """
        negated_phenotypic_features = []
        all_phenotypic_features = self.phenotypic_features()
        for p in all_phenotypic_features:
            if p.excluded:
                negated_phenotypic_features.append(p)
        return negated_phenotypic_features

    def diseases(self) -> List[Disease]:
        """
        Retrieve a list of Diseases associated with the proband

        Returns:
            List[Disease]: List of diseases
        """
        if hasattr(self.phenopacket_contents, "proband"):
            return self.phenopacket_contents.proband.diseases
        else:
            return self.phenopacket_contents.diseases

    def _diagnosis_from_interpretations(self) -> List[ProbandDisease]:
        """
        Retrieve a list of disease diagnoses associated with the proband from the interpretations object

        Returns:
            List[ProbandDisease]: List of diagnosed diseases
        """
        diagnoses = []
        interpretation = self.interpretations()
        for i in interpretation:
            (
                diagnoses.append(
                    ProbandDisease(
                        disease_name=i.diagnosis.disease.label,
                        disease_identifier=i.diagnosis.disease.id,
                    )
                )
                if i.diagnosis.disease.label != "" and i.diagnosis.disease.id != ""
                else None
            )
        return diagnoses

    def _diagnosis_from_disease(self) -> List[ProbandDisease]:
        """
        Retrieve a list of disease diagnoses associated with the proband from the diseases object

        Returns:
            List[ProbandDisease]: List of diagnosed diseases
        """
        diagnoses = []
        for disease in self.diseases():
            diagnoses.append(
                ProbandDisease(disease_name=disease.term.label, disease_identifier=disease.term.id)
            )
        return diagnoses

    def diagnoses(self) -> List[ProbandDisease]:
        """
        Retrieve a unique list of disease diagnoses associated with the proband from a Phenopacket

        Returns:
            List[ProbandDisease]: List of diagnosed diseases
        """
        return list(set(self._diagnosis_from_interpretations() + self._diagnosis_from_disease()))

    def interpretations(self) -> List[Interpretation]:
        """
        Retrieve a list of interpretations from a Phenopacket

        Returns:
            List[Interpretation]: List of interpretations
        """
        if hasattr(self.phenopacket_contents, "proband"):
            return self.phenopacket_contents.proband.interpretations
        else:
            return self.phenopacket_contents.interpretations

    def causative_variants(self) -> List[ProbandCausativeVariant]:
        """
        Retrieve a list of causative variants listed in a Phenopacket

        Returns:
            List[ProbandCausativeVariant]: List of proband causative variants
        """
        all_variants = []
        interpretation = self.interpretations()
        for i in interpretation:
            for g in i.diagnosis.genomic_interpretations:
                vcf_record = g.variant_interpretation.variation_descriptor.vcf_record
                genotype = g.variant_interpretation.variation_descriptor.allelic_state
                variant_data = ProbandCausativeVariant(
                    self.phenopacket_contents.subject.id,
                    vcf_record.genome_assembly,
                    GenomicVariant(
                        vcf_record.chrom,
                        vcf_record.pos,
                        vcf_record.ref,
                        vcf_record.alt,
                    ),
                    genotype.label,
                    vcf_record.info,
                )
                all_variants.append(variant_data)
        return all_variants

    def files(self) -> List[File]:
        """
        Retrieve a list of files associated with a phenopacket

        Returns:
            List[File]: List of files associated with a phenopacket
        """
        return self.phenopacket_contents.files

    def vcf_file_data(self, phenopacket_path: Path, vcf_dir: Path) -> File:
        """
        Retrieve the genome assembly and VCF file name from a phenopacket.

        Args:
            phenopacket_path (Path): The path to the phenopacket file.
            vcf_dir (Path): The directory path where the VCF file is stored.

        Returns:
            File: The VCF file with updated URI pointing to the specified directory.

        Raises:
            IncorrectFileFormatError: If the provided file is not in .vcf or .vcf.gz format.
            IncompatibleGenomeAssemblyError: If the genome assembly of the VCF file is not compatible.

        Note:
            This function searches for a VCF file within the provided list of files, validates its format,
            and checks if the genome assembly is compatible. If the conditions are met, it updates the
            URI of the VCF file to the specified directory and returns the modified file object.
        """
        compatible_genome_assembly = ["GRCh37", "hg19", "GRCh38", "hg38"]
        vcf_data = [file for file in self.files() if file.file_attributes["fileFormat"] == "vcf"][0]
        if not Path(vcf_data.uri).name.endswith(".vcf") and not Path(vcf_data.uri).name.endswith(
            ".vcf.gz"
        ):
            raise IncorrectFileFormatError(Path(vcf_data.uri), ".vcf or .vcf.gz file")
        if vcf_data.file_attributes["genomeAssembly"] not in compatible_genome_assembly:
            raise IncompatibleGenomeAssemblyError(
                vcf_data.file_attributes["genomeAssembly"], phenopacket_path
            )
        vcf_data.uri = str(vcf_dir.joinpath(Path(vcf_data.uri).name))
        return vcf_data

    @staticmethod
    def _extract_diagnosed_gene(
        genomic_interpretation: GenomicInterpretation,
    ) -> ProbandCausativeGene:
        """
        Retrieve the disease causing genes from the variant descriptor field if not empty,
        otherwise, retrieves from the gene descriptor from a phenopacket.
        Args:
            genomic_interpretation (GenomicInterpretation): A genomic interpretation from a Phenopacket
        Returns:
            ProbandCausativeGene: The disease causing gene
        """
        if genomic_interpretation.variant_interpretation.ByteSize() != 0:
            return ProbandCausativeGene(
                genomic_interpretation.variant_interpretation.variation_descriptor.gene_context.symbol,
                genomic_interpretation.variant_interpretation.variation_descriptor.gene_context.value_id,
            )

        else:
            return ProbandCausativeGene(
                gene_symbol=genomic_interpretation.gene.symbol,
                gene_identifier=genomic_interpretation.gene.value_id,
            )

    def diagnosed_genes(self) -> List[ProbandCausativeGene]:
        """
        Retrieve the disease causing genes from a phenopacket.
        Returns:
            List[ProbandCausativeGene]: List of causative genes
        """
        pheno_interpretation = self.interpretations()
        genes = []
        for i in pheno_interpretation:
            for g in i.diagnosis.genomic_interpretations:
                genes.append(self._extract_diagnosed_gene(g))
                genes = list({gene.gene_symbol: gene for gene in genes}.values())
        return genes

    def diagnosed_variants(self) -> List[GenomicVariant]:
        """
        Retrieve a list of all known causative variants from a phenopacket.
        Returns:
            List[GenomicVariant]: List of causative variants
        """
        variants = []
        pheno_interpretation = self.interpretations()
        for i in pheno_interpretation:
            for g in i.diagnosis.genomic_interpretations:
                variant = GenomicVariant(
                    chrom=str(
                        g.variant_interpretation.variation_descriptor.vcf_record.chrom.replace(
                            "chr", ""
                        )
                    ),
                    pos=int(g.variant_interpretation.variation_descriptor.vcf_record.pos),
                    ref=g.variant_interpretation.variation_descriptor.vcf_record.ref,
                    alt=g.variant_interpretation.variation_descriptor.vcf_record.alt,
                )
                variants.append(variant)
        return variants

    def check_incomplete_variant_record(self) -> bool:
        """
        Check if any variant record in the phenopacket has incomplete information.

        This method iterates through the diagnosed variant records and checks if any of them
        have missing or incomplete information such as empty chromosome, position, reference,
        or alternate allele.

        Returns:
            bool: True if any variant record is incomplete, False otherwise.
        """
        variants = self.diagnosed_variants()
        for variant in variants:
            if (
                variant.chrom == ""
                or variant.pos == 0
                or variant.pos == ""
                or variant.ref == ""
                or variant.alt == ""
            ):
                return True
        return False

    def check_variant_alleles(self) -> bool:
        """
        Check if any variant record in the phenopacket has identical reference and alternate alleles.

        Returns:
            bool: True if the reference and alternate alleles are identical, False otherwise.
        """
        variants = self.diagnosed_variants()
        for variant in variants:
            if variant.ref == variant.alt:
                return True
        return False

    def check_incomplete_gene_record(self) -> bool:
        """
        Check if any gene record in the phenopacket has incomplete information.

        This method iterates through the diagnosed gene records and checks if any of them
        have missing or incomplete information such as gene name, or gene identifier.

        Returns:
            bool: True if any gene record is incomplete, False otherwise.
        """
        genes = self.diagnosed_genes()
        for gene in genes:
            if gene.gene_symbol == "" or gene.gene_identifier == "":
                return True
        return False

    def check_incomplete_disease_record(self) -> bool:
        """
        Check if any disease record in the phenopacket has incomplete information.

        This method iterates through the diagnosed disease records and checks if any of them
        have missing or incomplete information such as empty disease name, or disease identifier.

        Returns:
            bool: True if any disease record is incomplete, False otherwise.
        """
        if len(self.diagnoses()) == 0:
            return True
        return False

---

PhenopacketUtil proves particularly beneficial in scenarios where the tool for which you're crafting a runner implementation does not directly accept Phenopackets as inputs. Instead, it might require elements—such as HPO IDs— via the command-line interface (CLI). In this context, leveraging PhenopacketUtil within the runner's preparation phase enables the extraction of observed phenotypic features from the Phenopacket input, facilitating seamless processing.

An example of how this could be implemented is outlined here:

from pheval.utils.phenopacket_utils import phenopacket_reader
from pheval.utils.phenopacket_utils import PhenopacketUtil

phenopacket = phenopacket_reader("/path/to/phenopacket.json")
phenopacket_util = PhenopacketUtil(phenopacket)
# To return a list of all observed phenotypes for a phenopacket
observed_phenotypes = phenopacket_util.observed_phenotypic_features()
# To extract just the HPO ID as a list
observed_phenotypes_hpo_ids = [
    observed_phenotype.type.id for observed_phenotype in observed_phenotypes
]

Additional tool-specific configurations

For the pheval run command to execute successfully, a config.yaml should be found within the input directory supplied on the CLI.

tool: 
tool_version: 
variant_analysis: 
gene_analysis: 
disease_analysis: 
tool_specific_configuration_options:

The tool_specific_configuration_options is an optional field that can be populated with any variables specific to your runner implementation that is required for the running of your tool.

All other fields are required to be filled in. The variant_analysis, gene_analysis, and disease_analysis are set as booleans and are for specifying what type of analysis/prioritisation the tool outputs.

To populate the tool_specific_configurations_options with customised data, we suggest using the pydantic package as it can easily parse the data from the yaml structure.

e.g.,

Define a BaseModel class with the fields that will populate the tool_specific_configuration_options

from pydantic import BaseModel, Field

class CustomisedConfigurations(BaseModel):
    """
    Class for defining the customised configurations in tool_specific_configurations field,
    within the input_dir config.yaml
    Args:
        environment (str): Environment to run
    """
    environment: str = Field(...)

Within your runner parse the field into an object.

from dataclasses import dataclass
from pheval.runners.runner import PhEvalRunner
from pathlib import Path

@dataclass
class CustomPhevalRunner(PhEvalRunner):
    """CustomPhevalRunner Class."""

    input_dir: Path
    testdata_dir: Path
    tmp_dir: Path
    output_dir: Path
    config_file: Path
    version: str

    def prepare(self):
        """prepare method."""
        print("preparing")
        config = CustomisedConfigurations.parse_obj(
            self.input_dir_config.tool_specific_configuration_options
        )
        environment = config.environment

    def run(self):
        """run method."""
        print("running with custom pheval runner")

    def post_process(self):
        """post_process method."""
        print("post processing")

Post-processing methods

PhEval currently supports the benchmarking of gene, variant, and disease prioritisation results.

To benchmark these result types, PhEval parquet result files need to be generated.

PhEval can deal with the ranking and generation of these files to the correct location. However, the runner implementation must handle the extraction of essential data from the tool-specific raw results. This involves transforming them into a polars dataframe with the required columns for the benchmark type.

The columns representing essential information extracted from tool-specific output for gene, variant, and disease prioritisation are defined as follows:

Bases: Enum

Enum for different result schema formats. Attributes: GENE_RESULT_SCHEMA (pl.Schema): Schema for gene-based results. VARIANT_RESULT_SCHEMA (pl.Schema): Schema for variant-based results. DISEASE_RESULT_SCHEMA (pl.Schema): Schema for disease-based results.

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

class ResultSchema(Enum):
    """
    Enum for different result schema formats.
    Attributes:
        GENE_RESULT_SCHEMA (pl.Schema): Schema for gene-based results.
        VARIANT_RESULT_SCHEMA (pl.Schema): Schema for variant-based results.
        DISEASE_RESULT_SCHEMA (pl.Schema): Schema for disease-based results.
    """

    GENE_RESULT_SCHEMA = pl.Schema(
        {
            "gene_symbol": pl.String,
            "gene_identifier": pl.String,
            "score": pl.Float64,
            "grouping_id": pl.Utf8,
        }
    )
    VARIANT_RESULT_SCHEMA = pl.Schema(
        {
            "chrom": pl.String,
            "start": pl.Int64,
            "end": pl.Int64,
            "ref": pl.String,
            "alt": pl.String,
            "score": pl.Float64,
            "grouping_id": pl.Utf8,
        }
    )
    DISEASE_RESULT_SCHEMA = pl.Schema(
        {
            "disease_identifier": pl.String,
            "score": pl.Float64,
            "grouping_id": pl.Utf8,
        }
    )

    def validate(self, results: pl.DataFrame) -> bool:
        """
        Validate that a DataFrame follows the expected schema.
        Args:
            results (pl.DataFrame): The DataFrame to validate.
        Raises:
            ValueError: If a required column is missing or the grouping_id column contains a null value.
            TypeError: If a column exists but has an incorrect data type.
        Returns:
            bool: True if the DataFrame is valid according to the schema.
        """
        expected_schema = self.value

        if "grouping_id" in results.columns and results["grouping_id"].null_count() > 0:
            raise ValueError("'grouping_id' column should not contain null values if provided.")

        for col_name, expected_type in expected_schema.items():
            if col_name not in results.schema:
                if col_name == "grouping_id":
                    continue
                raise ValueError(f"Missing required column: {col_name}")

            if results.schema[col_name] != expected_type:
                raise TypeError(
                    f"Column '{col_name}' has type {results.schema[col_name]}, expected {expected_type}"
                )

        return True

GENE_RESULT_SCHEMA = pl.Schema({'gene_symbol': pl.String, 'gene_identifier': pl.String, 'score': pl.Float64, 'grouping_id': pl.Utf8}) class-attribute instance-attribute

VARIANT_RESULT_SCHEMA = pl.Schema({'chrom': pl.String, 'start': pl.Int64, 'end': pl.Int64, 'ref': pl.String, 'alt': pl.String, 'score': pl.Float64, 'grouping_id': pl.Utf8}) class-attribute instance-attribute

DISEASE_RESULT_SCHEMA = pl.Schema({'disease_identifier': pl.String, 'score': pl.Float64, 'grouping_id': pl.Utf8}) class-attribute instance-attribute

---

The grouping_id column is optional and is designed to handle cases where entities should be jointly ranked without being penalised. For example, in the ranking of compound heterozygous variant which occurs when two or more variants, inherited together, contribute to a phenotype. For this purpose, variants that are part of the same compound heterozygous group (e.g., within the same gene) should be assigned the same grouping_id. This ensures they are ranked as a single entity, preserving their combined significance. Variants that are not part of any compound heterozygous group should each have a unique grouping_id. This approach prevents any unintended overlap in ranking and ensures that each group or individual variant is accurately represented. The use of the grouping_id would also be suitable for the ranking and prioritisation of polygenic diseases.

Depending on whether you need to generate gene, variant, and or disease results depends on the final method called to generate the results from the polars dataframe. The methods are outlined below:

⚠️ Breaking Change (v0.5.0):
The helper method generate_pheval_result has been replaced with three separate methods for each result type:
- generate_gene_result
- generate_variant_result
- generate_disease_result
Update your runner implementation to call the appropriate method based on the type of result your tool produces.

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 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)

---

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
    )
    _write_disease_result(classified_results, output_file)

---

An example of how the method can be called is outlined here:

from pheval.post_processing.post_processing import generate_gene_result, SortOrder

generate_gene_result(
    results=pheval_gene_result,  # this is the polars dataframe containing extracted PhEval result requirements
    sort_order=SortOrder.DESCENDING,  # or can be ASCENDING - this determines in which order the scores will be ranked
    output_dir=output_directory,  # this can be accessed from the runner instance e.g., self.output_dir
    result_path=result_path  # this is the path to the tool-specific raw results file
    phenopacket_dir=phenopacket_dir # this is the path to the directory containing the phenopackets
)

Adding metadata to the results.yml

By default, PhEval will write a results.yml to the output directory supplied on the CLI.

The results.yml contains basic metadata regarding the run configuration, however, there is also the option to add customised run metadata to the results.yml in the tool_specific_configuration_options field.

To achieve this, you'll need to create a construct_meta_data() method within your runner implementation. This method is responsible for appending customised metadata to the metadata object in the form of a defined dataclass. It should return the entire metadata object once the addition is completed.

e.g.,

Defined customised metadata dataclass:

from dataclasses import dataclass

@dataclass
class CustomisedMetaData:
    customised_field: str

Example of implementation in the runner.

from dataclasses import dataclass
from pheval.runners.runner import PhEvalRunner
from pathlib import Path

@dataclass
class CustomPhevalRunner(PhEvalRunner):
    """CustomPhevalRunner Class."""

    input_dir: Path
    testdata_dir: Path
    tmp_dir: Path
    output_dir: Path
    config_file: Path
    version: str

    def prepare(self):
        """prepare method."""
        print("preparing")

    def run(self):
        """run method."""
        print("running with custom pheval runner")

    def post_process(self):
        """post_process method."""
        print("post processing")

    def construct_meta_data(self):
        """Add metadata."""
        self.meta_data.tool_specific_configuration_options = CustomisedMetaData(customised_field="customised_value")
        return self.meta_data

6. Test it.

To update your custom pheval runner implementation, you must first install the package

poetry install

Now you have to be able to run PhEval passing your custom runner as parameter. e.g.,

pheval run -i ./input_dir -t ./test_data_dir -r 'customphevalrunner' -o output_dir

The -r parameter stands for your plugin runner class name, and it must be entirely lowercase.

Output:

preparing
running
post processing