"""
LZ4 block encoder wrapper for GZCM v4 compression.
Uses the third-party ``lz4`` package (PyPI). This module must NOT be imported
at package load time: the codec is intended for GZCM v4 and is only required
when the v4 lz4 codec is selected. The dependency is intentionally lazy so
that users who do not need v4 lz4 do not pay the import cost.
Install with::
pip install gunz-cm[v4] # or: pip install lz4
Notes
-----
The block decoder requires the uncompressed size to allocate the output
buffer. We mirror the v2.14.0 8-byte (rows, cols) header pattern used by
``zstd_decoder.py`` and prepend an additional 4-byte little-endian uint32
``uncompressed_size`` field. ``lz4.block.compress`` is called with
``store_size=False`` so the body is a pure compressed bitstream with no
conflicting size prefix; the decoder reads the size exclusively from our
12-byte header. Total header size: 12 bytes.
Wire format (v4 ``lz4`` codec)::
[0:8] int32 little-endian (rows, cols)
[8:12] uint32 little-endian uncompressed_size (bytes)
[12:] lz4.block.compress(tile.tobytes(), store_size=False)
Examples
--------
>>> import numpy as np
>>> from gunz_cm.compressions.lz4 import Lz4Encoder, Lz4Decoder
>>> tile = np.zeros((256, 256), dtype=np.uint32)
>>> enc = Lz4Encoder(tile_size=256)
>>> dec = Lz4Decoder(tile_size=256)
>>> payload = enc.encode_tile(tile)
>>> np.array_equal(tile, dec.decode_tile(payload, shape=(256, 256)))
True
"""
__author__ = "Yeremia Gunawan Adhisantoso"
__email__ = "adhisant@tnt.uni-hannover.de"
__license__ = "Clear BSD"
import struct
import numpy as np
try:
import lz4.block as _lz4_block
HAS_LZ4 = True
except ImportError:
HAS_LZ4 = False
# Header layout: int32 rows, int32 cols, uint32 uncompressed_size -> 12 bytes.
_HEADER_STRUCT = struct.Struct("<iiI")
_HEADER_SIZE = _HEADER_STRUCT.size # 12
def _ensure_lz4():
"""Raise an informative ImportError if ``lz4`` is not installed.
Returns
-------
module
The ``lz4.block`` module.
Raises
------
ImportError
With a message that points to the correct install command.
"""
if not HAS_LZ4:
raise ImportError(
"lz4 is required for the GZCM v4 lz4 codec. "
"Install with: pip install gunz-cm[v4] (or pip install lz4)"
)
return _lz4_block
[docs]class Lz4Encoder:
"""LZ4 block encoder for contact matrix tiles.
Parameters
----------
tile_size : int, default=256
Tile side length in bins. Stored for ``get_compression_info``; the
encoder itself operates on whatever shape is passed to
``encode_tile``.
resolution : int, default=50000
Hi-C resolution in bp. Stored for metadata.
mode : str, default="default"
LZ4 block mode passed to ``lz4.block.compress``. One of
``"default"``, ``"fast"``, ``"high_compression"``.
acceleration : int, default=1
LZ4 acceleration factor (>=1). Higher values trade compression ratio
for speed; ``1`` is the slowest but best ratio.
hc_level : int, default=9
High-compression level (4-16) used when ``mode='high_compression'``.
Ignored otherwise. ``9`` matches the scheme-picker candidate
``lz4-hc-9`` defined in ``specs/gzcm-v4-design.md`` §4.7.
Examples
--------
"""
def __init__(
self,
tile_size: int = 256,
resolution: int = 50000,
mode: str = "default",
acceleration: int = 1,
hc_level: int = 9,
):
"""
Examples
--------
"""
self.tile_size = tile_size
self.resolution = resolution
self.mode = mode
self.acceleration = acceleration
self.hc_level = hc_level
[docs] def encode_tile(self, mat: np.ndarray) -> bytes:
"""Encode a single contact matrix tile.
The 12-byte header (rows, cols, uncompressed_size) is prepended so
the decoder can reshape edge tiles and call
``lz4.block.decompress`` with the correct ``uncompressed_size``.
We call ``lz4.block.compress(..., store_size=False)`` so the
uncompressed size is NOT also written into the body of the
compressed block. The block decoder receives the size exclusively
through our 12-byte header — otherwise the ``decompress`` call below
would see a size prefix that disagrees with ``uncompressed_size``.
Parameters
----------
mat : np.ndarray
2D contact matrix tile (any rectangular shape).
Returns
-------
bytes
12-byte header + lz4 block-compressed payload.
Examples
--------
"""
lz4_block = _ensure_lz4()
rows = int(mat.shape[0])
cols = int(mat.shape[1])
raw = mat.tobytes()
uncompressed_size = len(raw)
if self.mode == "high_compression":
compressed_body = lz4_block.compress(
raw,
mode="high_compression",
compression=self.hc_level,
store_size=False,
)
else:
compressed_body = lz4_block.compress(
raw,
mode=self.mode,
acceleration=self.acceleration,
store_size=False,
)
header = _HEADER_STRUCT.pack(rows, cols, uncompressed_size)
return header + compressed_body
[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_rows, tile_cols).
Returns
-------
list[bytes]
One payload per tile, in row-major order.
Examples
--------
"""
n_tile_rows, n_tile_cols = tiles.shape[0], tiles.shape[1]
results: list[bytes] = []
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 for the file header.
Returns
-------
dict
Compression parameters.
Examples
--------
"""
return {
"codec": "lz4",
"version": "1.0",
"tile_size": self.tile_size,
"resolution": self.resolution,
"mode": self.mode,
"acceleration": self.acceleration,
"hc_level": self.hc_level,
}
[docs]class Lz4Decoder:
"""LZ4 block decoder for contact matrix tiles.
Parameters
----------
tile_size : int, default=256
Default tile side length used when ``shape`` is not provided to
``decode_tile``.
resolution : int, default=50000
Hi-C resolution in bp. Stored for metadata symmetry with
``Lz4Encoder``; the decoder itself does not use it.
dtype : numpy data type, default=numpy.uint32
Output dtype for decoded tiles. Accepts anything
:func:`numpy.dtype` understands.
Examples
--------
"""
def __init__(
self,
tile_size: int = 256,
resolution: int = 50000,
dtype=np.uint32,
):
"""
Examples
--------
"""
self.tile_size = tile_size
self.resolution = resolution
self.dtype = np.dtype(dtype)
[docs] def decode_tile(
self,
payload: bytes,
shape: tuple[int, int] | None = None,
) -> np.ndarray:
"""Decode a single compressed tile.
Two payload conventions are supported:
- **v4 GZCM path** (``shape`` provided): ``payload`` begins with the
12-byte ``(rows, cols, uncompressed_size)`` header followed by the
lz4 block body. ``rows``/``cols`` are taken from ``shape`` (so the
caller can override if the tile is non-rectangular after a future
change); the ``uncompressed_size`` field is the authoritative hint
for ``lz4.block.decompress``.
- **Legacy / direct codec path** (``shape`` is ``None``): ``payload``
is treated as a raw lz4 block body and the tile is reshaped to
``(tile_size, tile_size)``.
Parameters
----------
payload : bytes
Compressed bitstream (with or without the 12-byte header).
shape : tuple of (int, int), optional
Actual tile shape ``(rows, cols)``. Required for edge tiles.
Returns
-------
np.ndarray
Decoded contact matrix tile.
Examples
--------
"""
lz4_block = _ensure_lz4()
if shape is not None:
rows, cols = shape
if len(payload) < _HEADER_SIZE:
raise ValueError(
f"lz4 payload too short: expected >= {_HEADER_SIZE} bytes "
"for the v4 header, got "
f"{len(payload)}"
)
_hdr_rows, _hdr_cols, uncompressed_size = _HEADER_STRUCT.unpack_from(
payload, 0
)
body = payload[_HEADER_SIZE:]
# The caller-provided shape wins; the header fields are only a
# diagnostic cross-check.
del _hdr_rows, _hdr_cols
else:
rows = cols = self.tile_size
uncompressed_size = self.tile_size * self.tile_size * self.dtype.itemsize
body = payload
raw = lz4_block.decompress(body, uncompressed_size=uncompressed_size)
return np.frombuffer(raw, dtype=self.dtype).reshape(rows, cols)
[docs] def decode_tiles(
self,
payloads: list[bytes],
shapes: list[tuple[int, int]] | None = None,
) -> np.ndarray:
"""Decode multiple tiles into a 4D array.
Parameters
----------
payloads : list[bytes]
List of encoded bitstreams.
shapes : list of tuple of (int, int), optional
Per-tile shapes matching ``payloads``.
Returns
-------
np.ndarray
4D array of decoded tiles ``(n_tile_rows, n_tile_cols, rows, cols)``.
Examples
--------
"""
n_tiles = len(payloads)
if shapes is None:
decoded = [self.decode_tile(p) for p in payloads]
else:
decoded = [
self.decode_tile(p, shape=s)
for p, s in zip(payloads, shapes, strict=True)
]
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