"""
CMC Transforms + Zstd Decoder for GZCM v3 compression.
Combines CMC's domain-specific transforms (diagonal transform, binarization)
with Zstd entropy coding for faster decode than pure CMC.
Examples
--------
"""
__author__ = "Yeremia Gunawan Adhisantoso"
__email__ = "adhisant@tnt.uni-hannover.de"
__license__ = "Clear BSD"
import typing as t
import zlib
import numpy as np
try:
import zstandard as zstd
HAS_ZSTD = True
except ImportError:
HAS_ZSTD = False
from . import _paths
# Module-level registry of CMC+Zstd decoder functions. See ``cmc_encoder.py``
# for the rationale (DOP Phase 3, v2.20.0).
_FUNCS: dict[str, t.Any] = {}
_LOADED: bool = False
def _ensure_loaded() -> None:
"""Lazily resolve CMC+Zstd decoder functions on first access.
See ``cmc_encoder.py`` for the full rationale.
"""
global _LOADED
if _LOADED:
return
_cmc = _paths.load_cmc_zstd_decoder()
_FUNCS["debinarize_rc_bin_split_v2"] = _cmc["debinarize_rc_bin_split_v2"]
_FUNCS["reverse_diag_transform_mode0"] = _cmc["reverse_diag_transform_mode0"]
_LOADED = True
def __getattr__(name: str) -> t.Any:
"""PEP 562 module-level lazy attribute access for CMC+Zstd decoder functions."""
if name in {"debinarize_rc_bin_split_v2", "reverse_diag_transform_mode0"}:
_ensure_loaded()
return _FUNCS[name]
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
[docs]class CmcZstdDecoder:
"""CMC Transforms + Zstd decoder for contact matrix tiles.
Uses Zstd decompression then reverses CMC's domain-specific transforms.
Parameters
----------
tile_size : int, default=256
Tile size for block processing.
resolution : int, default=50000
Hi-C resolution in bp.
dtype : np.dtype, default=np.uint32
Data type for decoded tiles.
Examples
--------
"""
def __init__(
self,
tile_size: int = 256,
resolution: int = 50000,
dtype: np.dtype = np.uint32,
):
"""
Examples
--------
"""
self.tile_size = tile_size
self.resolution = resolution
self.dtype = np.dtype(dtype)
[docs] def decode_tile(self, payload: bytes) -> np.ndarray:
"""Decode a single compressed tile.
Parameters
----------
payload : bytes
Compressed bitstream (shape info + encoded data).
Returns
-------
np.ndarray
Decoded contact matrix tile.
Examples
--------
"""
shape = np.frombuffer(payload[:8], dtype=np.int32)
encoded_data = payload[8:]
if HAS_ZSTD:
ctx = zstd.ZstdDecompressor()
data = ctx.decompress(encoded_data)
else:
data = zlib.decompress(encoded_data)
_ensure_loaded()
bin_mat = np.frombuffer(data, dtype=np.bool_).reshape(shape)
debinarized = _FUNCS["debinarize_rc_bin_split_v2"](bin_mat, axis=0)
return _FUNCS["reverse_diag_transform_mode0"](debinarized)
[docs] def decode_tiles(self, payloads: list[bytes]) -> np.ndarray:
"""Decode multiple tiles into a 4D array.
Parameters
----------
payloads : list[bytes]
List of encoded bitstreams.
Returns
-------
np.ndarray
4D array of decoded tiles (n_tile_rows, n_tile_cols, tile_size, tile_size).
Examples
--------
"""
n_tiles = len(payloads)
decoded = [self.decode_tile(p) for p in payloads]
tile_shape = decoded[0].shape
tile_rows = int(np.sqrt(n_tiles))
tile_cols = n_tiles // tile_rows if tile_rows > 0 else 1
result = np.empty((tile_rows, tile_cols, *tile_shape), dtype=self.dtype)
idx = 0
for i in range(tile_rows):
for j in range(tile_cols):
result[i, j] = decoded[idx]
idx += 1
return result