Spaces:
Sleeping
Sleeping
File size: 6,176 Bytes
d9df210 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 |
from MassSpecGym.massspecgym.utils import MyopicMCES
import numpy as np
import tqdm
from multiprocessing import Pool
import os
import pandas as pd
class Compute_Myopic_MCES:
mces_compute = MyopicMCES()
def compute_mces(tar_cand):
target, cand = tar_cand
dist = Compute_Myopic_MCES.mces_compute(target, cand)
return (tar_cand, dist)
def compute_mces_parallel(target_cand_list, n_processes=25):
with Pool(processes=n_processes) as pool:
results = list(tqdm.tqdm(pool.imap(Compute_Myopic_MCES.compute_mces, target_cand_list), total=len(target_cand_list)))
return results
class Compute_Myopic_MCES_timeout:
mces_compute = MyopicMCES()
@staticmethod
def compute_mces(tar_cand):
target, cand = tar_cand
dist = Compute_Myopic_MCES.mces_compute(target, cand)
return (tar_cand, dist)
@staticmethod
def compute_mces_parallel(target_cand_list, n_processes=35, timeout=60): # timeout in seconds
results = []
with Pool(processes=n_processes) as pool:
async_results = [
pool.apply_async(Compute_Myopic_MCES.compute_mces, args=(tar_cand,))
for tar_cand in target_cand_list
]
for async_res in tqdm.tqdm(async_results, total=len(target_cand_list)):
try:
result = async_res.get(timeout=timeout)
except Exception as e:
# You can log the error or return a default value
result = (None, f"Timeout or error")
results.append(result)
return results
def get_result_files(exp_dir, spec_type, views_type):
files = os.listdir(exp_dir)
mass_result = ''
form_result = ''
for f in files:
try:
_, s, views = f.split('_')
except:
continue
if s == spec_type and views == views_type:
print(exp_dir / f)
files = os.listdir(exp_dir / f)
for fr in files:
if 'mass_result' in fr:
mass_result = exp_dir / f / fr
elif 'result' in fr:
form_result = exp_dir / f/ fr
return mass_result, form_result
# get target
def get_target(candidates, labels):
return np.array(candidates)[labels][0]
# get mol rank at 1
def get_top_cand(candidates, scores):
return candidates[np.argmax(scores)]
# split into hit rates
def convert_rank_to_hit_rates(row, rank_col ,top_k=[1,5,20]):
top_k_hits ={}
rank = row[rank_col]
for k in top_k:
if rank <= k:
top_k_hits[f'{rank_col}-hit_rate@{k}'] = 1
else:
top_k_hits[f'{rank_col}-hit_rate@{k}'] = 0
return pd.Series(top_k_hits)
#################### Rank aggregation #######################
from collections import defaultdict
import numpy as np
from scipy.stats import rankdata
def borda_count(candidates, score_lists, target):
scores = defaultdict(int)
N = len(candidates)
for score_list in score_lists:
ranked_list = sorted(zip(candidates, score_list), key=lambda x: x[1], reverse=True)
for rank, (mol, _) in enumerate(ranked_list, start=1):
scores[mol] += N - rank + 1
ranked_candidates = [mol for mol, _ in sorted(scores.items(), key=lambda x: x[1], reverse=True)]
return ranked_candidates.index(target) + 1 if target in ranked_candidates else None
def average_rank(candidates, score_lists, target):
rank_sums = defaultdict(list)
for score_list in score_lists:
ranked_list = sorted(zip(candidates, score_list), key=lambda x: x[1], reverse=True)
for rank, (mol, _) in enumerate(ranked_list, start=1):
rank_sums[mol].append(rank)
avg_ranks = {mol: np.mean(ranks) for mol, ranks in rank_sums.items()}
ranked_candidates = [mol for mol, _ in sorted(avg_ranks.items(), key=lambda x: x[1])]
return ranked_candidates.index(target) + 1 if target in ranked_candidates else None
def reciprocal_rank_aggregation(candidates, score_lists, target):
scores = defaultdict(float)
for score_list in score_lists:
ranked_list = sorted(zip(candidates, score_list), key=lambda x: x[1], reverse=True)
for rank, (mol, _) in enumerate(ranked_list, start=1):
scores[mol] += 1 / rank
ranked_candidates = [mol for mol, _ in sorted(scores.items(), key=lambda x: x[1], reverse=True)]
return ranked_candidates.index(target) + 1 if target in ranked_candidates else None
def weighted_voting(candidates, score_lists, weights, target):
scores = defaultdict(float)
for weight, score_list in zip(weights, score_lists):
ranked_list = sorted(zip(candidates, score_list), key=lambda x: x[1], reverse=True)
for rank, (mol, _) in enumerate(ranked_list, start=1):
scores[mol] += weight / rank
ranked_candidates = [mol for mol, _ in sorted(scores.items(), key=lambda x: x[1], reverse=True)]
return ranked_candidates.index(target) + 1 if target in ranked_candidates else None
def median_rank(candidates, score_lists, target):
rank_sums = defaultdict(list)
for score_list in score_lists:
ranked_list = sorted(zip(candidates, score_list), key=lambda x: x[1], reverse=True)
for rank, (mol, _) in enumerate(ranked_list, start=1):
rank_sums[mol].append(rank)
median_ranks = {mol: np.median(ranks) for mol, ranks in rank_sums.items()}
ranked_candidates = [mol for mol, _ in sorted(median_ranks.items(), key=lambda x: x[1])]
return ranked_candidates.index(target) + 1 if target in ranked_candidates else None
def score_based_aggregation(candidates, score_lists, target):
scores = defaultdict(list)
for score_list in score_lists:
for mol, score in zip(candidates, score_list):
scores[mol].append(score)
avg_scores = {mol: np.mean(vals) for mol, vals in scores.items()}
ranked_candidates = [mol for mol, _ in sorted(avg_scores.items(), key=lambda x: x[1], reverse=True)]
return ranked_candidates.index(target) + 1 if target in ranked_candidates else None
|