"""
Compression codecs for GZCM contact matrix tiles.
Benchmarks (GM12878 chr1 @ 50kb, tile_size=512, window=1Mb):
============ ======== ========= ============ =============
Codec Size Convert Access Notes
============ ======== ========= ============ =============
cmc 1.57MB 6.56s 0.12ms Best compression
cmc_zstd 2.65MB 5.64s 0.11ms Recommended (*)
bsc_cmc 1.57MB 9.63s 0.09ms Best compression + fast access
bsc 2.78MB 17.64s 0.09ms Balanced
zstd 5.75MB 3.95s 0.10ms Fastest convert
============ ======== ========= ============ =============
(*) cmc_zstd offers the best balance of compression ratio and convert speed.
bsc_cmc: CMC transforms (binarization, diagonal transform) + BSC entropy coding.
Same compression as CMC with faster access. Best overall codec for storage-constrained use.
GZCM v4 adds the ``lz4`` codec (block format, lazy ``lz4.block`` import). See
:mod:`gunz_cm.compressions.lz4` and ``specs/gzcm-v4-design.md`` §4.7.
GZCM v5.1 adds a **codec registry** with an explicit wire-format contract
(:class:`WireFormat` in :mod:`gunz_cm.compressions._protocol`). Each codec's
``wire_format`` attribute (set automatically by the registry) tells the
container writer whether to prepend the 8-byte ``(rows, cols)`` shape header
or not. This is the fix for Bug 0.3 (writer prepended the header over LZ4's
own 12-byte header, causing decode to fail). Adding a new codec is now a
single ``register_codec(...)`` call.
Examples
--------
>>> from gunz_cm.compressions import CmcZstdEncoder, CmcZstdDecoder
>>> encoder = CmcZstdEncoder(tile_size=512)
>>> encoded = encoder.encode_tile(tile_data)
>>> decoder = CmcZstdDecoder(tile_size=512)
>>> decoded = decoder.decode_tile(encoded)
>>> from gunz_cm.compressions import get_codec, list_codecs
>>> list_codecs()
['cmc', 'cmc_zstd', 'bsc', 'bsc_cmc', 'zstd', 'lz4']
>>> enc_cls, dec_cls, wire_format = get_codec("lz4")
"""
__author__ = "Yeremia Gunawan Adhisantoso"
__email__ = "adhisant@tnt.uni-hannover.de"
__license__ = "Clear BSD"
from ._protocol import Codec, WireFormat # noqa: F401 (re-exported public API)
from .cmc_encoder import CmcEncoder
from .cmc_decoder import CmcDecoder
from .zstd_encoder import ZstdEncoder
from .zstd_decoder import ZstdDecoder
from .cmc_zstd_encoder import CmcZstdEncoder
from .cmc_zstd_decoder import CmcZstdDecoder
from .bsc_encoder import BscEncoder
from .bsc_decoder import BscDecoder
from .bsc_cmc_encoder import BscCmcEncoder
from .bsc_cmc_decoder import BscCmcDecoder
from .lz4 import Lz4Encoder
from .lz4 import Lz4Decoder
# -----------------------------------------------------------------------------
# Codec registry (v5.1)
# -----------------------------------------------------------------------------
# Maps codec name -> (Encoder class, Decoder class, WireFormat). The
# container writer and reader both consult this registry instead of hardcoded
# if/elif chains. Adding a new codec is a single register_codec(...) call below.
# -----------------------------------------------------------------------------
_CODEC_REGISTRY: dict[str, tuple[type, type, WireFormat]] = {}
[docs]class UnknownCodecError(KeyError):
"""Raised by ``get_codec`` when the requested codec name is not registered."""
def __init__(self, name: str, available: list[str]):
self.name = name
self.available = available
super().__init__(
f"Unknown codec {name!r}. Available codecs: {available}. "
"Did you forget to call register_codec(...)?"
)
[docs]def register_codec(
name: str,
encoder_cls: type,
decoder_cls: type,
wire_format: WireFormat,
) -> None:
"""Register a codec in the global registry.
Parameters
----------
name : str
Codec identifier used in ``convert_to_gzcm(..., compression=name)``.
encoder_cls, decoder_cls : type
Subclasses implementing :class:`Codec`.
wire_format : WireFormat
Whether the writer should prepend the 8-byte shape header.
For codecs whose encoder adds its own header (e.g. LZ4), use
:attr:`WireFormat.SELF_DESCRIBING`. Otherwise use
:attr:`WireFormat.OPAQUE_PAYLOAD`.
"""
_CODEC_REGISTRY[name] = (encoder_cls, decoder_cls, wire_format)
[docs]def get_codec(name: str) -> tuple[type, type, WireFormat]:
"""Look up a registered codec by name.
Returns
-------
(Encoder class, Decoder class, WireFormat)
Callers instantiate the encoder/decoder with ``tile_size=...`` and
use ``wire_format`` to decide whether to prepend a shape header.
Raises
------
UnknownCodecError
If ``name`` is not in the registry at all.
CodecUnavailableError
If ``name`` is registered but the codec requires a native
binary (CMC, BSC, ...) that is not installed on this machine.
Use :func:`codec_available` to probe before requesting.
"""
if name not in _CODEC_REGISTRY:
raise UnknownCodecError(name, sorted(_CODEC_REGISTRY))
enc_cls, dec_cls, wire_format = _CODEC_REGISTRY[name]
if not _probe_codec_availability(name, enc_cls):
raise CodecUnavailableError(name, reason=unavailable_reason(name, enc_cls))
return enc_cls, dec_cls, wire_format
[docs]def list_codecs() -> list[str]:
"""Return the names of all registered codecs (sorted).
Includes codecs that depend on native binaries (e.g., ``cmc``,
``bsc``) even when those binaries are missing. Use
:func:`list_available_codecs` to filter by availability, and
:func:`codec_available` to probe a single codec.
"""
return sorted(_CODEC_REGISTRY)
[docs]class CodecUnavailableError(RuntimeError):
"""Raised when a codec is registered but its native binary is missing.
Attributes
----------
name : str
Codec name (e.g., ``"cmc"``).
reason : str
Human-readable reason (e.g., the FileNotFoundError message).
"""
def __init__(self, name: str, *, reason: str):
self.name = name
self.reason = reason
super().__init__(
f"Codec {name!r} is registered but its native binary is not "
f"available on this machine: {reason}. "
"Use codec_available(name) to probe before requesting a codec, "
"or list_available_codecs() to enumerate only what works. "
"To resolve, install the binary (set GUNZ_CM_CMC_DIR / "
"GUNZ_CM_BSC_BIN), or pick a different codec (zstd, lz4)."
)
# Why: cmc, cmc_zstd, bsc_cmc depend on the CMC third-party library
# (GUNZ_CM_CMC_DIR); bsc depends on the `bsc` CLI. When those binaries
# are missing, the picker already swallows the FileNotFoundError -- but
# list_codecs()/get_codec() advertise them as if available. These probes
# let caller code enumerate "what actually works here" without forcing a
# user to discover the missing binary by hitting a FileNotFoundError.
# Crucially, get_codec() still raises only at use-time, so an explicit
# user choice gets a clear error rather than a silent skip.
def _probe_codec_availability(name: str, encoder_cls: type) -> bool:
"""Return True if the codec's runtime dependencies are satisfied.
Pure-Python codecs (zstd, lz4) always return True. Native-binary
codecs (cmc, cmc_zstd, bsc_cmc, bsc) return False if the required
directory / binary is missing. We probe the env-var resolver rather
than instantiating the encoder, because instantiation is cheap but
first-use triggers a sys.path side effect that we don't want hidden
inside a probe.
"""
if name in {"zstd", "lz4"}:
return True
if name in {"cmc", "cmc_zstd", "bsc_cmc"}:
from . import _paths
try:
_paths.resolve_cmc_dir()
except FileNotFoundError:
return False
return True
if name == "bsc":
from . import _paths
try:
_paths.resolve_bsc_binary()
except FileNotFoundError:
return False
return True
# Unknown / custom codecs are presumed pure-Python (no native
# binary required). Probing them by instantiation would side-effect
# on first use (e.g. CMC's _ensure_loaded sys.path mutation); the
# safer default is "available" -- if the codec turns out to need
# a binary, the codec's own encode/decode will raise. This matches
# the user's directive: "cmc is to be ignored if not detected but
# do not show error unless codec is chosen" -- meaning the probe
# must not pre-judge codecs it does not know about.
return True
[docs]def unavailable_reason(name: str, encoder_cls: type | None = None) -> str:
"""Human-readable reason a codec is unavailable, for error messages."""
from . import _paths
if name in {"cmc", "cmc_zstd", "bsc_cmc"}:
try:
_paths.resolve_cmc_dir()
except FileNotFoundError as exc:
return str(exc)
return "available"
if name == "bsc":
try:
_paths.resolve_bsc_binary()
except FileNotFoundError as exc:
return str(exc)
return "available"
return f"unknown reason (codec {name!r} is not native)"
[docs]def codec_available(name: str) -> bool:
"""Return True if the named codec's runtime dependencies are satisfied.
A codec that is not registered at all returns False (it is
``unknown``, not ``unavailable``; the distinction lets the caller
decide whether to register it first or just skip it).
"""
if name not in _CODEC_REGISTRY:
return False
return _probe_codec_availability(name, _CODEC_REGISTRY[name][0])
[docs]def list_available_codecs() -> list[str]:
"""Return the names of registered codecs whose binaries are present.
Excludes codecs that depend on a missing native binary (cmc /
cmc_zstd / bsc_cmc when GUNZ_CM_CMC_DIR is unset; bsc when neither
GUNZ_CM_BSC_BIN nor `bsc` on PATH resolves).
"""
return sorted(name for name in _CODEC_REGISTRY if codec_available(name))
# Auto-register the six built-in codecs. Adding a new codec = add one line here.
register_codec("cmc", CmcEncoder, CmcDecoder, WireFormat.OPAQUE_PAYLOAD)
register_codec("zstd", ZstdEncoder, ZstdDecoder, WireFormat.OPAQUE_PAYLOAD)
register_codec("cmc_zstd", CmcZstdEncoder, CmcZstdDecoder, WireFormat.OPAQUE_PAYLOAD)
register_codec("bsc", BscEncoder, BscDecoder, WireFormat.OPAQUE_PAYLOAD)
register_codec("bsc_cmc", BscCmcEncoder, BscCmcDecoder, WireFormat.OPAQUE_PAYLOAD)
register_codec("lz4", Lz4Encoder, Lz4Decoder, WireFormat.SELF_DESCRIBING)
__all__ = [
# Codec protocol (v5.1)
"Codec",
"WireFormat",
"register_codec",
"get_codec",
"list_codecs",
"list_available_codecs",
"codec_available",
"unavailable_reason",
"UnknownCodecError",
"CodecUnavailableError",
# Codec classes
"CmcEncoder",
"CmcDecoder",
"ZstdEncoder",
"ZstdDecoder",
"CmcZstdEncoder",
"CmcZstdDecoder",
"BscEncoder",
"BscDecoder",
"BscCmcEncoder",
"BscCmcDecoder",
"Lz4Encoder",
"Lz4Decoder",
]