Truvari

https://truvari.readthedocs.io/en/latest/
Admin
Truvari is a benchmarking tool for genomic structural variants, designed to compare comparison calls...
Tokens:39,653
Snippets:243
Trust Score:9.6
Update:2 months ago
Show doc for...
Tokens:
Raw
### Initialize and Run Benchmarking

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Demonstrates how to instantiate the Bench class and execute the benchmarking process on existing VCF files.

```python
m_bench = truvari.Bench(params, base_vcf, comp_vcf, outdir)
output = m_bench.run()
```

--------------------------------

### Get variant boundaries

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/variant_record.html

Retrieves the start and end coordinates of a variant, with an option to inflate boundaries for insertions.

```python
[docs]
    def boundaries(self, ins_inflate=False):
        """
        Get the start and end of an entry

        :param `ins_inflate`: inflate INS boundaries by sv length
        :type `ins_inflate`: bool, optional

        :return: start/end
        :rtype: tuple (int, int)
        """
        start = self.start
        end = self.end
        if ins_inflate and self.var_type() == truvari.SV.INS:
            size = self.var_size()
            start -= size // 2
            end += size // 2
        return start, end
```

--------------------------------

### coords_within

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/comparisons.html

Checks if a span is within a provided range [start, end).

```APIDOC
## coords_within(qstart, qend, rstart, rend, end_within)

### Description
Returns if a span is within the provided [start, end). All coordinates are assumed to be 0-based.

### Parameters
- **qstart** (integer) - Query start position
- **qend** (integer) - Query end position
- **rstart** (integer) - Start of span
- **rend** (integer) - End of span
- **end_within** (bool) - If true, qend <= rend, else qend < rend

### Returns
- **bool** - True if the coordinates are within the span.
```

--------------------------------

### setup_logging(debug=False, stream=sys.stderr, log_format=..., show_version=False)

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/utils.html

Configures the default logging behavior for the application.

```APIDOC
## setup_logging(debug=False, stream=sys.stderr, log_format="%(asctime)s [%(levelname)s] %(message)s", show_version=False)

### Description
Configures the standard logging system. Can set log levels, output streams, and format strings.

### Parameters
- **debug** (boolean) - Optional - If True, sets log level to logging.DEBUG.
- **stream** (file handler) - Optional - The output stream for logs.
- **log_format** (string) - Optional - The format string for log messages.
- **show_version** (bool) - Optional - If True, logs the Truvari version and command line arguments.
```

--------------------------------

### Calculate Variant Distance

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/variant_record.html

Calculates the start and end distances between two variants.

```python
def distance(self, other):
        """
        Calculate the start and end distances of the pair. Negative distances
        indicate self is upstream of other

        :param `other`: Other to compare
        :type `other`: :class:`truvari.VariantRecord`

        :return: starts and ends distance
        :rtype: tuple (int, int)
        """
        astart, aend = self.boundaries()
        bstart, bend = other.boundaries()
        return astart - bstart, aend - bend
```

--------------------------------

### Execute main benchmarking process

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Entry point for command line execution that initializes parameters, runs the benchmark, and logs statistics.

```python
def bench_main(cmdargs):
    """
    Main - entry point from command line
    """
    args = parse_args(cmdargs)

    if check_params(args) or check_inputs(args):
        sys.stderr.write("Couldn't run Truvari. Please fix parameters\n")
        sys.exit(100)

    if args.bSample is None or args.cSample is None:
        if args.refine:
            sys.stderr.write("Cannot hook to `--refine` on VCFs without SAMPLE columns")
            sys.exit(100)
        skip_gt = True
    else:
        skip_gt = False

    params = truvari.VariantParams(args,
                                   skip_gt=skip_gt,
                                   short_circuit=args.short,
                                   decompose=args.no_decompose)

    m_bench = Bench(params=params,
                    base_vcf=args.base,
                    comp_vcf=args.comp,
                    outdir=args.output,
                    includebed=args.includebed,
                    extend=args.extend,
                    debug=args.debug,
                    do_logging=True)

    output = m_bench.run()

    logging.info("Stats: %s", json.dumps(output.stats_box, indent=4))
    logging.info("Finished bench")

    if args.refine:
        refine_main([args.output])
```

--------------------------------

### truvari.Bench.__init__

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Initializes the Bench object with VCF files and configuration parameters.

```APIDOC
## Constructor: truvari.Bench(params=None, base_vcf=None, comp_vcf=None, outdir=None, includebed=None, extend=0, debug=False, do_logging=False)

### Description
Initializes a new Bench instance for comparing VCF files.

### Parameters
- **params** (VariantParams) - Optional - Configuration parameters for variants.
- **base_vcf** (str) - Optional - Path to the base VCF file.
- **comp_vcf** (str) - Optional - Path to the comparison VCF file.
- **outdir** (str) - Optional - Directory path to write output files.
- **includebed** (str) - Optional - Path to a BED file for region filtering.
- **extend** (int) - Optional - Distance to extend regions.
- **debug** (bool) - Optional - Enable debug logging.
- **do_logging** (bool) - Optional - Enable standard logging.
```

--------------------------------

### Calculate Reciprocal Overlap

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/variant_record.html

Example usage of the recovl method to determine the reciprocal overlap between two variant records.

```python
>>> import truvari
>>> v = truvari.VariantFile('repo_utils/test_files/variants/input1.vcf.gz')
>>> a = next(v)
>>> b = next(v)
>>> a.recovl(b)
0
```

--------------------------------

### msa2vcf(msa, anchor_base='N')

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/msatovcf.html

Parses an MSA dictionary where keys are names and values are alignments, returning the corresponding VCF entries as a string. It expects one entry in the MSA to follow the naming convention 'ref_${chrom}:${start}-${end}' to define the VCF coordinates.

```APIDOC
## msa2vcf(msa, anchor_base='N')

### Description
Parses an MSA dictionary of {name: alignment, ...} and returns its VCF entries as a string. Assumes one entry in the MSA has the name `ref_${chrom}:${start}-${end}` which provides the VCF entries coordinates.

### Parameters
- **msa** (dict) - Required - A dictionary where keys are sample/reference names and values are alignment sequences.
- **anchor_base** (str) - Optional - The base to use as an anchor if the reference sequence starts with a gap. Defaults to 'N'.

### Returns
- **str** - A string containing the VCF entries.
```

--------------------------------

### Initialize Bench with Defaults

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Create a Bench worker using default matching parameters.

```python
m_bench = truvari.Bench()
```

--------------------------------

### Initialize Bench with Custom Parameters

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Customize benchmarking parameters by passing a VariantParams object to the Bench constructor.

```python
params = truvari.VariantParams(pctseq=0.50)
m_bench = truvari.Bench(params)
```

--------------------------------

### VariantFile.__init__

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/variant_file.html

Initializes the VariantFile wrapper for a VCF file.

```APIDOC
## VariantFile(filename, *args, params=None, **kwargs)

### Description
Initializes the VariantFile wrapper around a VCF file, allowing for iteration and region-based fetching.

### Parameters
- **filename** (str) - Required - Path to the VCF file to be opened.
- **params** (truvari.VariantParams) - Optional - Parameters to apply to all VariantRecords.
- **args** (list) - Optional - Additional positional arguments to pass to pysam.VariantFile.
- **kwargs** (dict) - Optional - Additional keyword arguments to pass to pysam.VariantFile.
```

--------------------------------

### Open files with optional Gzip support

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/utils.html

Returns a generator yielding lines from a file, automatically handling plain-text or .gz compressed files.

```python
def opt_gz_open(in_fn):
    """
    Chooses file handler for plain-text files or `*.gz` files.

    returns a generator which yields lines of the file
    """
    def gz_hdlr(fn):
        with gzip.open(fn) as fh:
            for line in fh:
                yield line.decode()

    def fh_hdlr(fn):
        with open(fn) as fh:
            yield from fh

    if in_fn.endswith('.gz'):
        return gz_hdlr(in_fn)

    return fh_hdlr(in_fn)
```

--------------------------------

### help_unknown_cmd

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/utils.html

Suggests the most similar command from a list of available commands.

```APIDOC
## help_unknown_cmd(user_cmd, avail_cmds, threshold=0.8)

### Description
Guesses the command in avail_cmds that is most similar to user_cmd based on a similarity threshold.

### Parameters
- **user_cmd** (string) - Required - The command provided by the user.
- **avail_cmds** (list of strings) - Required - The list of valid commands.
- **threshold** (float) - Optional - The maximum score (0-1) required to return a guess.
```

--------------------------------

### Execute vcf2df main process

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/vcf2df.html

Orchestrates the conversion process by invoking the appropriate parser and saving the resulting DataFrame to the specified output format.

```python
def vcf2df_main(args):
    """
    Main entry point for running DataFrame builder
    """
    args = parse_args(args)

    out = None
    if args.bench_dir:
        out = bench_dir_to_df(args.vcf, args.info,
                              args.format, args.sample, args.no_prefix, args.alleles)
    else:
        out = vcf_to_df(args.vcf, args.info, args.format,
                        args.sample, args.no_prefix, args.alleles)

    # compression -- this is not super important for most VCFs
    logging.info("Optimizing DataFrame memory")
    pre_size, post_size = optimize_df_memory(out)
    logging.info("Optimized %.2fMB to %.2fMB", pre_size / 1e6, post_size / 1e6)
    if args.parquet:
        out.to_parquet(args.output)
    else:
        joblib.dump(out, args.output, compress=args.compress)
    logging.info("Finished vcf2df")
```

--------------------------------

### Suggest commands with truvari.help_unknown_cmd

Source: https://truvari.readthedocs.io/en/latest/truvari.package.html

Guesses the most similar command from a list based on a similarity threshold.

```python
>>> import truvari
>>> truvari.help_unknown_cmd('banno', ["bench", "anno"])
'anno'
>>> truvari.help_unknown_cmd('random', ["bench", "anno"])
```

--------------------------------

### truvari.Bench.run

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Executes the benchmarking process and returns the results.

```APIDOC
## Method: run()

### Description
Runs the benchmarking process using the initialized VCF files and parameters. Returns a BenchOutput object containing the results.

### Returns
- **BenchOutput** - The resulting benchmarking output object.
```

--------------------------------

### Parse Benchmarking Arguments

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Defines the command-line interface for the bench tool, including input files, comparison thresholds, and genotype settings.

```python
def parse_args(args):
    """
    Pull the command line parameters
    """
    defaults = truvari.VariantParams()
    parser = argparse.ArgumentParser(prog="bench", description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("-b", "--base", type=str, required=True,
                        help="Baseline truth-set calls")
    parser.add_argument("-c", "--comp", type=str, required=True,
                        help="Comparison set of calls")
    parser.add_argument("-o", "--output", type=str, required=True,
                        help="Output directory")
    parser.add_argument("-f", "--reference", type=str, default=None,
                        help="Fasta used to call variants. Only needed with symbolic variants.")
    parser.add_argument("-w", "--write-resolved", action="store_true",
                        help="Replace resolved symbolic variants with their sequence")
    parser.add_argument("--short", action="store_true",
                        help="Short circuit comparisions. Faster, but fewer annotations")
    parser.add_argument("--refine", action="store_true",
                        help="Run refinement with defaults on the result (requires --reference)")
    parser.add_argument("--debug", action="store_true", default=False,
                        help="Verbose logging")

    thresg = parser.add_argument_group("Comparison Threshold Arguments")
    thresg.add_argument("-r", "--refdist", type=truvari.restricted_int, default=defaults.refdist,
                        help="Max reference distance (%(default)s)")
    thresg.add_argument("-p", "--pctseq", type=truvari.restricted_float, default=defaults.pctseq,
                        help="Min sequence similarity. Set to 0 to ignore (%(default)s)")
    thresg.add_argument("-P", "--pctsize", type=truvari.restricted_float, default=defaults.pctsize,
                        help="Min variant size similarity (%(default)s)")
    thresg.add_argument("-O", "--pctovl", type=truvari.restricted_float, default=defaults.pctovl,
                        help="Min reciprocal overlap (%(default)s)")
    thresg.add_argument("-t", "--typeignore", action="store_true", default=defaults.typeignore,
                        help="Don't compare variant types (%(default)s)")
    thresg.add_argument("--pick", type=str, default=defaults.pick, choices=PICKERS.keys(),
                        help="Number of matches reported per-call (%(default)s)")
    thresg.add_argument("-d", "--dup-to-ins", action="store_true",
                        help="Assume DUP svtypes are INS (%(default)s)")
    thresg.add_argument("-B", "--bnddist", type=int, default=defaults.bnddist,
                        help="Maximum distance allowed between BNDs (%(default)s; -1=off)")
    thresg.add_argument("-N", "--no-decompose", action="store_false",
                        help="Disallow symbolic variant decomposition to BNDs")
    thresg.add_argument("-m", "--max-resolve", type=int, default=defaults.max_resolve,
                        help="Maximum size of variant to attempt to sequence resolve (%(default)s)")
    thresg.add_argument("-C", "--chunksize", type=truvari.restricted_int, default=defaults.chunksize,
                        help="Max reference distance to compare calls (%(default)s)")
    thresg.add_argument("-n", "--no-roll", action="store_true",
                        help="Turn off rolling sequence similarity")

    genoty = parser.add_argument_group("Genotype Comparison Arguments")
    genoty.add_argument("--bSample", type=str, default=None,
                        help="Baseline calls' sample to use (first)")
    genoty.add_argument("--cSample", type=str, default=None,
                        help="Comparison calls' sample to use (first)")
```

--------------------------------

### opt_gz_open

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/utils.html

Opens a file for reading, automatically handling plain-text or gzip-compressed files.

```APIDOC
## opt_gz_open(in_fn)

### Description
Chooses the appropriate file handler for plain-text files or *.gz files and returns a generator yielding lines of the file.

### Parameters
- **in_fn** (string) - Required - The filename to open.
```

--------------------------------

### Build a region tree from VCF and BED files

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/region_vcf_iter.html

Constructs a dictionary of IntervalTree objects based on VCF contigs and optional BED file regions. Requires pysam.VariantFile objects as input.

```python
def build_region_tree(vcfA, vcfB=None, includebed=None):
    """
    Builds a dictionary of chromosome names mapped to `IntervalTree` objects
    containing regions based on the input VCF files and an optional BED file.

    Parameters
    ----------
    vcfA : pysam.VariantFile
        The primary VCF file used to define the regions and contigs.
    vcfB : pysam.VariantFile, optional
        The secondary VCF file used for comparison. If provided, its contigs
        are considered to exclude any contigs not present in `vcfA`.
        Default is None.
    includebed : str, optional
        Path to a BED file specifying regions to include. If provided,
        these regions override those defined by the VCF files. Contigs in the
        BED file not found in `vcfA` are excluded. Default is None.

    Returns
    -------
    dict
        A dictionary where keys are chromosome names (str), and values are
        `IntervalTree` objects representing the regions.

    Notes
    -----
    - If `includebed` is provided:
        - The function reads regions from the BED file and uses them to define the
          regions in `all_regions`.
        - Contigs in the BED file that are not present in `vcfA` are excluded,
          and a warning is logged.
    - Contigs in `vcfB` that are not present in `vcfA` are ignored, and a
      warning is logged.
    - If a contig in `vcfA` lacks a length definition, the program logs an
      error and exits with code 10.

    Raises
    ------
    SystemExit
        If a contig in `vcfA` has no length defined, the program terminates
        with an error message and exit code 10.

    Logging
    -------
    - Logs warnings if contigs in `vcfB` are not found in `vcfA`.
    - Logs warnings if contigs in the BED file are not found in `vcfA`.
    - Logs an error and exits if any contig in `vcfA` lacks a length definition.

    """
    contigA_set = set(vcfA.header.contigs.keys())
    contigB_set = set(vcfB.header.contigs.keys()) if vcfB else contigA_set

    all_regions = defaultdict(IntervalTree)
    excluding = contigB_set - contigA_set
    if excluding:
        logging.warning(
            "%d contigs present in comparison VCF header are not in baseline VCF.", len(excluding))

    if includebed is not None:
        all_regions, counter = read_bed_tree(includebed)
        logging.info("Including %d bed regions", counter)
        excluding = set(all_regions.keys()) - contigA_set
        if excluding:
            logging.warning(
                "Excluding %d contigs in BED not in baseline VCF.", len(excluding))
        for i in excluding:
            del all_regions[i]
        return all_regions

    for contig in contigA_set:
        name = vcfA.header.contigs[contig].name
        length = vcfA.header.contigs[contig].length
        if not length:
            logging.error(
                "Contig %s has no length definition. Fix header.", name)
            sys.exit(10)
        all_regions[name].addi(0, length + 1)
    return all_regions
```

--------------------------------

### truvari.Bench.param_dict

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Returns the current benchmarking parameters as a dictionary.

```APIDOC
## Method: param_dict()

### Description
Returns the current configuration parameters of the Bench instance as a dictionary.

### Returns
- **dict** - A dictionary containing base, comp, output, includebed, extend, and debug settings.
```

--------------------------------

### cmd_exe(cmd, stdin=None, timeout=-1, cap_stderr=True, pipefail=False)

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/utils.html

Executes a shell command with support for timeouts and output capture.

```APIDOC
## cmd_exe(cmd, stdin=None, timeout=-1, cap_stderr=True, pipefail=False)

### Description
Executes a command through the shell and captures the output, enabling process handling with timeouts.

### Parameters
- **cmd** (string) - Required - The command to execute.
- **stdin** (bytes) - Optional - Standard input to pipe to the command.
- **timeout** (int) - Optional - Timeout in minutes. -1 means no timeout.
- **cap_stderr** (boolean) - Optional - If True, captures stderr in the result.
- **pipefail** (boolean) - Optional - Set to True if the command contains pipes.
```

--------------------------------

### StatsBox Class

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

A container for tracking benchmarking counts and calculating performance metrics.

```python
class StatsBox(OrderedDict):
    """
    Make a blank stats box for counting TP/FP/FN and calculating performance
    """

    def __init__(self):
        super().__init__()
        self["TP-base"] = 0
        self["TP-comp"] = 0
        self["FP"] = 0
        self["FN"] = 0
        self["precision"] = 0
        self["recall"] = 0
        self["f1"] = 0
        self["base cnt"] = 0
        self["comp cnt"] = 0
        self["TP-comp_TP-gt"] = 0
        self["TP-comp_FP-gt"] = 0
        self["TP-base_TP-gt"] = 0
        self["TP-base_FP-gt"] = 0
        self["gt_concordance"] = 0
        self["gt_matrix"] = defaultdict(Counter)

    def calc_performance(self):
        """
        Calculate the precision/recall
        """
        if self["TP-base"] == 0 and self["FN"] == 0:
            logging.warning("No TP or FN calls in --base VCF!")
        elif self["TP-comp"] == 0 and self["FP"] == 0:
            logging.warning("No TP or FP calls in --comp VCF!")

        precision, recall, f1 = truvari.performance_metrics(
            self["TP-base"], self["TP-comp"], self["FN"], self["FP"])

        self["precision"] = precision
        self["recall"] = recall
        self["f1"] = f1
        if self["TP-comp_TP-gt"] + self["TP-comp_FP-gt"] != 0:
            self["gt_concordance"] = float(self["TP-comp_TP-gt"]) / (self["TP-comp_TP-gt"] +
                                                                     self["TP-comp_FP-gt"])

    def write_json(self, out_name):
        """
        Write stats as json to file
        """
        with open(out_name, 'w') as fout:
            fout.write(json.dumps(self, indent=4))
```

--------------------------------

### VariantParams.__init__

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/variant_params.html

Initializes the variant parameters configuration object. Users can provide an argparse.Namespace object or keyword arguments to override the default settings.

```APIDOC
## class truvari.variant_params.VariantParams

### Description
Initializes a new instance of the VariantParams class. This object holds the configuration for variant parsing and matching logic.

### Constructor
`VariantParams(args=None, **kwargs)`

### Parameters
- **args** (argparse.Namespace) - Optional - An object containing attributes that match the parameter names to override defaults.
- **kwargs** (dict) - Optional - Key-value pairs to override specific parameters. Raises a ValueError if an invalid parameter key is provided.

### Configuration Attributes
- **refdist** (int) - Distance threshold for comparing positions in the reference genome. Default: 500.
- **pctseq** (float) - Minimum percentage of sequence similarity required for a match. Default: 0.70.
- **pctsize** (float) - Minimum percentage of size similarity required for a match. Default: 0.70.
- **pctovl** (float) - Minimum percentage of reciprocal overlap required for comparing variants. Default: 0.0.
- **typeignore** (bool) - Whether to ignore variant type mismatches. Default: False.
- **sizemin** (int) - Minimum variant size to consider. Default: 50.
- **sizemax** (int) - Maximum variant size to consider. Default: 50000.
- **passonly** (bool) - Whether to only consider variants with a "PASS" filter status. Default: False.
- **pick** (str) - Strategy for picking matches (single, ac, multi). Default: "single".
```

--------------------------------

### BenchOutput Class

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Handles the generation of output files for benchmarking runs.

```python
class BenchOutput():
    """
    Makes all of the output files for a Bench.run
```

--------------------------------

### bench_dir_to_df(dir_name, *args, **kwargs)

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/vcf2df.html

Parses a Truvari benchmark directory and aggregates the results into a single pandas DataFrame.

```APIDOC
## bench_dir_to_df(dir_name, *args, **kwargs)

### Description
Parses a benchmark directory by identifying VCF files and converting them into a combined pandas DataFrame, adding a 'state' column to categorize the results.

### Parameters
- **dir_name** (str) - Required - The path to the Truvari benchmark directory.
- ***args, **kwargs** - Optional - Arguments passed directly to the underlying vcf_to_df function.

### Returns
- **pandas.DataFrame** - A concatenated DataFrame containing all parsed VCF data with an added 'state' categorical column.
```

--------------------------------

### Parse Command Line Arguments

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/stratify.html

Configures and parses command line arguments for the stratify utility, including input regions, VCF paths, and output settings.

```python
def parse_args(args):
    """
    Pull the command line parameters
    """
    parser = argparse.ArgumentParser(prog="stratify", description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("regions", metavar="BED",
                        help="Regions to process")
    parser.add_argument("in_vcf", metavar="VCF",
                        help="Truvari bench result directory or a single VCF")
    parser.add_argument("-o", "--output", metavar="OUT", default="/dev/stdout",
                        help="Output bed-like file")
    parser.add_argument("-H", "--header", action="store_true",
                        help="Input regions have header to preserve in output")
    parser.add_argument("-v", "--complement", action="store_false",
                        help="Only count variants not within region boundaries")
    parser.add_argument("--debug", action="store_true",
                        help="Verbose logging")
    args = parser.parse_args(args)
    truvari.setup_logging(args.debug, show_version=True)
    return args
```

--------------------------------

### Compress and index VCF files

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/utils.html

Sorts, compresses, and indexes a VCF file using bcftools and tabix.

```python
def compress_index_vcf(fn, fout=None, remove=True):
    """
    Compress and index a vcf. If no fout is provided, write to fn + '.gz'

    :param `fn`: filename of vcf to work on
    :type `fn`: string
    :param `fout`: output filename
    :type `fout`: string, optional
    :param `remove`: remove fn after fout is made
    :type `remove`: bool
    """
    if fout is None:
        fout = fn + '.gz'
    with tempfile.TemporaryDirectory() as tmpdirname:
        bcftools.sort(fn, "-o", fout, "-O", "z", "--temp-dir", tmpdirname, catch_stdout=False)
    pysam.tabix_index(fout, force=True, preset="vcf")
    if remove:
        os.remove(fn)
```

--------------------------------

### Suggest similar commands

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/utils.html

Guesses the most similar command from a list based on a similarity threshold.

```python
def help_unknown_cmd(user_cmd, avail_cmds, threshold=0.8):
    """
    Guess the command in avail_cmds that's most similar to user_cmd. If there is
    no guess below threshold, don't guess.

    :param `user_cmd`: user command
    :type `user_cmd`: string
    :param `avail_cmds`: available commands
    :type `avail_cmds`: list of strings
    :param `threshold`: max score (0-1) to give a guess
    :type `threshold`: float, optional

    :return: best guess from :param: `avail_cmds` or None
    :rtype: string

    Example
        >>> import truvari
        >>> truvari.help_unknown_cmd('banno', ["bench", "anno"])
        'anno'
        >>> truvari.help_unknown_cmd('random', ["bench", "anno"])
    """
    guesses = []
    for real in avail_cmds:
        score = truvari.seqsim(user_cmd, real)
        if score >= threshold:
            guesses.append((score, real))
    guesses.sort()
    if not guesses:
        return None
    return guesses[-1][1]
```

--------------------------------

### build_region_tree(vcfA, vcfB=None, includebed=None)

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/region_vcf_iter.html

Builds a dictionary of chromosome names mapped to IntervalTree objects containing regions based on the input VCF files and an optional BED file.

```APIDOC
## build_region_tree

### Description
Builds a dictionary of chromosome names mapped to `IntervalTree` objects containing regions based on the input VCF files and an optional BED file.

### Parameters
- **vcfA** (pysam.VariantFile) - Required - The primary VCF file used to define the regions and contigs.
- **vcfB** (pysam.VariantFile) - Optional - The secondary VCF file used for comparison.
- **includebed** (str) - Optional - Path to a BED file specifying regions to include.

### Returns
- **dict** - A dictionary where keys are chromosome names (str), and values are `IntervalTree` objects.
```

--------------------------------

### Build Match Matrix

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Generate a match matrix for a chunk of calls to enable custom match picking or additional checks.

```python
match_matrix = m_bench.build_matrix(base_variants, comp_variants)
```

--------------------------------

### truvari.cmd_exe

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/utils.html

Executes a shell command and returns a namedtuple containing the exit code, stdout, stderr, and execution time.

```APIDOC
## truvari.cmd_exe(cmd, stdin=None, pipefail=False, cap_stderr=True, timeout=0)

### Description
Executes a shell command and captures the output. Returns a namedtuple with fields: ret_code, stdout, stderr, and run_time.

### Parameters
- **cmd** (string) - The command to execute.
- **stdin** (string, optional) - Input to pass to the command.
- **pipefail** (boolean, optional) - Whether to use pipefail.
- **cap_stderr** (boolean, optional) - Whether to capture stderr.
- **timeout** (float, optional) - Timeout in minutes.

### Returns
- **namedtuple** - (ret_code, stdout, stderr, run_time)
```

--------------------------------

### Define match picker strategies

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Maps string identifiers to specific matching strategy functions.

```python
PICKERS = {"single": pick_single_matches,
           "ac": pick_ac_matches,
           "multi": pick_multi_matches
           }
```

--------------------------------

### Generate VCF String

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/msatovcf.html

Constructs a VCF-formatted string from a dictionary of variants and sample names.

```python
def make_vcf(variants, sample_names):
    """
    Write VCF lines - building GTs
    """
    out = StringIO()
    for var in variants:
        out.write(var)
        for sample in sample_names:
            out.write('\t')
            gt = ["0", "0"]
            if sample + '_1' in variants[var]:
                gt[0] = "1"
            if sample + '_2' in variants[var]:
                gt[1] = "1"
            out.write("|".join(gt))
        out.write('\n')
    out.seek(0)
    return out.read()
```

--------------------------------

### Run Bench on Variant Chunks

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Execute benchmarking on pre-loaded lists of VariantRecord objects.

```python
matches = m_bench.compare_calls(base_variants, comp_variants)
```

--------------------------------

### Retrieve percentage size threshold

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/variant_params.html

Returns the flat percentage size threshold or the dynamic threshold if configured.

```python
    def get_pctsize(self, size):
        """
        Returns either the flat self.pctsize or the dynamic threshold based on the point
        """
        return self.calc_dyn_thresh(size) if self.dynthresh else self.pctsize
```

--------------------------------

### read_bed_tree(filename, chrom_col=0, start_col=1, end_col=2, one_based=False, comment='#', idxfmt=None)

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/region_vcf_iter.html

Builds a dictionary of IntervalTrees for each chromosome from a tab-delimited annotation file.

```APIDOC
## read_bed_tree

### Description
Builds a dictionary of `IntervalTree` objects for each chromosome from a tab-delimited annotation file.

### Parameters
- **filename** (str) - Required - Path to the annotation file.
- **chrom_col** (int) - Optional - Column index for chromosome name.
- **start_col** (int) - Optional - Column index for start position.
- **end_col** (int) - Optional - Column index for end position.
- **one_based** (bool) - Optional - Whether the file is one-based.
- **comment** (str) - Optional - Comment character.
- **idxfmt** (str) - Optional - Index format.
```

--------------------------------

### to_key(prefix, bounds)

Source: https://truvari.readthedocs.io/en/latest/truvari.package.html

Generates a hashable key string for a variant entry.

```APIDOC
## to_key(prefix, bounds)

### Description
Turn a vcf entry into a hashable key string. Use the prefix to indicate the source VCF when consolidating multiple files' calls.

### Parameters
- **prefix** (string) - Optional - prefix
- **bounds** (bool) - Optional - use entry_boundaries

### Returns
- **string** - hashable string uniquely identifying the variant
```

--------------------------------

### Verify Sample in VCF

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Checks if a specified sample ID exists within the provided VCF file header.

```python
def check_sample(vcf_fn, sample_id=None):  # pragma: no cover
    """
    Checks that a sample is inside a vcf
    Returns True if check failed
    """
    vcf_file = truvari.VariantFile(vcf_fn)
    check_fail = False
    if sample_id is not None and sample_id not in vcf_file.header.samples:
        logging.error("Sample %s not found in vcf (%s)", sample_id, vcf_fn)
        check_fail = True
    if len(vcf_file.header.samples) != 0 and sample_id is None:
        sample_id = vcf_file.header.samples[0]
    return check_fail, sample_id
```

--------------------------------

### Comparing Variant Records

Source: https://truvari.readthedocs.io/en/latest/truvari.examples.html

Shows how to compare two VariantRecords and customize matching thresholds using VariantParams.

```python
# Given two `truvari.VariantRecords`, entry1 and entry2
match = entry1.match(entry2)
print("Entries' Sequence Similarity:", match.seqsim)
print("Entries' Size Similarity:", match.sizesim)
print("Is the match above thresholds:", match.state)
```

```python
# Disable sequence and size similarity; enable reciprocal overlap
p = truvari.VariantParams(pctseq=0, pctsize=0, pctovl=0.5)
vcf = truvari.VariantFile("input.vcf.gz", params=p)
entry1 = next(vcf)
entry2 = next(vcf)
match = entry1.match(entry2)
```

--------------------------------

### get_scalebin(x, rmin, rmax, tmin, tmax, step)

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/vcf2df.html

Scales a variable from a source range to a target range and returns the corresponding bin label and index.

```APIDOC
## get_scalebin(x, rmin, rmax, tmin, tmax, step)

### Description
Scales a numeric value from a source domain to a target domain and identifies which bin it falls into based on a specified step size.

### Parameters
- **x** (number) - Required - The value to scale.
- **rmin** (number) - Optional - Minimum of the source range.
- **rmax** (number) - Optional - Maximum of the source range.
- **tmin** (number) - Optional - Minimum of the target range.
- **tmax** (number) - Optional - Maximum of the target range.
- **step** (number) - Optional - The bin step size.

### Returns
- **tuple** - A tuple containing the bin-string and the bin index.
```

--------------------------------

### truvari.sizesim(sizeA, sizeB)

Source: https://truvari.readthedocs.io/en/latest/truvari.package.html

Calculates the size similarity percentage and the size difference between two sizes.

```APIDOC
## truvari.sizesim(sizeA, sizeB)

### Description
Calculate the size similarity percent and size diff for two sizes.

### Parameters
- **sizeA** (int) - Required - First size
- **sizeB** (int) - Required - Second size

### Returns
- (tuple) - (float, int) representing size similarity percent and size diff (A - B)
```

--------------------------------

### Calculate size similarity

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/comparisons.html

Calculates the size similarity percent and size difference for two sizes.

```python
def sizesim(sizeA, sizeB):
    """
    Calculate the size similarity percent and size diff for two sizes

    :param `sizeA`: first size
    :type `sizeA`: int
    :param `sizeB`: second size
    :type `sizeB`: int
```

--------------------------------

### Calculate performance metrics

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/utils.html

Computes precision, recall, and F1 score based on TP, FP, and FN counts.

```python
def performance_metrics(tpbase, tp, fn, fp):
    """
    Calculates precision, recall, and f1 given counts by state
    Can return None if values are zero

    :param `tpbase`: found base count
    :type `tpbase`: int
    :param `tp`: found comp count
    :type `tp`: int
    :param `fn`: missed base count
    :type `fn`: int
    :param `fp`: missed comp count
    :type `fp`: int

    :return: tuple of precision, recall, f1
    :rtype: tuple, float
    """
    base_cnt = tpbase + fn
    comp_cnt = tp + fp
    if base_cnt == 0 or comp_cnt == 0:
        return None, None, None
    precision = tp / comp_cnt
    recall = tpbase / base_cnt
    neum = recall * precision
    denom = recall + precision
    f1 = 2 * (neum / denom) if denom != 0 else None
    return precision, recall, f1
```

--------------------------------

### sizesim(other)

Source: https://truvari.readthedocs.io/en/latest/truvari.package.html

Calculates the size similarity and difference between two variant entries.

```APIDOC
## sizesim(other)

### Description
Calculate the size similarity and difference for two entries.

### Parameters
- **other** (truvari.VariantRecord) - Required - Other entry to compare with

### Returns
- **(float, int)** - size similarity and size diff (A - B)
```

--------------------------------

### Build match matrix

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Constructs a numpy matrix of MatchResults by comparing base and comparison variants.

```python
    def build_matrix(self, base_variants, comp_variants, chunk_id=0):
        """
        Builds MatchResults, returns them as a numpy matrix
        """
        if not base_variants or not comp_variants:
            raise RuntimeError(
                "Expected at least one base and one comp variant")
        match_matrix = []
        for bid, base in enumerate(base_variants):
            base_matches = []
            for cid, comp in enumerate(comp_variants):
                mat = base.match(comp)
                mat.matid = [f"{chunk_id}.{bid}", f"{chunk_id}.{cid}"]
                logging.debug("Made mat -> %s", mat)
                base_matches.append(mat)
            match_matrix.append(base_matches)

        return np.array(match_matrix)
```

--------------------------------

### Subsetting VCF to Regions

Source: https://truvari.readthedocs.io/en/latest/truvari.examples.html

Demonstrates fetching variants within specific regions defined by a BED file or an in-memory IntervalTree.

```python
for entry in vcf.fetch_bed("regions.bed"):
    print("Entry's variant type:", entry.var_type())
    print("Entry's variant size:", entry.var_size())
```

```python
from collections import defaultdict
from pyintervaltree import IntervalTree
tree = defaultdict(IntervalTree)
tree['chr1'].addi(10, 100)
tree['chr2'].addi(2000, 2200)
count = 0
for entry in vcf.fetch_regions(tree):
    count += 1
print(f"Total of {count} variants")
```

--------------------------------

### Scale variable into bins

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/vcf2df.html

Scales a numeric value from a source range to a target range and returns the corresponding bin string and index.

```python
>>> import truvari
>>> truvari.get_scalebin(4, 1, 5, 0, 20, 5)
('[15,20)', 3)
>>> truvari.get_scalebin(6, 1, 5, 0, 20, 5)
('>=20', 4)
```

--------------------------------

### Define QUALBINS list

Source: https://truvari.readthedocs.io/en/latest/truvari.package.html

A list of quality score bin strings with a step size of 10.

```python
truvari.QUALBINS = ['[0,10)', '[10,20)', '[20,30)', '[30,40)', '[40,50)', '[50,60)', '[60,70)', '[70,80)', '[80,90)', '[90,100)', '>=100']
```

--------------------------------

### Configure dual-stream logging

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/utils.html

LogFileStderr allows writing log output simultaneously to stderr and a specified file.

```python
>>> import truvari
>>> truvari.setup_logging(stream=LogFileStderr('log.txt'))
```

--------------------------------

### Execute shell commands with truvari.cmd_exe

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/utils.html

Executes a shell command and returns a namedtuple containing the exit code, standard output, standard error, and execution time.

```python
>>> import truvari
>>> ret = truvari.cmd_exe("echo 'hello world'")
>>> ret.ret_code
0
>>> ret.stdout
'hello world\n'
```

```python
>>> import truvari
>>> ret = truvari.cmd_exe("sleep 5", pipefail=True, timeout=2/60) # Error logged is caught
>>> ret.ret_code
214
```

--------------------------------

### performance_metrics

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/utils.html

Calculates precision, recall, and F1 score based on VCF comparison counts.

```APIDOC
## performance_metrics(tpbase, tp, fn, fp)

### Description
Calculates precision, recall, and f1 given counts by state. Returns None if values are zero.

### Parameters
- **tpbase** (int) - Required - Found base count.
- **tp** (int) - Required - Found comp count.
- **fn** (int) - Required - Missed base count.
- **fp** (int) - Required - Missed comp count.
```

--------------------------------

### truvari.Bench

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

The Bench class is the primary interface for performing benchmarking operations. It can be initialized with default or custom parameters and provides methods for comparing variant calls.

```APIDOC
## Class: truvari.Bench

### Description
Object to perform operations of truvari bench. It supports custom matching parameters via VariantParams.

### Methods

#### __init__(params=None)
Initializes the Bench worker. Pass a `truvari.VariantParams` object to customize matching parameters.

#### compare_calls(base_variants, comp_variants)
Runs the benchmark on a chunk of `truvari.VariantRecord` objects. Returns a list of matches.

#### build_matrix(base_variants, comp_variants)
Builds the match matrix for a chunk of calls, useful for custom match pickers or additional matching checks.
```

--------------------------------

### Validate Benchmarking Inputs

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/bench.html

Checks if the provided base and comparison VCF files contain the expected samples.

```python
def check_inputs(args):
    """
    Checks the inputs to ensure expected values are found inside of files
    Returns True if check failed
    """
    b_check, args.bSample = check_sample(args.base, args.bSample)
    c_check, args.cSample = check_sample(args.comp, args.cSample)
    return b_check or c_check
```

--------------------------------

### Implement file_zipper utility

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/matching.html

The file_zipper function coordinates iteration across multiple variant files, yielding entries in order.

```python
def file_zipper(*start_files):
    """
    Zip multiple files to yield their entries in order.

    The function takes as input tuples of (`key`, `iterable`), where:

    - `key` is an identifier (used to track which file the yielded entry comes from).
    - `iterable` is an iterable object, typically a `truvari.VariantFile`.

    The function iterates through all input files in a coordinated manner, yielding the entries in order.

    :param start_files: A variable-length argument list of tuples (`key`, `iterable`).
    :type start_files: tuple

    :yields: A tuple of (`key`, `truvari.VariantRecord`), where `key` is the file identifier and the second element is the next record from the corresponding file.
    :rtype: tuple

    :raises StopIteration: Raised when all input files have been exhausted.

    **Logs**:
        - Logs a summary of the zipping process after all files have been processed.
    """
    markers = []  # list of lists: [name, file_handler, top_entry]
    file_counts = Counter()
    for name, i in start_files:
        try:
            markers.append([name, i, next(i)])
        except StopIteration:
            # For when there are no variants in the file
            pass
```

--------------------------------

### Main Stratification Execution

Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/stratify.html

Orchestrates the stratification process by reading regions and invoking the appropriate counting method based on input type.

```python
def stratify_main(cmdargs):
    """
    stratify
    """
    args = parse_args(cmdargs)
    read_header = 0 if args.header else None
    regions = pd.read_csv(args.regions, sep='\t', header=read_header)
    r_list = regions.to_numpy().tolist()  # the methods expect lists
    if os.path.isdir(args.in_vcf):
        counts = benchdir_count_entries(args.in_vcf, r_list, args.complement)
    else:
        chroms = np.array([_[0] for _ in r_list])
        intvs = np.array([[_[1], _[2]] for _ in r_list])
        counts = count_entries(truvari.VariantFile(args.in_vcf),
                               chroms, intvs, args.complement)
        counts = pd.Series(counts, name="count").to_frame()
    if not args.complement:
        total = counts.sum(axis=0).to_frame().T
        total.to_csv(args.output, header=args.header, index=False, sep='\t')
    else:
        counts.index = regions.index
        regions = regions.join(counts)
        regions.to_csv(args.output, header=args.header, index=False, sep='\t')
    logging.info("Finished")
```
Tokens:
Raw