from __future__ import annotations
# -*- coding: utf-8 -*-
"""
Module.
"""
from pydantic import validate_call
__author__ = "Yeremia Gunawan Adhisantoso"
__credits__ = ["Yeremia Gunawan Adhisantoso"]
__license__ = "Clear BSD"
# __version__ = "1.0."
__maintainer__ = "Yeremia Gunawan Adhisantoso"
__email__ = "adhisant@tnt.uni-hannover.de"
# __status__ = "Production"
from collections import namedtuple
from functools import lru_cache
import typing as t
import pandas as pd
from gunz_cm.consts import DataFrameSpecs, Format, SUPPORTED_COMPRESSION_SCHEMES
from ..exceptions import InvalidRegionFormatError, LoaderError
ROW_IDS_COLNAME = DataFrameSpecs.ROW_IDS
COL_IDS_COLNAME = DataFrameSpecs.COL_IDS
COUNTS_COLNAME = DataFrameSpecs.COUNTS
class Constant:
"""Bind instance to name to get a unique object.
Examples
--------
"""
pass
ClosedInterval = namedtuple("ClosedInterval", ["start", "end"])
[docs]class Region:
"""Represent a range of loci and interface with the textual UCSC style in the form 'chr22:1,000,000-1,500,000'
Use static method Region.from_string to parse USCS string. Converting back to string will canonicalize
Attributes
----------
chromosome : int or str
1-22 or string X/Y/M (use chromname() to get chrN string)
region: ClosedInterval or the constant Region.ALL_LOCI
For 'chr1:100-500', region.start == 100 and region.end == 500
Examples
--------
Parsing tries to be more lenient than the canonical form requires.
>>> str(Region.from_string('1:1,000-1,500')) == 'chr1:1000-1500'
True
>>> str(Region.from_string('chry')) == 'chrY'
True
"""
ALL_LOCI = Constant()
def __init__(
self,
chromosome: int | str, # Number or X, Y
region: tuple[int, int] | Constant,
):
"""
Function __init__.
Parameters
----------
chromosome : t.Union[int, str]
The chromosome identifier.
region : t.Union[t.Tuple[int, int], Constant]
The region interval or constant.
Returns
-------
None
Examples
--------
Notes
-----
"""
self.chromosome = chromosome
if region is Region.ALL_LOCI:
self.region = region
else:
self.region = ClosedInterval(*region)
[docs] def is_full_chrom(self) -> bool:
"""This region describes the full chromosome, so region is 0:N
Returns
-------
bool
True if region is the full chromosome, False otherwise.
Examples
--------
"""
return self.region is Region.ALL_LOCI
[docs] def chromname(self) -> str:
"""
Function chromname.
Returns
-------
str
The formatted chromosome string.
Examples
--------
Notes
-----
"""
return f"chr{self.chromosome}"
[docs] @staticmethod
def from_string(
region: str,
) -> Region:
"""
Function from_string.
Parameters
----------
region : str
The string to parse.
Returns
-------
Region
The parsed Region object.
Examples
--------
Notes
-----
"""
chrom, *rest = region.lstrip("chr").split(":")
if len(rest) == 0:
interval = Region.ALL_LOCI
elif len(rest) == 1:
try:
interval = ClosedInterval(*map(int, rest[0].replace(",", "").split("-")))
except:
raise LoaderError(
"Invalid region format: Must match template 'chr<c>:<start>-<end>' or 'chr<c>'\n"
)
if interval.start > interval.end:
raise LoaderError("Start must be smaller than end")
else:
raise LoaderError(
"Invalid region format: Must match template 'chr<c>:<start>-<end>' or 'chr<c>'\n"
)
return Region(chrom, interval)
def __str__(self) -> str:
"""
Function __str__.
Returns
-------
str
The canonical UCSC region string format.
Examples
--------
Notes
-----
"""
if self.is_full_chrom():
return self.chromname()
else:
return f"{self.chromname()}:{self.region.start}-{self.region.end}"
@validate_call(config=dict(arbitrary_types_allowed=True))
def _generate_region_mask(
df: pd.DataFrame,
region: Region | str,
bin_size_bp: int
) -> pd.Series:
"""
Generate a pandas boolean mask that would filter out everything besides the specified region.
Pure function.
Params:
df: Read only dataframe that the mask will be created for
region: string of UCSC format f"{int(locus_start)}-{int(locus_end)}"
Returns:
Boolean mask s.t. df[mask] only contains the loci specified in region.
Examples
--------
"""
if isinstance(region, str):
region = Region.from_string(region)
# Pattern matching will make this cleaner. Edit this once 3.10 is the minimum supported version
if region.is_full_chrom():
start = 1
end = None
else:
start, end = region.region
# https://genome.ucsc.edu/goldenPath/help/query.html
# UCSC indexes start at 1, but our python arrays are 0-indexed. Thus, subtract 1
start_idx = max(0, start - 1)
mask = df[ROW_IDS_COLNAME] >= start_idx
mask &= df[COL_IDS_COLNAME] >= start_idx
if end is not None:
# UCSC ranges are inclusive, but need to offset 1 due to python 0-indexing
mask &= df[ROW_IDS_COLNAME] <= (end - 1)
mask &= df[COL_IDS_COLNAME] <= (end - 1)
return mask
def _get_fileext_without_compression(
fname: str,
) -> str:
"""
Find the fileextension before the compression extension. Case insensitive.
Parameters
----------
fname : str
Path/ name of a file
Returns
-------
str
The real format of the file.
Examples
--------
>>> _get_fileext_without_compression("path/to/file.txt.tar.gz")
'txt'
"""
return __get_fileext_without_compression_impl(fname.split("."))
def __get_fileext_without_compression_impl(
parts: list[str],
) -> str:
"""
Function __get_fileext_without_compression_impl.
Parameters
----------
parts : t.List[str]
A list of string parts.
Returns
-------
str
The file extension.
Examples
--------
Notes
-----
"""
# Helper so caller doesn't have to split the string needlessly.
ext = parts[-1].lower()
if ext in SUPPORTED_COMPRESSION_SCHEMES:
return __get_fileext_without_compression_impl(parts[:-1])
return ext
# v2.8.0: Chromosome name normalization.
# Accepts both 'chr1' (UCSC) and '1' (Ensembl) for primary chromosomes,
# matching the user's "Accept both, document the difference" decision.
# Spec: docs/design/specs/chrom-name-normalization.md
# Formats that have actual chromosome names we can look up.
# CSV/COO/MCOO don't have a chrom column (only bin indices), so
# there's nothing to normalize. GINTERACTIONS has chr columns but
# the user's region1 selects which chr1 entries to keep, not a
# name to look up against a fixed set.
_FORMATS_WITH_CHROM_NAMES = (Format.HIC, Format.COOLER, Format.MEMMAP)
def _get_file_chrom_names(fpath: str, fformat: Format) -> set[str]:
"""Return the set of chromosome names stored in the file.
Only works for formats that store explicit chrom names
(HIC, COOLER, MEMMAP). For other formats, returns an empty
set, which signals to the normalizer to pass through.
"""
if fformat not in _FORMATS_WITH_CHROM_NAMES:
return set()
try:
# Lazy import to avoid circular import at module load
from . import _CHROM_INFO_GETTERS
getter = _CHROM_INFO_GETTERS.get(fformat)
if getter is None:
return set()
# COOLER's get_chrom_infos returns Dict[str, Dict[str, int]],
# HIC and MEMMAP return Dict[str, int]. Handle both.
result = getter(fpath)
if not result:
return set()
first_value = next(iter(result.values()))
if isinstance(first_value, dict):
return set(result.keys())
return set(result.keys())
except Exception:
return set()
@lru_cache(maxsize=128)
def _resolve_chrom(
fpath: str,
name: str,
fformat_value: str,
) -> str:
"""Resolve a single chrom name against the file's actual chroms.
Per the spec: accepts both 'chr1' and '1' for primary chromosomes
(1-22, X, Y). Returns the file's actual chrom name. Raises
InvalidRegionFormatError if the name can't be resolved.
Cached per-process with maxsize=128. Exceptions are NOT cached
(lru_cache doesn't cache exceptions by default; the user can
fix the file or input and retry).
Parameters
----------
fpath : str
Path to the contact matrix file.
name : str
The chrom name to resolve (e.g., 'chr1' or '1'). Must
already be the chrom part only (no ':start-end' suffix).
fformat_value : str
The format value (e.g., 'hic', 'cooler'). String because
lru_cache requires hashable args, and Format enums may not
be hashable across processes.
Returns
-------
str
The file's actual chrom name. May be the same as the input
(if it already matched) or the normalized form.
"""
fformat = Format(fformat_value)
file_chroms = _get_file_chrom_names(fpath, fformat)
# If the format has no chrom names (e.g., CSV), pass through
if not file_chroms:
return name
# 1. Exact match — no normalization needed
if name in file_chroms:
return name
# 2. Try the chr-prefix swap
candidates = []
if name.startswith("chr"):
stripped = name[3:]
if stripped and stripped in file_chroms:
candidates.append(stripped)
else:
prefixed = f"chr{name}"
if prefixed in file_chroms:
candidates.append(prefixed)
# 3. Resolve based on candidate count
if len(candidates) == 1:
return candidates[0]
if len(candidates) == 0:
raise InvalidRegionFormatError(
region=name,
message=(
f"Chromosome '{name}' not found in '{fpath}'. "
"Use 'gunz-cm loaders get-chrom-infos <file>' to "
"list available chromosomes."
),
)
# > 1 candidates — ambiguous (file has both 'chr1' and '1')
raise InvalidRegionFormatError(
region=name,
message=(
f"Ambiguous chromosome '{name}': matches multiple file "
f"chroms: {candidates}. Specify the file's actual "
"chrom name explicitly."
),
)
def _normalize_chrom_name(
fpath: str,
name: str,
fformat: Format,
) -> str:
"""Normalize a chromosome name against the file's actual chroms.
Accepts both 'chr1' (UCSC) and '1' (Ensembl). Returns the
file's actual chrom name. Handles interval syntax by splitting
on ':' and normalizing only the chrom part, preserving the
interval.
Parameters
----------
fpath : str
Path to the contact matrix file.
name : str
The chrom name to normalize. May include an interval
suffix like 'chr1:1000-2000'.
fformat : Format
The file format (e.g., Format.HIC).
Returns
-------
str
The normalized name with the file's actual chrom prefix
and the original interval (if any).
Examples
--------
>>> _normalize_chrom_name('file.hic', 'chr1', Format.HIC)
'1' # if the HIC file stores chroms as '1'
>>> _normalize_chrom_name('file.hic', 'chr1:1000-2000', Format.HIC)
'1:1000-2000'
>>> _normalize_chrom_name('file.cool', '1', Format.COOLER)
'chr1' # if the COOLER file stores chroms as 'chr1'
"""
chrom_part, sep, rest = name.partition(":")
if not sep:
return _resolve_chrom(fpath, chrom_part, fformat.value)
# Interval syntax: normalize the chrom part, preserve the rest
normalized_chrom = _resolve_chrom(fpath, chrom_part, fformat.value)
return f"{normalized_chrom}:{rest}"