Skip to content

Universe Module

def get_uni(file, chrom, cutoff=None)

For each position check if coverage is bigger than cut-off.

If cut-off not provided, calculate value that gives maximum likelihood universe.

Args: file (str): Coverage file. chrom (str): Chromosome to analyse. cutoff (int): Base pairs with values greater than or equal to cut-off can be included in universe.

Returns: ndarray: Vector with universes states; 0 - background, 1 - universe.

def save_simple(file_out, inter_pos, chrom)

Save cut-off universe to a file without any processing.

Args: file_out (str): Output file. inter_pos (ndarray): Boolean vector indicating whether each position should be included in universe. chrom (str): Chromosome to analyse.

def marge_filter(file_out, inter_pos, chrom, merge_dist=100, size_flt=1000)

Save cut-off universe to a file with filtering region size and merging close regions.

Args: file_out (str): Output file. inter_pos (ndarray): Boolean vector indicating whether each position should be included in universe. chrom (str): Chromosome to analyse. merge_dist (int): Regions closer than merge_dist will be merged into one. size_flt (int): Regions smaller than size_flt will not be reported.

def cc_universe(cove, file_out, cove_prefix='all', merge=0, filter_size=0, cutoff=None)

Create cut-off coverage universe based on coverage track.

Args: cove (str): Path to coverage folder. file_out (str): Output file. cove_prefix (str): Prefix of the coverage file. merge (int): Regions closer than this value will be merged into one. filter_size (int): Regions smaller than this value will not be reported. cutoff (int): Base pairs with coverage equal to or greater than this value will be included in the universe.

def ana_region(reg, start_s, starts, ends, track_val)

Check how many regions given part of universe contains.

Args: reg (ndarray): Vector with universes states; 0 - background, 1 - boundary, 2 - core. start_s (int): Start of the part of the universe. starts (list): List of region starts in given part of universe. ends (list): List of region ends in given part of universe. track_val (ndarray): Genome coverage by the collection for given part of universe.

def save_regions(inter_pos, chrom, bedname, track)

Save regions from universes to file.

Args: inter_pos (ndarray): Vector with universes states; 0 - background, 1 - boundary, 2 - core. chrom (str): Chromosome to analyse. bedname (str): Output file. track (ndarray): Vector with coverage values.

def get_uni(file, chrom, bedname)

Build cut-off coverage flexible universes from coverage track.

Args: file (str): Coverage file. chrom (str): Chromosome to analyse. bedname (str): Output file.

def ccf_universe(cove, file_out, cove_prefix='all')

Create cut-off flexible universe based on coverage track.

Args: cove (str): Path to coverage folder. file_out (str): Output file. cove_prefix (str): Prefix used for creating signal tracks.

def norm(track: np.ndarray, mode: str) -> None

Normalize the coverage track depending on track type.

For each unique value in the track, calculates the corresponding quantile taking into account that values occur different number of times.

Args: track (ndarray): Coverage track that will be modified in-place. mode (str): Type of track ('ends' or 'core').

Returns: None: The function modifies the track array in-place.

def process_bigwig(file: pyBigWig.pyBigWig, seq: np.ndarray, p: int, chrom: str, chrom_size: int, normalize: bool = True, mode: Optional[str] = None) -> None

Preprocess bigWig file.

Args: file: BigWig file object. seq (ndarray): Sequence array to store results, modified in-place. p (int): Position in sequence array. chrom (str): Chromosome name. chrom_size (int): Chromosome size. normalize (bool): Whether to normalize the track. mode (str): Normalization mode ('ends' or 'core').

Returns: None: The function modifies the seq array in-place.

def read_data(start: str, core: str, end: str, chrom: str, normalize: bool = True) -> Tuple[int, np.ndarray]

Read in and preprocess data.

Args: start (str): Path to file with start coverage. core (str): Path to file with core coverage. end (str): Path to file with end coverage. chrom (str): Chromosome to analyse. normalize (bool): Whether to normalize the coverage.

Returns: tuple: Chromosome size and coverage matrix.

def split_predict(seq: np.ndarray, empty_starts: List[int], empty_ends: List[int], model) -> np.ndarray

Make model prediction only for regions containing nonzero positions.

Args: seq (ndarray): Coverage sequence. empty_starts (list): List of start positions. empty_ends (list): List of end positions. model: HMM model.

Returns: ndarray: HMM predictions.

def run_hmm(start: str, core: str, end: str, chrom: str, normalize: bool = True) -> Tuple[np.ndarray, any]

Make HMM prediction for given chromosome.

Args: start (str): Path to start coverage file. core (str): Path to core coverage file. end (str): Path to end coverage file. chrom (str): Chromosome to analyse. normalize (bool): Whether to normalize the coverage.

Returns: tuple: HMM predictions and model.

def hmm_universe(coverage_folder: str, out_file: str, prefix: str = 'all', normalize: bool = True, save_max_cove: bool = False) -> None

Create HMM based universe from coverage.

Args: coverage_folder (str): Path to folder with coverage files. out_file (str): Path to the output file with universe. prefix (str): Prefix for coverage file names. normalize (bool): Whether to normalize file. save_max_cove (bool): Whether to save the maximum peak coverage.

Returns: None: The function writes predictions to the output file.

def test_hmm(message: str) -> None

Print a test message.

Args: message (str): Message to print.

Returns: None: The function logs the message.

def process_part(cove, model_start=np.array([[]]), model_core=np.array([[]]), model_end=np.array([[]]))

Find ML path through matrix using dynamic programming with numba.

Args: cove (ndarray): Coverage tracks. model_start (ndarray): Likelihood model for starts. model_core (ndarray): Likelihood model for core. model_end (ndarray): Likelihood model for ends.

Returns: ndarray: ML path through matrix.

def make_ml_flexible_universe(model_lh, cove_folder, cove_prefix, chrom, file_out)

Make ML flexible universe per chromosome.

Args: model_lh (ModelLH): Likelihood model. cove_folder (str): Path to a folder with genome coverage by tracks. cove_prefix (str): Prefix used in uniwig for creating coverage. chrom (str): Chromosome to be processed. file_out (str): Output file with the universe.

def ml_universe(model_file, cove_folder, cove_prefix, file_out)

Make ML flexible universe.

Args: model_file (str): Input name with likelihood models. cove_folder (str): Path to a folder with genome coverage by tracks. cove_prefix (str): Prefix used in uniwig for creating coverage. file_out (str): Output file with the universe.

  • package_name = 'numba'

class Model(trans_matrix, init_para, para, save_matrix)

Bases: ABC

Abstract class of HMM models.

Initialize HMM model.

Args: trans_matrix (ndarray): Transition matrix. init_para (str): Parameters that need to be initialized before training. para (str): Parameters that are updated during training. save_matrix (bool): Whether to save transition matrix to a file.

  • init_para
  • para
  • state_no
  • save_matrix
  • trans_matrix
  • start_matrix

def save_tras(out_folder)

class PoissonModel(trans_matrix, lambdas_matrix, init_para='', para='', save_matrix=True, out_folder='')

Bases: Model

HMM model with Poisson emissions.

Initialize Poisson HMM model.

Args: trans_matrix (ndarray): Transition matrix. lambdas_matrix (ndarray): Lambda values for emission probabilities. init_para (str): Parameters to initialize. para (str): Parameters to update during training. save_matrix (bool): Whether to save parameter matrix. out_folder (str): Folder to which save parameter matrix.

  • lambdas_matrix
  • model

class GaussianModel(trans_matrix, means_matrix, covars_matrix, init_para='', para='', save_matrix=True, out_folder='')

Bases: Model

HMM model with Gaussian emissions.

Initialize Gaussian HMM model.

Args: trans_matrix (ndarray): Transition matrix. means_matrix (ndarray): Mean values for emission probabilities. covars_matrix (ndarray): Covariance values for emission probabilities. init_para (str): Parameters to initialize. para (str): Parameters to update during training. save_matrix (bool): Whether to save parameter matrix. out_folder (str): Folder to which save parameter matrix.

  • means_matrix
  • covars_matrix
  • model

class NBModel(trans_matrix, failures_matrix, prob_matrix, init_para='', para='', save_matrix=True, out_folder='')

Bases: Model

HMM model with Negative Binomial emissions.

Initialize Negative Binomial HMM model.

Args: trans_matrix (ndarray): Transition matrix. failures_matrix (ndarray): Number of failures for emission probabilities. prob_matrix (ndarray): Success probability for emission probabilities. init_para (str): Parameters to initialize. para (str): Parameters to update during training. save_matrix (bool): Whether to save parameter matrix. out_folder (str): Folder to which save parameter matrix.

  • failures_matrix
  • prob_matrix
  • model

class BetaModel(trans_matrix, alpha_matrix, beta_matrix, init_para='', para='', save_matrix=True, out_folder='')

Bases: Model

HMM model with Beta emissions.

Initialize Beta HMM model.

Args: trans_matrix (ndarray): Transition matrix. alpha_matrix (ndarray): Alpha values for emission probabilities. beta_matrix (ndarray): Beta values for emission probabilities. init_para (str): Parameters to initialize. para (str): Parameters to update during training. save_matrix (bool): Whether to save parameter matrix. out_folder (str): Folder to which save parameter matrix.

  • alpha_matrix
  • beta_matrix
  • model

class NBHMM(n_components=1, startprob_prior=1.0, transmat_prior=1.0, failures_prior=0.0, prob_prior=0.0, failures_weight=0.0, prob_weight=0.0, algorithm='viterbi', random_state=None, n_iter=10, tol=0.01, verbose=False, params='strp', init_params='strp', implementation='log')

Bases: BaseHMM

Hidden Markov Model with Negative Binomial emission distributions.

This class extends BaseHMM to use Negative Binomial distributions for emission probabilities, suitable for modeling count data with overdispersion.

Args: n_components (int): Number of hidden states. startprob_prior (float): Prior for start probabilities. transmat_prior (float): Prior for transition matrix. failures_prior (float): Prior for number of failures parameter. prob_prior (float): Prior for success probability parameter. failures_weight (float): Weight for failures prior. prob_weight (float): Weight for probability prior. algorithm (str): Decoding algorithm ('viterbi' or 'map'). random_state: Random state for reproducibility. n_iter (int): Maximum number of iterations for training. tol (float): Convergence threshold. verbose (bool): Whether to print training progress. params (str): String of parameters to update during training. init_params (str): String of parameters to initialize. implementation (str): Implementation type ('log' or 'scaling').

  • failures_prior
  • failures_weight
  • prob_prior
  • prob_weight

class BetaHMM(n_components=1, startprob_prior=1.0, transmat_prior=1.0, alpha_prior=0.0, beta_prior=0.0, alpha_weight=0.0, beta_weight=0.0, algorithm='viterbi', random_state=None, n_iter=10, tol=0.01, verbose=False, params='strp', init_params='strp', implementation='log')

Bases: BaseHMM

Hidden Markov Model with Beta emission distributions.

This class extends BaseHMM to use Beta distributions for emission probabilities, suitable for modeling continuous data in the [0, 1] range.

Args: n_components (int): Number of hidden states. startprob_prior (float): Prior for start probabilities. transmat_prior (float): Prior for transition matrix. alpha_prior (float): Prior for alpha parameters of Beta distributions. beta_prior (float): Prior for beta parameters of Beta distributions. alpha_weight (float): Weight for alpha prior. beta_weight (float): Weight for beta prior. algorithm (str): Decoding algorithm ('viterbi' or 'map'). random_state: Random state for reproducibility. n_iter (int): Maximum number of iterations for training. tol (float): Convergence threshold. verbose (bool): Whether to print training progress. params (str): String of parameters to update during training. init_params (str): String of parameters to initialize. implementation (str): Implementation type ('log' or 'scaling').

  • alpha_prior
  • alpha_weight
  • beta_prior
  • beta_weight

def ana_region(region, start_s)

Helper for saving HMM prediction into a file.

Args: region (ndarray): Region array containing state information. start_s (int): Start position.

Returns: tuple: Start end, end start coordinates.

def predictions_to_bed(states, chrom, bedname, save_max_cove=False, cove_file=None)

Save HMM prediction into a file.

Args: states (ndarray): Result of HMM prediction. chrom (str): Which chromosome is being analyzed. bedname (str): Path to the output file. save_max_cove (bool): Whether to save the maximum peak coverage to output file. Can result in non-standard BED file. cove_file (str): File with core coverage, required for saving maximum peak coverage.

def find_full_full_pos(seq, gap_size=1000, area_size=500)

Look for nonzero positions in coverage matrix when most positions are zero.

Args: seq (ndarray): Vector with information about non-zero positions. gap_size (int): Size of minimum gap between non-zero positions that are separated. area_size (int): Size of the area around non-zero positions to be included in the result.

Returns: tuple: List of starts of non-zero regions and list of ends of non-zero regions.

def find_full_empty_pos(seq, gap_size=10000, area_size=1000)

Look for nonzero positions in coverage matrix when most positions are nonzero.

Args: seq (ndarray): Vector with information about non-zero positions. gap_size (int): Size of minimum gap between non-zero positions that are separated. area_size (int): Size of the area around non-zero positions to be included in the result.

Returns: tuple: List of starts of non-zero regions and list of ends of non-zero regions.

def find_full(seq)

Look for nonzero positions in coverage matrix.

Args: seq (ndarray): Coverage matrix.

Returns: tuple: List of starts and ends of regions with non-zero positions.