Evaluation Module

get_ctt_score

def get_ctt_score(path: str, embed_type: str, seed: int = 42, num_data: int = 10000, num_workers: int = 10) -> float

Runs the cluster tendency test (CTT) on a model.

Args: path (str): The path to a model. embed_type (str): The type of the model: "region2vec" or "base". seed (int, optional): Random seed. Defaults to 42. num_data (int, optional): Number of embeddings used for evaluation. Defaults to 10000. num_workers (int, optional): Number of parallel processes used. Defaults to 10.

Raises: ValueError: The number of samples is too small. ZeroDivisionError: The denominator of the CTT score is zero.

Returns: float: The CTT score for the model.

get_ctt_batch

def get_ctt_batch(batch: List[Tuple[str, str]], seed: int = 42, num_data: int = 10000, save_path: str = None, num_workers: int = 10) -> List[Tuple[str, float]]

Runs the cluster tendency test (CTT) on a batch of models.

Args: batch (list[tuple[str, str]]): A list of (model path, model type) tuples. Model type could be "region2vec" or "base". seed (int, optional): Random seed. Defaults to 42. num_data (int, optional): Number of embeddings used for evaluation. Defaults to 10000. save_path (str, optional): Save the results to save_path. Defaults to None. num_workers (int, optional): Number of parallel processes used. Defaults to 10.

Returns: List[Tuple[str, float]]: A list of (model path, CTT score) tuples.

ctt_eval

def ctt_eval(batch: List[Tuple[str, str]], num_runs: int = 20, num_data: int = 10000, save_folder: str = None, num_workers: int = 10) -> List[Tuple[str, list[float]]]

Runs the CTT on a batch of models for multiple times.

Runs the cluster tendency test (CTT) for a batch of models for num_runs times with different random seeds.

Args: batch (list[tuple[str, str]]): A list of (model path, model type) tuples. Model type could be "region2vec" or "base". num_runs (int, optional): Number of runs. Defaults to 20. num_data (int, optional): Number of embeddings used for evaluation. Defaults to 10000. save_folder (str, optional): Folder to save the results from each run. Defaults to None. num_workers (int, optional): Number of parallel processes used. Defaults to 10.

Returns: list[tuple[str, list[float]]]: A list of (model path, CTT scores from num_runs) tuples.

sample_from_vocab

def sample_from_vocab(vocab: List[str], num_samples: int, seed: int = 42) -> List[str]

Samples regions from vocab.

Samples regions proportionally from each chromosome.

Args: vocab (list[str]): A list of regions. num_samples (int): Number of regions to sample. seed (int, optional): Random seed. Defaults to 42.

Returns: list[str]: A list of sampled regions.

get_gdst_score

def get_gdst_score(path: str, embed_type: str, num_samples: int = 10000, seed: int = 42, queue: Queue = None, worker_id: int = None) -> Union[float, Tuple[int, str, float]]

Runs the GDST on a model.

Args: path (str): The path to a model. embed_type (str): The model type: "region2vec" or "base". num_samples (int, optional): Number of embeddings used for evaluation. Defaults to 10000. seed (int, optional): Random seed. Defaults to 42. queue (Queue, optional): The multiprocessing queue used to store results. Defaults to None. worker_id (int, optional): Worker id. Defaults to None.

Returns: Union[float, tuple[int, str, float]]: A GDST score when used in a single process; or, a tuple of (worker id, model path, GDST score) when used in multiple processes.

writer_multiprocessing

def writer_multiprocessing(save_path: str, num: int, q: Queue) -> List[Tuple[str, float]]

Writes results from multiple processes to a list.

Args: save_path (str): The path to the saved results. num (int): The number of results. q (Queue): A multiprocessing queue.

Returns: list[tuple[str, float]]: A list of (model path, GDST score) tuples.

get_gdst_score_batch

def get_gdst_score_batch(batch: List[Tuple[str, str]], num_samples: int = 10000, seed: int = 42, save_path: str = None, num_workers: int = 1) -> List[Tuple[str, float]]

Runs the GDST on a batch of models.

Args: batch (list[tuple[str, str]]): A list of (model path, model type) tuples. num_samples (int, optional): Number of embeddings used for evaluation. Defaults to 10000. seed (int, optional): Random seed. Defaults to 42. save_path (str, optional): Save the results to save_path. Defaults to None. num_workers (int, optional): Number of parallel processes used. Defaults to 1.

Returns: list[tuple[str, float]]: A list of (model path, GDST score) tuples.

gdst_eval

def gdst_eval(batch: List[Tuple[str, str]], num_runs: int = 20, num_samples: int = 1000, save_folder: str = None, num_workers: int = 10) -> List[Tuple[str, List[float]]]

Runs the GDST on a batch of models for multiple times.

Args: batch (list[tuple[str, str]]): A list of (model path, model type) tuples. num_runs (int, optional): Number of runs. Defaults to 20. num_samples (int, optional): Number of embeddings used for evaluation. Defaults to 1000. save_folder (str, optional): Folder to save the results from each run. Defaults to None. num_workers (int, optional): Number of parallel processes used. Defaults to 10.

Returns: list[tuple[str, list[float]]]: A list of (model path, GDST scores from num_runs) tuples.

get_topk_embed

def get_topk_embed(i: int, K: int, embed: np.ndarray, dist: str = 'cosine') -> Tuple[np.ndarray, np.ndarray]

Gets the nearest K embedding indexes to the i-th embedding.

Args: i (int): The index for the query embedding. K (int): The number of nearest embeddings to select. embed (np.ndarray): An array of embedding vectors dist (str, optional): The distance function used. Defaults to "cosine".

Returns: tuple[np.ndarray, np.ndarray]: K indexes of nearest embeddings and the corresponding similarities.

find_kneighbors

def find_kneighbors(region_array: List[Tuple[str, int, int]], index: int, k: int) -> List[int]

Finds the indexes of the K nearest regions of a query region on genome.

region_array must be sorted, and all regions are on the same chromosome.

Args: region_array (list[tuple[str, int, int]]): A list of (chromosome, start position, end position) tuples. index (int): The index of the query region. k (int): Specifies the number of nearest neighbors of the query region.

Returns: list[int]: A list of indexes of the K nearest neighbors in region_array.

calculate_overlap_bins

def calculate_overlap_bins(local_idx: int, K: int, chromo: str, region_array: List[Tuple[str, int, int]], region2index: dict[str, int], embed_rep: np.ndarray, res: int = 10, dist: str = 'cosine', same_chromo: bool = True) -> np.ndarray

Calculates the overlap ratios for a region.

Calculates the overlap ratios for a region between its K-nearest neighbor set obtained using genome distance and its K-nearest neighbor set obtained using embedding distance. If res < K, then calculates ratios for size res1, res2, ..., min(res*n, K).

Args: local_idx (int): The local index of a region on its chromosome. K (int): Specifies the number of nearest neighbors. chromo (str): Chromosome. region_array (list[tuple[str, int, int]]): A list of (chromosome, start position, end position) tuples. region2index (dict[str, int]): A dictionary of (region, index). embed_rep (np.ndarray): An array of embedding vectors. res (int, optional): Resolution. Size of neighborhood set. Defaults to 10. dist (str, optional): Distance function. Defaults to "cosine". same_chromo (bool, optional): Whether to find nearest neighbors on the same chromosome in the embedding space. Defaults to True.

Returns: np.ndarray: An array of overlap ratios.

cal_snpr

def cal_snpr(ratio_embed: np.ndarray, ratio_random: np.ndarray) -> np.ndarray

Calculates SNPR values.

Parameters:

• ratio_embed (np.ndarray): Overlap ratios for query embeddings.
• ratio_random (np.ndarray): Overlap ratios for random embeddings.

Returns:

• np.ndarray: SNPR values.

worker_func

def worker_func(i: int, K: int, chromo: str, region_array: List[Tuple[str, int, int]], embed_type: str, resolution: int, dist: str) -> np.ndarray

Wrapper for calculate_overlap_bins

Args: i (int): The local index of a region on its chromosome. K (int): Specifies the number of nearest neighbors. chromo (str): Chromosome. region_array (list[tuple[str, int, int]]): A list of (chromosome, start position, end position) tuples. embed_type (str): Embedding type, "region2vec" or "base". resolution (int): Resolution. dist (str): Distance function.

Returns: np.ndarray: An array of overlap ratios.

init_worker

def init_worker(embed_rep: np.ndarray, ref_embed: np.ndarray, region2index: Dict[str, int]) -> None

Initializes data used by workers.

Args: embed_rep (np.ndarray): Query embeddings. ref_embed (np.ndarray): Random embeddings. region2index (dict[str, int]): A region to index dictionary.

get_npt_score

def get_npt_score(model_path: str, embed_type: str, K: int, num_samples: int = 100, seed: int = 0, resolution: int = 10, dist: str = 'cosine', num_workers: int = 10) -> Dict[str, Union[int, np.ndarray, str]]

Runs the NPT on a model.

If num_samples > 0, then randomly sample num_samples regions proportional from each chromosome. If num_samples == 0, all regions are used in the test. If K > resolution, then returns an array of NPT scores; otherwise, returns one NPT score.

Args: model_path (str): The path to a model. embed_type (str): The model type: "region2vec" or "base". K (int): Specifies the number of nearest neighbors. num_samples (int, optional): Number of embeddings used for evaluation. Defaults to 100. seed (int, optional): Random seed. Defaults to 0. resolution (int, optional): Resolution of a neighborhood set. Defaults to 10. dist (str, optional): Distance function. Defaults to "cosine". num_workers (int, optional): Number of parallel processes used. Defaults to 10.

Returns: dict[str, Union[int, np.ndarray, str]]: NPT results in a dictionary "K": K, "Avg_qNPR": Average NPR ratios for query embeddings, "Avg_rNPR": Average NPR ratios for random embeddings,, "SNPR": SNPR values, "Resolution": Resolution, "Path": Model path,

writer_multiprocessing

def writer_multiprocessing(save_path: str, num: int, q: Queue) -> List[Tuple[str, float]]

Writes results from multiple processes to a list.

Args: save_path (str): The path to the saved results. num (int): The number of results. q (Queue): A multiprocessing queue.

Returns: list[tuple[str, float]]: A list of (model path, NPT score) tuples.

get_npt_score_batch

def get_npt_score_batch(batch: List[Tuple[str, str]], K: int, num_samples: int = 100, num_workers: int = 10, seed: int = 0, resolution: int = 10, dist: str = 'cosine', save_path: str = None) -> List[Dict[str, Union[int, np.ndarray, str]]]

Runs the NPT on a batch of models.

Args: batch (list[tuple[str, str]]): A list of (model path, model type) tuples. K (int): Specifies the number of nearest neighbors. num_samples (int, optional): Number of embeddings used for evaluation. Defaults to 100. num_workers (int, optional): Number of parallel processes used. Defaults to 10. seed (int, optional): Random seed. Defaults to 0. resolution (int, optional): Resolution of a neighborhood set. Defaults to 10. dist (str, optional): Distance function. Defaults to "cosine". save_path (str, optional): Save the results to save_path. Defaults to None.

Returns: list[dict[str, Union[int, np.ndarray, str]]]: A list of dictionaries of NPT results.

npt_eval

def npt_eval(batch: List[Tuple[str, str]], K: int, num_samples: int = 100, num_workers: int = 10, num_runs: int = 20, resolution: int = 10, dist: str = 'cosine', save_folder: str = None) -> List[Tuple[str, np.ndarray, int]]

Runs the NPT on a batch of models for multiple times.

Args: batch (list[tuple[str, str]]): A list of (model path, model type) tuples. K (int): Specifies the number of nearest neighbors. num_samples (int, optional): Number of embeddings used for evaluation. Defaults to 100. num_workers (int, optional): Number of parallel processes used. Defaults to 10. num_runs (int, optional): Number of runs. Defaults to 20. resolution (int, optional): Resolution of a neighborhood set. Defaults to 10. dist (str, optional): Distance function. Defaults to "cosine". save_folder (str, optional): Folder to save the results from each run. Defaults to None.

Returns: list[tuple[str, np.ndarray, int]]: A list of (model path, snprs from num_runs, resolution) tuples.

Attributes

• var_dict = {}

get_rct_score

def get_rct_score(path: str, embed_type: str, bin_path: str, out_dim: int = -1, cv_num: int = 5, seed: int = 42, num_workers: int = 10) -> float

Runs the RCT on a model.

Args: path (str): The path to a model. embed_type (str): The type of the model: "region2vec" or "base". bin_path (str): The path to a binary embedding model. out_dim (int, optional): Output dimension of a prediction. Defaults to -1, and out_dim is the same as the dimension of a binary embedding. cv_num (int, optional): Number of folds in cross-validation. Defaults to 5. seed (int, optional): Random seed. Defaults to 42. num_workers (int, optional): Number of parallel processes used. Defaults to 10.

Returns: float: the RCT score.

reconstruction_batch

def reconstruction_batch(batch: list[tuple[str, str, str]], cv_num: int, out_dim: int = -1, seed: int = 42, save_path: str = None, num_workers: int = 10) -> list[tuple[str, float]]

Runs the RCT on a batch of models.

Args: batch (list[tuple[str, str, str]]): A list of (model path, model type, binary embedding path) tuples. Model type could be "region2vec" or "base". cv_num (int): Number of folds in cross-validation. Defaults to 5. out_dim (int, optional): Output dimension of a prediction. Defaults to -1, and out_dim is the same as the dimension of a binary embedding. seed (int, optional): Random seed. Defaults to 42. save_path (str, optional): Save the results to save_path. Defaults to None. num_workers (int, optional): Number of parallel processes used. Defaults to 10.

Returns: list[tuple[str, float]]: A list of (model path, RCT score) tuples.

rct_eval

def rct_eval(batch: list[tuple[str, str, str]], num_runs: int = 5, cv_num: int = 5, out_dim: int = -1, save_folder: str = None, num_workers: int = 10) -> list[tuple[str, list[float]]]

Runs the RCT on a batch of models for multiple times.

Args: batch (list[tuple[str, str, str]]): A list of (model path, model type, binary embedding path) tuples. Model type could be "region2vec" or "base". num_runs (int, optional): Number of runs. Defaults to 5. cv_num (int, optional): Number of folds in cross-validation. Defaults to 5. out_dim (int, optional): Output dimension of a prediction. Defaults to -1, and out_dim is the same as the dimension of a binary embedding. save_folder (str, optional): Folder to save the results from each run. Defaults to None. num_workers (int, optional): Number of parallel processes used. Defaults to 10.

Returns: list[tuple[str, list[float]]]: A list of (model path, RCT scores from num_runs) tuples.

genome_distance

def genome_distance(u: Tuple[int, int], v: Tuple[int, int]) -> float

Computes the genome distance between two regions.

Assumes that the two regions, u and v, are on the same chromosome.

Args: u (tuple[int, int]): A region denoted by its start and end positions. v (tuple[int, int]): A region denoted by its start and end positions.

Returns: float: The genome distance between the two regions.

cosine_distance

def cosine_distance(x: np.ndarray, y: np.ndarray) -> float

Calculates the cosine distance between two embedding vectors.

Args: x (np.ndarray): An embedding vector. y (np.ndarray): An embedding vector.

Returns: float: The cosine distance between two embedding vectors.

BaseEmbeddings

class BaseEmbeddings(embeddings, vocab)

Wraps embeddings and the corresponding regions in one object.

Attributes: embeddings (np.ndarray): Region embedding vectors. vocab (list[str]): A list of regions in the format of chr:start-end.

Attributes

• embeddings
• vocab

get_bin_embeddings

def get_bin_embeddings(universe_file: str, tokenized_files: list[str]) -> BaseEmbeddings

Gets a BaseEmbeddings object for binary embeddings.

Args: universe_file (str): The path to a universe file. tokenized_files (list[str]): A list of tokoenized BED files (in full paths).

Returns: BaseEmbeddings: A BaseEmbeddings object for binary embeddings.

get_pca_embeddings

def get_pca_embeddings(bin_embed_obj: BaseEmbeddings, dim: int, kwargs: Dict[str, Union[int, float]] = {}) -> BaseEmbeddings

Gets PCA embeddings from binary embeddings.

Args: bin_embed_obj (BaseEmbeddings): A BaseEmbeddings object for binary embeddings. dim (int): Number of dimensions for PCA embeddings. kwargs (dict[str, Union[int, float]], optional): Parameters passed to PCA. Defaults to {}.

Returns: BaseEmbeddings: A BaseEmbeddings object for PCA embeddings.

get_umap_embeddings

def get_umap_embeddings(bin_embed_obj: BaseEmbeddings, dim: int, kwargs: Dict[str, Union[int, float]] = {}) -> BaseEmbeddings

Gets UMAP embeddings from binary embeddings.

Args: bin_embed_obj (BaseEmbeddings): A BaseEmbeddings object for binary embeddings. dim (int): Number of dimensions for UMAP embeddings. kwargs (dict[str, Union[int, float]], optional): Parameters passed to UMAP. Defaults to {}.

Returns: BaseEmbeddings: A BaseEmbeddings object for UMAP embeddings.

save_base_embeddings

def save_base_embeddings(base_embed_obj: BaseEmbeddings, file_name: str) -> None

Saves the BaseEmbeddings object to disk.

Args: base_embed_obj (BaseEmbeddings): A BaseEmbeddings object. file_name (str): Save the BaseEmbeddings object to file_name.

load_base_embeddings

def load_base_embeddings(path: str) -> Tuple[np.ndarray, List[str]]

Loads a BaseEmbeddings object.

Args: path (str): The path to a BaseEmbeddings object.

Returns: tuple[np.ndarray, list[str]]: Embedding vectors and the corresponding region list.

load_genomic_embeddings

def load_genomic_embeddings(model_path: str, embed_type: str = 'exmodel') -> Tuple[np.ndarray, List[str]]

Loads genomic region embeddings based on the type.

Args: model_path (str): The path to a saved model, or a huggingface repo of a model. embed_type (str, optional): The model type. Defaults to "region2vec". Can be "region2vec", "base", or "exmodel".

Returns: tuple[np.ndarray, list[str]]: Embedding vectors and the corresponding region list.

sort_key

def sort_key(x: str) -> Tuple[int, int]

Extracts chromosome in number and the start position of a region.

Args: x (str): A region in the chr:start-end position.

Returns: tuple[int, int]: Chromosome in number and the start position.

get_vocab

def get_vocab(model_path: str, type: str = 'base', ordered: bool = True) -> List[str]

Gets vocab from a model.

Args: model_path (str): The path to a saved model. type (str, optional): The embedding type. Defaults to "base". ordered (bool, optional): Choose whether to sort the regions. Defaults to True.

Returns: list[str]: A list of regions.

write_vocab

def write_vocab(vocab: List[str], file_name: str) -> None

Writes a list of regions to a file.

Args: vocab (list[str]): A list of regions in the format of chr:start-end. file_name (str): Saves vocab as file_name.