VCF Annotator CLI support for reading/writing stdin/stdout (#531)

Closes #529 

Tested with stdin/out and this vcf file:

```
$ curl -O https://raw.githubusercontent.com/c-BIG/NPM-sample-qc/refs/heads/master/tests/NA12878-chr14-AKT1_1000genomes-dragen-3.7.6/NA12878-chr14-AKT1.vcf.gz

$ export GA4GH_VRS_DATAPROXY_URI=seqrepo+file://$HOME/dev/data/seqrepo/2024-12-20

# existing functionality (file in, file out)
$ vrs-annotate vcf NA12878-chr14-AKT1.vcf.gz --vcf-out NA12878-chr14-AKT1.VRS-ANNOTATED.vcf.gz

# file in, stdout out
$ vrs-annotate vcf NA12878-chr14-AKT1.vcf.gz --vcf-out -

# stdin in, file out
$ cat NA12878-chr14-AKT1.vcf.gz | vrs-annotate vcf - --vcf-out NA12878-chr14-AKT1.VRS-ANNOTATED.vcf.gz

# stdin in, stdout out
$ cat NA12878-chr14-AKT1.vcf.gz | vrs-annotate vcf - --vcf-out -
```

Author: theferrit32

src/ga4gh/vrs/extras/annotator/cli.py

@@ -70,19 +70,45 @@ def _set_log_level(ctx: dict, param: str, value: _LogLevel) -> None:  # noqa: AR
     )(func)
 
 
+class PathOrDash(click.ParamType):
+    """click ParamType to support converting to Path and allowing '-' for stdin/out"""
+
+    name = "path-or-dash"
+
+    def __init__(self, **kwargs) -> None:
+        """Initialize PathOrDash"""
+        self._path_type = kwargs.pop("path_type", Path)
+        self._inner = click.Path(**kwargs, path_type=self._path_type)
+
+    def convert(
+        self, value: str, param: click.Parameter | None, ctx: click.Context | None
+    ) -> str | Path:
+        """Convert the value using click.Path if it is not '-'"""
+        if value == "-":
+            return value
+        # Use click.Path for validation
+        return self._inner.convert(value, param, ctx)
+
+
 @_cli.command(name="vcf")
 @_log_level_option
 @click.argument(
     "vcf-in",
     nargs=1,
-    type=click.Path(exists=True, readable=True, dir_okay=False, path_type=Path),
+    type=PathOrDash(
+        allow_dash=True,
+        exists=True,
+        readable=True,
+        dir_okay=False,
+        path_type=Path,
+    ),
 )
 @click.option(
     "--vcf-out",
     required=False,
-    type=click.Path(writable=True, allow_dash=False, path_type=Path),
+    type=PathOrDash(writable=True, allow_dash=True, path_type=Path),
     help=(
-        "Declare save location for output annotated VCF. At least one form of output must be declared."
+        'Declare save location for output annotated VCF (or "-" to write to stdout). At least one form of output must be declared.'
     ),
 )
 @click.option(
@@ -141,8 +167,8 @@ def _set_log_level(ctx: dict, param: str, value: _LogLevel) -> None:  # noqa: AR
     help="Suppress messages printed to stdout",
 )
 def _annotate_vcf_cli(
-    vcf_in: Path,
-    vcf_out: Path | None,
+    vcf_in: Path | str,
+    vcf_out: Path | str | None,
     pkl_out: Path | None,
     ndjson_out: Path | None,
     vrs_attributes: bool,
@@ -152,7 +178,7 @@ def _annotate_vcf_cli(
     require_validation: bool,
     silent: bool,
 ) -> None:
-    """Extract VRS objects from VCF located at VCF_IN.
+    """Extract VRS objects from VCF located at VCF_IN. VCF_IN can be "-" to read from stdin.
 
         $ vrs-annotate vcf input.vcf.gz --vcf-out output.vcf.gz --pkl-out vrs_objects.pkl
 
@@ -186,11 +212,10 @@ def _annotate_vcf_cli(
     start = timer()
     msg = f"Annotating {vcf_in} with the VCF Annotator..."
     _logger.info(msg)
-    if not silent:
-        click.echo(msg)
+
     try:
         annotator.annotate(
-            vcf_in.absolute(),
+            vcf_in,
             output_vcf_path=vcf_out,
             vrs_attributes=vrs_attributes,
             assembly=assembly,

src/ga4gh/vrs/extras/annotator/vcf.py

@@ -191,8 +191,8 @@ def _update_vcf_header(
     @use_ga4gh_compute_identifier_when(VrsObjectIdentifierIs.MISSING)
     def annotate(
         self,
-        input_vcf_path: Path,
-        output_vcf_path: Path | None = None,
+        input_vcf_path: Path | Literal["-"],
+        output_vcf_path: Path | Literal["-"] | None = None,
         vrs_attributes: bool = False,
         assembly: str = "GRCh38",
         compute_for_ref: bool = True,
@@ -202,8 +202,8 @@ def annotate(
         """Given a VCF, produce an output VCF annotated with VRS allele IDs, and/or
         additional storage outputs as implemented in a specific child class.
 
-        :param input_vcf_path: Location of input VCF
-        :param output_vcf_path: The path for the output VCF file
+        :param input_vcf_path: Location of input VCF (or "-" to read from stdin)
+        :param output_vcf_path: The path for the output VCF file (or "-" to write to stdout)
         :param vrs_attributes: If `True`, include VRS_Start, VRS_End, VRS_State
             properties in the VCF INFO field. If `False` will not include these
             properties. Only used if `output_vcf_path` is defined.
@@ -217,13 +217,17 @@ def annotate(
         :raise VCFAnnotatorError: if no output formats are selected
         """
         self.raise_for_output_args(output_vcf_path, **kwargs)
-
-        vcf = pysam.VariantFile(filename=str(input_vcf_path.absolute()))
+        # This can be pushed up to the click arg parsing too
+        pysam_in_filename = (
+            "-" if input_vcf_path == "-" else str(input_vcf_path.absolute())
+        )
+        vcf = pysam.VariantFile(filename=pysam_in_filename, mode="r")
         if output_vcf_path:
             self._update_vcf_header(vcf, compute_for_ref, vrs_attributes)
-            vcf_out = pysam.VariantFile(
-                str(output_vcf_path.absolute()), "w", header=vcf.header
+            pysam_out_filename = (
+                "-" if output_vcf_path == "-" else str(output_vcf_path.absolute())
             )
+            vcf_out = pysam.VariantFile(pysam_out_filename, mode="w", header=vcf.header)
         else:
             vcf_out = None