### 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") ``` -------------------------------- ### Copy VariantRecord Source: https://truvari.readthedocs.io/en/latest/_modules/truvari/variant_record.html Creates a copy of the current VariantRecord instance. ```python def copy(self): """ Copies this truvari.VariantRecord """ return VariantRecord(self._record.copy(), params=self.params) ```