Source code for gunz_cm.cli.loaders

import typer
import pathlib
import typing as t
from loguru import logger
import pandas as pd
from rich.console import Console

from .. import loaders
from ..consts import DataStructure

app = typer.Typer(help="CM Library - Loaders")
console = Console()

[docs] @app.command() def get_resolutions( input_path: str = typer.Argument(..., help="Path to the contact matrix file"), ): """ Gets the resolutions available in a contact matrix file. """ try: resolutions = loaders.get_resolutions(input_path) logger.info(f"Available resolutions for {input_path}:") console.print(resolutions) except Exception as e: logger.error(f"Failed to get resolutions: {e}") raise typer.Exit(code=1) from e
[docs] @app.command() def get_chrom_infos( input_path: str = typer.Argument(..., help="Path to the contact matrix file"), ): """ Queries chromosome names and lengths from a .hic or .cool file. """ try: chrom_infos = loaders.get_chrom_infos(input_path) logger.info(f"Chromosome info for {input_path}:") for chrom, length in chrom_infos.items(): console.print(f" {chrom}: {length}") except Exception as e: logger.error(f"Failed to get chromosome info: {e}") raise typer.Exit(code=1) from e
[docs] @app.command() def get_bins( input_path: str = typer.Argument(..., help="Path to the contact matrix file"), resolution: int = typer.Argument(..., help="Resolution"), ): """ Gets the binnified index from a .hic or .cool file. """ try: bins = loaders.get_bins(input_path, resolution) logger.info(f"Bins for {input_path} at resolution {resolution}:") console.print(bins.head()) console.print(f"... ({len(bins)} total rows)") except Exception as e: logger.error(f"Failed to get bins: {e}") raise typer.Exit(code=1) from e
[docs] @app.command() def get_balancing( input_path: str = typer.Argument(..., help="Path to the contact matrix file"), resolution: int = typer.Argument(..., help="Resolution"), chrom: str = typer.Argument(..., help="Chromosome"), ): """ Gets available balancing methods for a region in a .hic or .cool file. """ try: balancing_methods = loaders.get_balancing(input_path, resolution, chrom) logger.info(f"Available balancing methods for {chrom} at {resolution}:") console.print(balancing_methods) except Exception as e: logger.error(f"Failed to get balancing methods: {e}") raise typer.Exit(code=1) from e
[docs] @app.command() def load_data( input_path: str = typer.Argument(..., help="Path to the contact matrix file"), resolution: int = typer.Argument(..., help="Resolution"), region1: t.Optional[str] = typer.Option(None, help="First genomic region (e.g., 'chr1')"), region2: t.Optional[str] = typer.Option(None, help="Second genomic region"), balancing: t.Optional[str] = typer.Option(None, help="Balancing method"), output_format: DataStructure = typer.Option(DataStructure.DF, help="Output format (df, coo, etc.)"), ): """ Loads contact matrix data and displays a summary. """ try: logger.info(f"Loading data from {input_path}") data = loaders.load_cm_data( fpath=pathlib.Path(input_path), resolution=resolution, region1=region1, region2=region2, balancing=balancing, output_format=output_format ) logger.success("Successfully loaded data.") if isinstance(data, pd.DataFrame): console.print(data.head()) console.print(f"... shape: {data.shape}") elif isinstance(data, tuple) and len(data) == 3: # COO format rows, cols, values = data console.print(f"COO format: {len(values)} entries") console.print(f"Row indices head: {rows[:5]}") console.print(f"Col indices head: {cols[:5]}") console.print(f"Values head: {values[:5]}") else: console.print(f"Loaded data type: {type(data)}") except Exception as e: logger.error(f"Failed to load data: {e}") raise typer.Exit(code=1) from e
if __name__ == "__main__": app()