code: def run_vep(
      vep_cmd,
      input_file,
      block_size,
      consequence,
      tolerate_parse_error,
      part_id,
      env,
  ):
      results = []
  
      with open(input_file, "r") as inp:
          header = consume_header(inp)
          for block_id, block in enumerate(
              grouped_iterator(block_size, inp)
          ):
              n_processed = len(block)
              start_time = time.monotonic()
  
              variants = [
                  Variant.from_vcf_line(l.rstrip()) for l in block
              ]
              non_star_to_orig_variants = {
                  str(v.strip_star_allele()): v for v in variants
              }
  
              with sp.Popen(
                  vep_cmd,
                  env=env,
                  stdin=sp.PIPE,
                  stdout=sp.PIPE,
                  stderr=sp.PIPE,
                  encoding="utf-8",
              ) as proc:
                  data = f'{header}{"".join(block)}'
  
                  stdout, stderr = proc.communicate(data)
                  if stderr:
                      print(stderr)
  
                  if proc.returncode != 0:
                      raise ValueError(
                          f"VEP command {vep_cmd} failed with non-zero exit status {proc.returncode}\n"
                          f"VEP error output:\n"
                          f"{stderr}"
                      )
  
              for line in stdout.split("\n"):
                  line = line.rstrip()
  
                  if line == "" or line.startswith("#"):
                      continue
  
                  if consequence:
                      vep_v = Variant.from_vcf_line(line)
                      orig_v = non_star_to_orig_variants.get(str(vep_v))
  
                      if orig_v is None:
                          raise ValueError(
                              f"VEP output variant {vep_v} not found in original variants. VEP output is {line}"
                          )
  
                      x = CONSEQUENCE_REGEX.findall(line)
                      if x:
                          first: str = x[0]
                          result = (
                              orig_v,
                              json.dumps(first[4:].split(",")),
                              part_id,
                              block_id,
                          )
                      else:
                          print(
                              f"WARNING: No CSQ INFO field for VEP output variant {vep_v}. VEP output is {line}"
                          )
                          result = (orig_v, None, part_id, block_id)
                  else:
                      try:
                          jv = json.loads(line)
                      except json.decoder.JSONDecodeError as e:
                          msg = (
                              f"VEP failed to produce parsable JSON!\n"
                              f"json: {line}\n"
                              f"error: {e.msg}"
                          )
                          if tolerate_parse_error:
                              print(msg)
                              continue
                          raise Exception(msg) from e
  
                      variant_string = jv.get("input")
                      if variant_string is None:
                          raise ValueError(
                              f"VEP generated null variant string\n"
                              f"json: {line}\n"
                              f"parsed: {jv}"
                          )
                      v = Variant.from_vcf_line(variant_string)
                      orig_v = non_star_to_orig_variants.get(str(v))
                      if orig_v is not None:
                          result = (orig_v, line, part_id, block_id)
                      else:
                          raise ValueError(
                              f"VEP output variant {vep_v} not found in original variants. VEP output is {line}"
                          )
  
                  results.append(result)
  
              elapsed_time = time.monotonic() - start_time
              print(
                  f"processed {n_processed} variants in {round(elapsed_time, 2)}s"
              )
  
      return results
  
imports: import gzip
  import io
  import json
  import os
  import re
  import shlex
  import subprocess
  import sys
  import time
  from typing import List
label_desc: Use image node_id 4 for calling a global function with example usage: run_vep(vep_cmd, input_file, block_size, consequence, tolerate_parse_error, part_id, env) and returns: results