1
|
|
|
""" |
2
|
|
|
Calculations of concordance between annotations. |
3
|
|
|
""" |
4
|
|
|
import abc |
5
|
|
|
import math |
6
|
|
|
from collections import defaultdict |
7
|
|
|
from typing import Collection, Sequence |
8
|
|
|
|
9
|
|
|
import numpy as np |
|
|
|
|
10
|
|
|
|
11
|
|
|
from mandos.analysis import AnalysisUtils as Au |
12
|
|
|
from mandos.analysis import SimilarityDf |
13
|
|
|
from mandos.model.hits import AbstractHit |
14
|
|
|
|
15
|
|
|
# note that most of these math functions are much faster than their numpy counterparts |
16
|
|
|
# if we're not broadcasting, it's almost always better to use them |
17
|
|
|
# some are more accurate, too |
18
|
|
|
# e.g. we're using fsum rather than sum |
19
|
|
|
|
20
|
|
|
|
21
|
|
|
class MatrixCalculator(metaclass=abc.ABCMeta): |
|
|
|
|
22
|
|
|
def calc(self, hits: Sequence[AbstractHit]) -> SimilarityDf: |
|
|
|
|
23
|
|
|
raise NotImplemented() |
|
|
|
|
24
|
|
|
|
25
|
|
|
|
26
|
|
|
class JPrimeMatrixCalculator(MatrixCalculator): |
|
|
|
|
27
|
|
|
def calc(self, hits: Sequence[AbstractHit]) -> SimilarityDf: |
28
|
|
|
inchikey_to_hits = Au.hit_multidict(hits, "origin_inchikey") |
29
|
|
|
data = defaultdict(dict) |
30
|
|
|
for (c1, hits1), (c2, hits2) in zip(inchikey_to_hits.items(), inchikey_to_hits.items()): |
|
|
|
|
31
|
|
|
data[c1][c2] = self._j_prime(hits1, hits2) |
32
|
|
|
return SimilarityDf.from_dict(data) |
33
|
|
|
|
34
|
|
|
def _j_prime(self, hits1: Collection[AbstractHit], hits2: Collection[AbstractHit]) -> float: |
35
|
|
|
sources = {h.data_source for h in hits1}.intersection({h.data_source for h in hits2}) |
36
|
|
|
if len(sources) == 0: |
37
|
|
|
return np.nan |
38
|
|
|
values = [ |
39
|
|
|
self._jx( |
40
|
|
|
[h for h in hits1 if h.data_source == source], |
41
|
|
|
[h for h in hits1 if h.data_source == source], |
42
|
|
|
) |
43
|
|
|
for source in sources |
44
|
|
|
] |
45
|
|
|
return float(math.fsum(values) / len(values)) |
46
|
|
|
|
47
|
|
|
def _jx(self, hits1: Collection[AbstractHit], hits2: Collection[AbstractHit]) -> float: |
48
|
|
|
pair_to_weights = Au.weights_of_pairs(hits1, hits2) |
49
|
|
|
values = [self._wedge(ca, cb) / self._vee(ca, cb) for ca, cb in pair_to_weights.values()] |
50
|
|
|
return float(math.fsum(values) / len(values)) |
51
|
|
|
|
52
|
|
|
def _wedge(self, ca: float, cb: float) -> float: |
|
|
|
|
53
|
|
|
return math.sqrt(Au.elle(ca) * Au.elle(cb)) |
54
|
|
|
|
55
|
|
|
def _vee(self, ca: float, cb: float) -> float: |
|
|
|
|
56
|
|
|
return Au.elle(ca) + Au.elle(cb) - math.sqrt(Au.elle(ca) * Au.elle(cb)) |
57
|
|
|
|
58
|
|
|
|
59
|
|
|
__all__ = ["MatrixCalculator", "JPrimeMatrixCalculator"] |
60
|
|
|
|