"""
BSC Encoder wrapper for GZCM v3 compression.
Uses BSC CLI subprocess with LD_LIBRARY_PATH for libomp.
Examples
--------
"""
__author__ = "Yeremia Gunawan Adhisantoso"
__email__ = "adhisant@tnt.uni-hannover.de"
__license__ = "Clear BSD"
import os
import subprocess
import tempfile
import pathlib
import numpy as np
[docs]class BscEncoder:
"""BSC encoder for contact matrix tiles.
Uses bsc CLI subprocess for true BSC (Block Sorting Compression).
Parameters
----------
tile_size : int, default=512
Tile size for block processing.
resolution : int, default=50000
Hi-C resolution in bp.
level : int, default=3
Compression level (0-9, higher = better compression).
Examples
--------
"""
def __init__(
self,
tile_size: int = 512,
resolution: int = 50000,
level: int = 3,
):
"""
Examples
--------
"""
self.tile_size = tile_size
self.resolution = resolution
self.level = level
self._bsc_path = pathlib.Path("/home/adhisant/tmp/bin/bsc")
self._env = os.environ.copy()
self._env["LD_LIBRARY_PATH"] = "/home/adhisant/tmp/miniforge3/envs/gunz_cm/lib"
[docs] def encode_tile(self, mat: np.ndarray) -> bytes:
"""Encode a single contact matrix tile.
Parameters
----------
mat : np.ndarray
2D contact matrix tile.
Returns
-------
bytes
BSC-compressed bitstream.
Examples
--------
"""
with tempfile.NamedTemporaryFile(suffix=".dat", delete=False) as f_in:
f_in.write(mat.tobytes())
f_in.flush()
in_path = pathlib.Path(f_in.name)
with tempfile.NamedTemporaryFile(suffix=".dat", delete=False) as f_out:
out_path = pathlib.Path(f_out.name)
try:
subprocess.run([str(self._bsc_path), "e", str(in_path), str(out_path)], check=True, capture_output=True, env=self._env)
with open(out_path, "rb") as f:
data = f.read()
finally:
in_path.unlink(missing_ok=True)
out_path.unlink(missing_ok=True)
return data
[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": "bsc",
"version": "1.0",
"tile_size": self.tile_size,
"resolution": self.resolution,
"level": self.level,
}