code: def vep(
      dataset: Union[Table, MatrixTable],
      config: Optional[Union[str, VEPConfig]] = None,
      block_size: int = 1000,
      name: str = "vep",
      csq: bool = False,
      tolerate_parse_error: bool = False,
  ):
      """Annotate variants with VEP.
  
      .. include:: ../_templates/req_tvariant.rst
  
      :func:`.vep` runs `Variant Effect Predictor
      <http://www.ensembl.org/info/docs/tools/vep/index.html>`__ on the
      current dataset and adds the result as a row field.
  
      Examples
      --------
  
      Add VEP annotations to the dataset:
  
      >>> result = hl.vep(dataset, "data/vep-configuration.json") # doctest: +SKIP
  
      Notes
      -----
  
      **Installation**
  
      This VEP command only works if you have already installed VEP on your
      computing environment. If you use `hailctl dataproc` to start Hail clusters,
      installing VEP is achieved by specifying the `--vep` flag. For more detailed instructions,
      see :ref:`vep_dataproc`. If you use `hailctl hdinsight`, see :ref:`vep_hdinsight`.
  
      **Spark Configuration**
  
      :func:`.vep` needs a configuration file to tell it how to run VEP. This is the ``config`` argument
      to the VEP function. If you are using `hailctl dataproc` as mentioned above, you can just use the
      default argument for ``config`` and everything will work. If you need to run VEP with Hail in other environments,
      there are detailed instructions below.
  
      The format of the configuration file is JSON, and :func:`.vep`
      expects a JSON object with three fields:
  
      - `command` (array of string) -- The VEP command line to run.  The string literal `__OUTPUT_FORMAT_FLAG__` is replaced with `--json` or `--vcf` depending on `csq`.
      - `env` (object) -- A map of environment variables to values to add to the environment when invoking the command.  The value of each object member must be a string.
      - `vep_json_schema` (string): The type of the VEP JSON schema (as produced by the VEP when invoked with the `--json` option).  Note: This is the old-style 'parseable' Hail type syntax.  This will change.
  
      Here is an example configuration file for invoking VEP release 85
      installed in `/vep` with the Loftee plugin:
  
      .. code-block:: text
  
          {
              "command": [
                  "/vep",
                  "--format", "vcf",
                  "__OUTPUT_FORMAT_FLAG__",
                  "--everything",
                  "--allele_number",
                  "--no_stats",
                  "--cache", "--offline",
                  "--minimal",
                  "--assembly", "GRCh37",
                  "--plugin", "LoF,human_ancestor_fa:/root/.vep/loftee_data/human_ancestor.fa.gz,filter_position:0.05,min_intron_size:15,conservation_file:/root/.vep/loftee_data/phylocsf_gerp.sql,gerp_file:/root/.vep/loftee_data/GERP_scores.final.sorted.txt.gz",
                  "-o", "STDOUT"
              ],
              "env": {
                  "PERL5LIB": "/vep_data/loftee"
              },
              "vep_json_schema": "Struct{assembly_name:String,allele_string:String,ancestral:String,colocated_variants:Array[Struct{aa_allele:String,aa_maf:Float64,afr_allele:String,afr_maf:Float64,allele_string:String,amr_allele:String,amr_maf:Float64,clin_sig:Array[String],end:Int32,eas_allele:String,eas_maf:Float64,ea_allele:String,ea_maf:Float64,eur_allele:String,eur_maf:Float64,exac_adj_allele:String,exac_adj_maf:Float64,exac_allele:String,exac_afr_allele:String,exac_afr_maf:Float64,exac_amr_allele:String,exac_amr_maf:Float64,exac_eas_allele:String,exac_eas_maf:Float64,exac_fin_allele:String,exac_fin_maf:Float64,exac_maf:Float64,exac_nfe_allele:String,exac_nfe_maf:Float64,exac_oth_allele:String,exac_oth_maf:Float64,exac_sas_allele:String,exac_sas_maf:Float64,id:String,minor_allele:String,minor_allele_freq:Float64,phenotype_or_disease:Int32,pubmed:Array[Int32],sas_allele:String,sas_maf:Float64,somatic:Int32,start:Int32,strand:Int32}],context:String,end:Int32,id:String,input:String,intergenic_consequences:Array[Struct{allele_num:Int32,consequence_terms:Array[String],impact:String,minimised:Int32,variant_allele:String}],most_severe_consequence:String,motif_feature_consequences:Array[Struct{allele_num:Int32,consequence_terms:Array[String],high_inf_pos:String,impact:String,minimised:Int32,motif_feature_id:String,motif_name:String,motif_pos:Int32,motif_score_change:Float64,strand:Int32,variant_allele:String}],regulatory_feature_consequences:Array[Struct{allele_num:Int32,biotype:String,consequence_terms:Array[String],impact:String,minimised:Int32,regulatory_feature_id:String,variant_allele:String}],seq_region_name:String,start:Int32,strand:Int32,transcript_consequences:Array[Struct{allele_num:Int32,amino_acids:String,biotype:String,canonical:Int32,ccds:String,cdna_start:Int32,cdna_end:Int32,cds_end:Int32,cds_start:Int32,codons:String,consequence_terms:Array[String],distance:Int32,domains:Array[Struct{db:String,name:String}],exon:String,gene_id:String,gene_pheno:Int32,gene_symbol:String,gene_symbol_source:String,hgnc_id:String,hgvsc:String,hgvsp:String,hgvs_offset:Int32,impact:String,intron:String,lof:String,lof_flags:String,lof_filter:String,lof_info:String,minimised:Int32,polyphen_prediction:String,polyphen_score:Float64,protein_end:Int32,protein_start:Int32,protein_id:String,sift_prediction:String,sift_score:Float64,strand:Int32,swissprot:String,transcript_id:String,trembl:String,uniparc:String,variant_allele:String}],variant_class:String}"
          }
  
      The configuration files used by``hailctl dataproc`` can be found at the following locations:
  
       - ``GRCh37``: ``gs://hail-us-vep/vep85-loftee-gcloud.json``
       - ``GRCh38``: ``gs://hail-us-vep/vep95-GRCh38-loftee-gcloud.json``
  
      If no config file is specified, this function will check to see if environment variable `VEP_CONFIG_URI` is set with a path to a config file.
  
      **Batch Service Configuration**
  
      If no config is specified, Hail will use the user's Service configuration parameters to find a supported VEP configuration.
      However, if you wish to use your own implementation of VEP, then see the documentation for :class:`.VEPConfig`.
  
      **Annotations**
  
      A new row field is added in the location specified by `name` with type given
      by the type given by the `json_vep_schema` (if `csq` is ``False``) or
      :class:`.tarray` of :py:data:`.tstr` (if `csq` is ``True``).
  
      If csq is ``True``, then the CSQ header string is also added as a global
      field with name ``name + '_csq_header'``.
  
      Parameters
      ----------
      dataset : :class:`.MatrixTable` or :class:`.Table`
          Dataset.
      config : :class:`str` or :class:`.VEPConfig`, optional
          Path to VEP configuration file or a VEPConfig object.
      block_size : :obj:`int`
          Number of rows to process per VEP invocation.
      name : :class:`str`
          Name for resulting row field.
      csq : :obj:`bool`
          If ``True``, annotates with the VCF CSQ field as a :py:data:`.tstr`.
          If ``False``, annotates as the `vep_json_schema`.
      tolerate_parse_error : :obj:`bool`
          If ``True``, ignore invalid JSON produced by VEP and return a missing annotation.
  
      Returns
      -------
      :class:`.MatrixTable` or :class:`.Table`
          Dataset with new row-indexed field `name` containing VEP annotations.
  
      """
  
      if isinstance(dataset, MatrixTable):
          require_row_key_variant(dataset, "vep")
          ht = dataset.select_rows().rows()
      else:
          require_table_key_variant(dataset, "vep")
          ht = dataset.select()
  
      ht = ht.distinct()
  
      backend = hl.current_backend()
      if isinstance(backend, ServiceBackend):
          with hl.TemporaryDirectory(
              prefix="qob/vep/inputs/"
          ) as vep_input_path:
              with hl.TemporaryDirectory(
                  prefix="qob/vep/outputs/"
              ) as vep_output_path:
                  annotations = _service_vep(
                      backend,
                      ht,
                      config,
                      block_size,
                      csq,
                      tolerate_parse_error,
                      vep_input_path,
                      vep_output_path,
                  )
                  annotations = annotations.checkpoint(new_temp_file())
      else:
          if config is None:
              maybe_cloud_spark_provider = guess_cloud_spark_provider()
              maybe_config = os.getenv("VEP_CONFIG_URI")
              if maybe_config is not None:
                  config = maybe_config
              elif maybe_cloud_spark_provider == "hdinsight":
                  warning(
                      "Assuming you are in a hailctl hdinsight cluster. If not, specify the config parameter to `hl.vep`."
                  )
                  config = "file:/vep_data/vep-azure.json"
              else:
                  raise ValueError(
                      "No config set and VEP_CONFIG_URI was not set."
                  )
  
          annotations = Table(
              TableToTableApply(
                  ht._tir,
                  {
                      "name": "VEP",
                      "config": config,
                      "csq": csq,
                      "blockSize": block_size,
                      "tolerateParseError": tolerate_parse_error,
                  },
              )
          ).persist()
  
      if csq:
          dataset = dataset.annotate_globals(
              **{
                  name
                  + "_csq_header": annotations.index_globals()[
                      "vep_csq_header"
                  ]
              }
          )
  
      if isinstance(dataset, MatrixTable):
          vep = annotations[dataset.row_key]
          return dataset.annotate_rows(
              **{name: vep.vep, name + "_proc_id": vep.vep_proc_id}
          )
      else:
          vep = annotations[dataset.key]
          return dataset.annotate(
              **{name: vep.vep, name + "_proc_id": vep.vep_proc_id}
          )
  
imports: import abc
  import logging
  import hail
  from collections import Counter
  import os
  from shlex import quote
  from typing import Dict, Tuple, List, Optional, Union
  from hailtop import pip_version
  from hailtop.utils import async_to_blocking
  import hailtop.batch_client
  from hailtop.config import get_deploy_config
  from hailtop import yamlx
  from hail.backend.service_backend import ServiceBackend
  from hail.typecheck import typecheck, oneof, anytype, nullable, numeric
  from hail.expr.expressions.expression_typecheck import expr_float64
  from hail.utils import FatalError
  from hail.utils.java import Env, info, warning
  from hail.utils.misc import divide_null, guess_cloud_spark_provider, new_temp_file
  from hail.matrixtable import MatrixTable
  from hail.table import Table
  from hail.ir import TableToTableApply
  from .misc import require_biallelic, require_row_key_variant, require_col_key_str, require_table_key_variant, require_alleles_field
label_desc: Use image node_id 6 for calling a global function with example usage: vep(dataset, config, block_size, name, csq, tolerate_parse_error) and returns: dataset, dataset