Wire up CRAM 3.1 codecs for reading. (#1736)

* Wire up CRAM 3.1 codecs for reading!

Author: cmnbroad

src/main/java/htsjdk/beta/codecs/hapref/fasta/FASTACodecV1_0.java

@@ -12,9 +12,15 @@
 import htsjdk.io.IOPath;
 import htsjdk.beta.plugin.HtsVersion;
 import htsjdk.beta.plugin.hapref.HaploidReferenceFormats;
+import htsjdk.samtools.util.BlockCompressedStreamConstants;
 import htsjdk.samtools.util.FileExtensions;
+import htsjdk.samtools.util.IOUtil;
 import htsjdk.utils.ValidationUtils;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.zip.GZIPInputStream;
+
 /**
  * The v1.0 FASTA codec.
  */
@@ -32,20 +38,26 @@ public String getFileFormat() {
     }
 
     @Override
-    public int getSignatureLength() {
-        return 1;
-    }
+    public int getSignatureLength() { return BlockCompressedStreamConstants.MAX_COMPRESSED_BLOCK_SIZE; }
 
     @Override
     public boolean canDecodeSignature(final SignatureStream signatureStream, final String sourceName) {
-        ValidationUtils.nonNull(signatureStream, "signatureStream");
+        ValidationUtils.nonNull(signatureStream, "input signatureStream may notbe null");
         ValidationUtils.nonNull(sourceName, "sourceName");
-        int ch = signatureStream.read();
-        if (ch == -1) {
-            throw new HtsjdkIOException(
-                    String.format("Codec %s failed probing signature for resource %s", this.getDisplayName(), sourceName));
+
+        try {
+            final InputStream wrappedInputStream = IOUtil.isGZIPInputStream(signatureStream) ?
+                    new GZIPInputStream(signatureStream) :
+                    signatureStream;
+            int ch = wrappedInputStream.read();
+            if (ch == -1) {
+                throw new HtsjdkIOException(
+                        String.format("Codec %s failed probing signature for resource %s", this.getDisplayName(), sourceName));
+            }
+            return ((char) ch) == '>';  // for FASTA, this is all we have to go on...
+        } catch (IOException e) {
+            throw new HtsjdkIOException(String.format("Failure reading signature from stream for %s", sourceName), e);
         }
-        return ((char) ch) == '>';  // for FASTA, this is all we have to go on...
     }
 
     @Override

src/main/java/htsjdk/beta/codecs/reads/cram/cramV3_1/CRAMCodecV3_1.java

@@ -0,0 +1,61 @@
+package htsjdk.beta.codecs.reads.cram.cramV3_1;
+
+import htsjdk.beta.codecs.reads.cram.CRAMCodec;
+import htsjdk.beta.codecs.reads.cram.CRAMDecoder;
+import htsjdk.beta.codecs.reads.cram.CRAMEncoder;
+import htsjdk.beta.exception.HtsjdkIOException;
+import htsjdk.beta.io.bundle.Bundle;
+import htsjdk.beta.io.bundle.SignatureStream;
+import htsjdk.beta.plugin.HtsVersion;
+import htsjdk.beta.plugin.reads.ReadsDecoderOptions;
+import htsjdk.beta.plugin.reads.ReadsEncoderOptions;
+import htsjdk.samtools.cram.structure.CramHeader;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+/**
+ * CRAM v3.1 codec
+ */
+public class CRAMCodecV3_1 extends CRAMCodec {
+    public static final HtsVersion VERSION_3_1 = new HtsVersion(3, 1, 0);
+    private static final String CRAM_MAGIC_3_1 = new String(CramHeader.MAGIC) + "\3\1";
+
+    @Override
+    public HtsVersion getVersion() {
+        return VERSION_3_1;
+    }
+
+    @Override
+    public int getSignatureLength() {
+        return CRAM_MAGIC_3_1.length();
+    }
+
+    @Override
+    public boolean canDecodeSignature(final SignatureStream signatureStream, final String sourceName) {
+        try {
+            final byte[] signatureBytes = new byte[getSignatureLength()];
+            final int numRead = signatureStream.read(signatureBytes);
+            if (numRead < getSignatureLength()) {
+                throw new HtsjdkIOException(String.format("Failure reading content from stream for %s", sourceName));
+            }
+            return Arrays.equals(signatureBytes, getSignatureString().getBytes());
+        } catch (IOException e) {
+            throw new HtsjdkIOException(String.format("Failure reading content from stream for %s", sourceName));
+        }
+    }
+
+    @Override
+    public CRAMDecoder getDecoder(final Bundle inputBundle, final ReadsDecoderOptions readsDecoderOptions) {
+        return new CRAMDecoderV3_1(inputBundle, readsDecoderOptions);
+    }
+
+    @Override
+    public CRAMEncoder getEncoder(final Bundle outputBundle, final ReadsEncoderOptions readsEncoderOptions) {
+        return new CRAMEncoderV3_1(outputBundle, readsEncoderOptions);
+    }
+
+    @Override
+    protected String getSignatureString() { return CRAM_MAGIC_3_1; }
+
+}

src/main/java/htsjdk/beta/codecs/reads/cram/cramV3_1/CRAMDecoderV3_1.java

@@ -0,0 +1,31 @@
+package htsjdk.beta.codecs.reads.cram.cramV3_1;
+
+import htsjdk.beta.codecs.reads.cram.CRAMDecoder;
+import htsjdk.beta.io.bundle.Bundle;
+import htsjdk.beta.io.bundle.BundleResourceType;
+import htsjdk.beta.plugin.HtsVersion;
+import htsjdk.beta.plugin.reads.ReadsDecoderOptions;
+
+/**
+ * CRAM v3.1 decoder.
+ */
+public class CRAMDecoderV3_1 extends CRAMDecoder {
+
+    /**
+     * Create a new CRAM v3.1 decoder. The primary resource in the input
+     * bundle must have content type {@link BundleResourceType#CT_ALIGNED_READS} (to find a decoder for a bundle,
+     * see {@link htsjdk.beta.plugin.registry.ReadsResolver}).
+     *
+     * @param bundle input {@link Bundle} to decode
+     * @param readsDecoderOptions {@link ReadsDecoderOptions} to use
+     */
+    public CRAMDecoderV3_1(final Bundle bundle, final ReadsDecoderOptions readsDecoderOptions) {
+        super(bundle, readsDecoderOptions);
+    }
+
+    @Override
+    public HtsVersion getVersion() {
+        return CRAMCodecV3_1.VERSION_3_1;
+    }
+
+}

src/main/java/htsjdk/beta/codecs/reads/cram/cramV3_1/CRAMEncoderV3_1.java

@@ -0,0 +1,33 @@
+package htsjdk.beta.codecs.reads.cram.cramV3_1;
+
+import htsjdk.beta.codecs.reads.cram.CRAMEncoder;
+import htsjdk.beta.io.bundle.Bundle;
+import htsjdk.beta.io.bundle.BundleResourceType;
+import htsjdk.beta.plugin.HtsVersion;
+import htsjdk.beta.plugin.reads.ReadsEncoderOptions;
+import htsjdk.samtools.cram.CRAMException;
+
+/**
+ * CRAM v3.1 encoder.
+ */
+public class CRAMEncoderV3_1 extends CRAMEncoder {
+
+    /**
+     * Create a new CRAM v3.1 encoder for the given output bundle. The primary resource in the
+     * bundle must have content type {@link BundleResourceType#CT_ALIGNED_READS} (to find an encoder for a bundle,
+     * see {@link htsjdk.beta.plugin.registry.ReadsResolver}).
+     *
+     * @param outputBundle output {@link Bundle} to encode
+     * @param readsEncoderOptions {@link ReadsEncoderOptions} to use
+     */
+    public CRAMEncoderV3_1(final Bundle outputBundle, final ReadsEncoderOptions readsEncoderOptions) {
+        super(outputBundle, readsEncoderOptions);
+        throw new CRAMException("CRAM v3.1 encoding is not yet supported");
+    }
+
+    @Override
+    public HtsVersion getVersion() {
+        return CRAMCodecV3_1.VERSION_3_1;
+    }
+
+}

src/main/java/htsjdk/beta/plugin/registry/HtsCodecResolver.java

@@ -373,7 +373,11 @@ private SignatureStream getIOPathSignatureProbingStream(
         return bundleResource.getSignatureStream(streamPrefixSize);
     }
 
-    private List<C> filterByVersion(final List<C> candidateCodecs, final HtsVersion htsVersion) {
+    /**
+     * Filter the candidate codecs, returning only those that match the requested version.
+     * Only called for encoding.
+     */
+    protected List<C> filterByVersion(final List<C> candidateCodecs, final HtsVersion htsVersion) {
         ValidationUtils.nonNull(htsVersion, "htsVersion");
         if (candidateCodecs.isEmpty()) {
             return candidateCodecs;

src/main/java/htsjdk/beta/plugin/registry/HtsDefaultRegistry.java

@@ -17,7 +17,7 @@ public class HtsDefaultRegistry {
     /**
      * statically populate the default registry with any codecs on the classpath
      */
-    static {ServiceLoader.load(HtsCodec .class).forEach(htsDefaultCodecRegistry::registerCodec);}
+    static {ServiceLoader.load(HtsCodec.class).forEach(htsDefaultCodecRegistry::registerCodec);}
 
     /**
      * Grt the {@link HaploidReferenceResolver} resolver for this registry.

src/main/java/htsjdk/beta/plugin/registry/ReadsResolver.java

@@ -1,5 +1,6 @@
 package htsjdk.beta.plugin.registry;
 
+import htsjdk.beta.codecs.reads.cram.cramV3_1.CRAMCodecV3_1;
 import htsjdk.beta.exception.HtsjdkException;
 import htsjdk.beta.exception.HtsjdkPluginException;
 import htsjdk.beta.plugin.HtsVersion;
@@ -11,9 +12,13 @@
 import htsjdk.beta.plugin.reads.ReadsDecoderOptions;
 import htsjdk.beta.plugin.reads.ReadsEncoder;
 import htsjdk.beta.plugin.reads.ReadsEncoderOptions;
+import htsjdk.beta.plugin.reads.ReadsFormats;
 import htsjdk.io.IOPath;
 import htsjdk.utils.ValidationUtils;
 
+import java.util.List;
+import java.util.stream.Collectors;
+
 /**
  * Class with methods for resolving inputs and outputs to reads encoders and decoders.
  * <p>
@@ -204,6 +209,30 @@ public ReadsEncoder getReadsEncoder(
                 .getEncoder(outputBundle, readsEncoderOptions);
     }
 
+    /**
+     * Temporarily override to remove the CRAM 3.1 codec from the list of candidate codecs when the request is for
+     * the newest version, since it has no write implementation yet.
+     */
+    @Override
+    protected List<ReadsCodec> filterByVersion(final List<ReadsCodec> candidateCodecs, final HtsVersion htsVersion) {
+        final List<ReadsCodec> preFilteredCodecs;
+        if (htsVersion.equals(HtsVersion.NEWEST_VERSION)) {
+            // if the request is for the newest version, then pre-filter out the CRAM 3.1 codec since it has no
+            // write implementation yet, and then delegate to the superclass to let it find the newest version among
+            // the remaining codecs
+            preFilteredCodecs = candidateCodecs.stream().filter(
+                    c -> !(c.getFileFormat().equals(ReadsFormats.CRAM)
+                            && c.getVersion().equals(CRAMCodecV3_1.VERSION_3_1)))
+                    .collect(Collectors.toList());
+            final HtsVersion newestVersion = preFilteredCodecs.stream()
+                    .map(c -> c.getVersion())
+                    .reduce(candidateCodecs.get(0).getVersion(),
+                            (HtsVersion a, HtsVersion b) -> a.compareTo(b) > 0 ? a : b);
+            return candidateCodecs.stream().filter(
+                    c -> c.getVersion().equals(newestVersion)).collect(Collectors.toList());
+        } else {
+            preFilteredCodecs = candidateCodecs;
+        }
+        return super.filterByVersion(preFilteredCodecs, htsVersion);
+    }
 }
-
-

src/main/java/htsjdk/samtools/cram/common/CramVersions.java

@@ -6,10 +6,12 @@
 public final class CramVersions {
     public static final CRAMVersion CRAM_v2_1 = new CRAMVersion(2, 1);
     public static final CRAMVersion CRAM_v3 = new CRAMVersion(3, 0);
+    public static final CRAMVersion CRAM_v3_1 = new CRAMVersion(3, 1);
 
     final static Set<CRAMVersion> supportedCRAMVersions = new HashSet<CRAMVersion>() {{
         add(CRAM_v2_1);
         add(CRAM_v3);
+        add(CRAM_v3_1);
     }};
 
     /**

src/main/java/htsjdk/samtools/cram/compression/BZIP2ExternalCompressor.java

@@ -25,6 +25,7 @@
 package htsjdk.samtools.cram.compression;
 
 import htsjdk.samtools.cram.io.InputStreamUtils;
+import htsjdk.samtools.cram.structure.CRAMCodecModelContext;
 import htsjdk.samtools.cram.structure.block.BlockCompressionMethod;
 import htsjdk.samtools.util.IOUtil;
 import htsjdk.samtools.util.RuntimeIOException;
@@ -42,7 +43,7 @@ public BZIP2ExternalCompressor() {
     }
 
     @Override
-    public byte[] compress(final byte[] data) {
+    public byte[] compress(final byte[] data, final CRAMCodecModelContext unused_contextModel) {
         final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
         try (final BZip2CompressorOutputStream bos = new BZip2CompressorOutputStream(byteArrayOutputStream)) {
             IOUtil.copyStream(new ByteArrayInputStream(data), bos);

src/main/java/htsjdk/samtools/cram/compression/CompressionUtils.java

@@ -175,4 +175,22 @@ public static ByteBuffer wrap(final byte[] inputBytes){
     public static ByteBuffer slice(final ByteBuffer inputBuffer){
         return inputBuffer.slice().order(ByteOrder.LITTLE_ENDIAN);
     }
+
+    /**
+     * Return a byte array with a size that matches the limit of the provided ByteBuffer. If the ByteBuffer is
+     * backed by a byte array that matches the limit of the ByteBuffer, the backing array will be returned directly.
+     * Otherwise, copy the contents of the ByteBuffer into a new byte array and return the new byte array.
+     * @param buffer input ByteBuffer which is the source of the byte array
+     * @return A byte array. If the ByteBuffer is backed by a byte array that matches the limit of the ByteBuffer,
+     * return the backing array directly. Otherwise, copy the contents of the ByteBuffer into a new byte array.
+     */
+    public static byte[] toByteArray(final ByteBuffer buffer) {
+        if (buffer.hasArray() && buffer.arrayOffset() == 0 && buffer.array().length == buffer.limit()) {
+            return buffer.array();
+        }
+
+        final byte[] bytes = new byte[buffer.limit() - buffer.arrayOffset()];
+        buffer.get(bytes);
+        return bytes;
+    }
 }