"""
CMC Encoder wrapper for GNZ v3 compression.
Examples
--------
"""
__author__ = "Yeremia Gunawan Adhisantoso"
__email__ = "adhisant@tnt.uni-hannover.de"
__license__ = "Clear BSD"
import typing as t
import numpy as np
from . import _paths
# Module-level registry of CMC encoder functions. Populated lazily on
# first access by ``_ensure_loaded()``. Replaces the previous pattern of
# writing to module ``globals()`` (DOP Phase 3, v2.20.0): a single
# private dict is both easier to grep and easier to test.
_FUNCS: dict[str, t.Any] = {}
_LOADED: bool = False
def _ensure_loaded() -> None:
"""Lazily resolve CMC encoder functions on first access.
This defers the resolution of the CMC third-party library until a CMC
codec function is actually invoked. Without this, importing this module
(which happens transitively when ``gunz_cm.compressions`` is imported for
any codec) raises FileNotFoundError when GUNZ_CM_CMC_DIR is unset, even
for ZSTD-only workloads.
"""
global _LOADED
if _LOADED:
return
_cmc = _paths.load_cmc_encoder()
_FUNCS["jbig_encode"] = _cmc["jbig_encode"]
_FUNCS["binarize_rc_bin_split_v2"] = _cmc["binarize_rc_bin_split_v2"]
_FUNCS["diag_transform"] = _cmc["diag_transform"]
_LOADED = True
def __getattr__(name: str) -> t.Any:
"""PEP 562 module-level lazy attribute access for CMC encoder functions."""
if name in {"jbig_encode", "binarize_rc_bin_split_v2", "diag_transform"}:
_ensure_loaded()
return _FUNCS[name]
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
[docs]class CmcEncoder:
"""CMC encoder for contact matrix tiles.
Parameters
----------
tile_size : int, default=256
Tile size for block processing.
resolution : int, default=50000
Hi-C resolution in bp.
diag_transform : bool, default=True
Apply diagonal transform before encoding.
Examples
--------
"""
def __init__(
self,
tile_size: int = 256,
resolution: int = 50000,
diag_transform: bool = True,
):
"""
Examples
--------
"""
self.tile_size = tile_size
self.resolution = resolution
self.diag_transform = diag_transform
[docs] def encode_tile(self, mat: np.ndarray) -> bytes:
"""Encode a single contact matrix tile.
Parameters
----------
mat : np.ndarray
2D contact matrix tile (upper triangular).
Returns
-------
bytes
CMC-encoded bitstream.
Examples
--------
"""
_ensure_loaded()
if self.diag_transform:
mat = _FUNCS["diag_transform"](mat, mode=0)
bin_mat = _FUNCS["binarize_rc_bin_split_v2"](mat, axis=0)
return _FUNCS["jbig_encode"](bin_mat)
[docs] def encode_tiles(self, tiles: np.ndarray) -> list[bytes]:
"""Encode multiple tiles.
Parameters
----------
tiles : np.ndarray
4D array of shape (n_tile_rows, n_tile_cols, tile_size, tile_size).
Returns
-------
list[bytes]
List of encoded bitstreams, one per tile.
Examples
--------
"""
n_tile_rows, n_tile_cols = tiles.shape[0], tiles.shape[1]
results = []
for i in range(n_tile_rows):
for j in range(n_tile_cols):
results.append(self.encode_tile(tiles[i, j]))
return results
[docs] def get_compression_info(self) -> dict:
"""Return compression metadata.
Returns
-------
dict
Compression parameters for header.
Examples
--------
"""
return {
"codec": "cmc",
"version": "1.0",
"tile_size": self.tile_size,
"resolution": self.resolution,
"diag_transform": self.diag_transform,
}