Suppress stderr printing when sam indices are older than sam file

fgpyo/io/__init__.py

@@ -44,9 +44,12 @@
 import gzip
 import io
 import os
+import sys
+from contextlib import contextmanager
 from pathlib import Path
 from typing import IO
 from typing import Any
+from typing import Generator
 from typing import Iterable
 from typing import Iterator
 from typing import Set
@@ -196,3 +199,30 @@ def write_lines(path: Path, lines_to_write: Iterable[Any]) -> None:
         for line in lines_to_write:
             writer.write(str(line))
             writer.write("\n")
+
+
+@contextmanager
+def redirect_to_dev_null(file_num: int) -> Generator[None, None, None]:
+    """A context manager that redirects output of file handle to /dev/null
+
+    Args:
+        file_num: number of filehandle to redirect.
+    """
+    # open /dev/null for writing
+    f_devnull = os.open(os.devnull, os.O_RDWR)
+    # save old file descriptor and redirect stderr to /dev/null
+    save_stderr = os.dup(file_num)
+    os.dup2(f_devnull, file_num)
+
+    yield
+
+    # restore file descriptor and close devnull
+    os.dup2(save_stderr, file_num)
+    os.close(f_devnull)
+
+
+@contextmanager
+def suppress_stderr() -> Generator[None, None, None]:
+    """A context manager that redirects output of stderr to /dev/null"""
+    with redirect_to_dev_null(file_num=sys.stderr.fileno()):
+        yield

fgpyo/sam/__init__.py

@@ -169,6 +169,7 @@
 from pysam import AlignmentFile as SamFile
 from pysam import AlignmentHeader as SamHeader
 
+import fgpyo.io
 from fgpyo.collections import PeekableIterator
 
 SamPath = Union[IO[Any], Path, str]
@@ -259,8 +260,11 @@ def _pysam_open(
     if unmapped and open_for_reading:
         kwargs["check_sq"] = False
 
-    # Open it!
-    return pysam.AlignmentFile(path, **kwargs)
+    # Open it alignment file, suppressing stderr in case index files are older than SAM file
+    with fgpyo.io.suppress_stderr():
+        alignment_file = pysam.AlignmentFile(path, **kwargs)
+    # now restore stderr and return the alignment file
+    return alignment_file
 
 
 def reader(

fgpyo/sam/builder.py

@@ -530,7 +530,6 @@ def to_path(
                 path = Path(fp.name)
 
         with NamedTemporaryFile(suffix=".bam", delete=True) as fp:
-
             with sam.writer(
                 fp.file, header=self._samheader, file_type=sam.SamFileType.BAM  # type: ignore
             ) as writer:

fgpyo/sam/tests/test_sam.py

@@ -280,10 +280,12 @@ def test_is_indel() -> None:
     indels = [op for op in CigarOp if op.is_indel]
     assert indels == [CigarOp.I, CigarOp.D]
 
+
 def test_is_clipping() -> None:
     clips = [op for op in CigarOp if op.is_clipping]
     assert clips == [CigarOp.S, CigarOp.H]
 
+
 def test_isize() -> None:
     builder = SamBuilder()
     r1, r2 = builder.add_pair(chrom="chr1", start1=100, cigar1="115M", start2=250, cigar2="40M")

fgpyo/tests/test_read_structure.py

@@ -124,7 +124,6 @@ def test_read_structure_variable_once_and_only_once_last_segment_exception(strin
 def test_read_structure_from_segments_reset_offset(
     segments: Tuple[ReadSegment, ...], expected: Tuple[ReadSegment, ...]
 ) -> None:
-
     read_structure = ReadStructure.from_segments(segments=segments, reset_offsets=True)
     assert read_structure.segments == expected
 

fgpyo/vcf/__init__.py

@@ -48,8 +48,6 @@
 order via the :func:`~VariantBuilder.to_sorted_list()` method.
 
 """
-import os
-import sys
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Generator
@@ -61,44 +59,29 @@
 from pysam import VariantFile as VcfWriter
 from pysam import VariantHeader
 
+import fgpyo.io
+
 """The valid base classes for opening a VCF file."""
 VcfPath = Union[Path, str, TextIO]
 
 
-@contextmanager
-def redirect_dev_null(file_num: int = sys.stderr.fileno()) -> Generator[None, None, None]:
-    """A context manager that redirects output of file handle to /dev/null
-
-    Args:
-        file_num: number of filehandle to redirect. Uses stderr by default
-    """
-    # open /dev/null for writing
-    f_devnull = os.open(os.devnull, os.O_RDWR)
-    # save old file descriptor and redirect stderr to /dev/null
-    save_stderr = os.dup(file_num)
-    os.dup2(f_devnull, file_num)
-
-    yield
-
-    # restore file descriptor and close devnull
-    os.dup2(save_stderr, file_num)
-    os.close(f_devnull)
-
-
 @contextmanager
 def reader(path: VcfPath) -> Generator[VcfReader, None, None]:
     """Opens the given path for VCF reading
 
     Args:
         path: the path to a VCF, or an open file handle
     """
-    with redirect_dev_null():
-        # to avoid spamming log about index older than vcf, redirect stderr to /dev/null: only
-        # when first opening the file
-        _reader = VariantFile(path, mode="r")  # type: ignore
-    # now stderr is back, so any later stderr messages will go through
-    yield _reader
-    _reader.close()
+    if isinstance(path, (str, Path, TextIO)):
+        with fgpyo.io.suppress_stderr():
+            # to avoid spamming log about index older than vcf, redirect stderr to /dev/null: only
+            # when first opening the file
+            _reader = VariantFile(path, mode="r")  # type: ignore
+        # now stderr is back, so any later stderr messages will go through
+        yield _reader
+        _reader.close()
+    else:
+        raise TypeError(f"Cannot open '{type(path)}' for VCF reading.")
 
 
 @contextmanager