|
1
|
|
|
from __future__ import annotations |
|
|
|
|
|
|
2
|
|
|
import abc |
|
3
|
|
|
import sys |
|
4
|
|
|
import re |
|
5
|
|
|
from pathlib import Path |
|
6
|
|
|
from typing import Sequence, Type, Set |
|
7
|
|
|
|
|
8
|
|
|
from mandos.model import MandosResources |
|
9
|
|
|
from mandos.model.chembl_api import ChemblApi |
|
10
|
|
|
from mandos.model.chembl_support.chembl_targets import ( |
|
11
|
|
|
DagTargetLinkType, |
|
12
|
|
|
ChemblTarget, |
|
13
|
|
|
TargetRelationshipType, |
|
14
|
|
|
TargetType, |
|
15
|
|
|
) |
|
16
|
|
|
|
|
17
|
|
|
|
|
18
|
|
|
class TargetTraversalStrategy(metaclass=abc.ABCMeta): |
|
|
|
|
|
|
19
|
|
|
"""""" |
|
20
|
|
|
|
|
21
|
|
|
@classmethod |
|
22
|
|
|
def api(cls) -> ChemblApi: |
|
|
|
|
|
|
23
|
|
|
raise NotImplementedError() |
|
24
|
|
|
|
|
25
|
|
|
def traverse(self, target: ChemblTarget) -> Sequence[ChemblTarget]: |
|
|
|
|
|
|
26
|
|
|
return self.__call__(target) |
|
27
|
|
|
|
|
28
|
|
|
def __call__(self, target: ChemblTarget) -> Sequence[ChemblTarget]: |
|
29
|
|
|
""" |
|
30
|
|
|
|
|
31
|
|
|
Returns: |
|
32
|
|
|
|
|
33
|
|
|
""" |
|
34
|
|
|
raise NotImplementedError() |
|
35
|
|
|
|
|
36
|
|
|
|
|
37
|
|
|
class StandardTargetTraversalStrategy(TargetTraversalStrategy, metaclass=abc.ABCMeta): |
|
|
|
|
|
|
38
|
|
|
"""""" |
|
39
|
|
|
|
|
40
|
|
|
@classmethod |
|
41
|
|
|
def edges(cls) -> Set[DagTargetLinkType]: |
|
|
|
|
|
|
42
|
|
|
raise NotImplementedError() |
|
43
|
|
|
|
|
44
|
|
|
@classmethod |
|
45
|
|
|
def read(cls, path: Path) -> Set[DagTargetLinkType]: |
|
|
|
|
|
|
46
|
|
|
lines = [ |
|
47
|
|
|
line |
|
48
|
|
|
for line in path.read_text(encoding="utf8").splitlines() |
|
49
|
|
|
if not line.startswith("#") and len(line.strip()) > 0 |
|
50
|
|
|
] |
|
51
|
|
|
return cls.parse(lines) |
|
52
|
|
|
|
|
53
|
|
|
@classmethod |
|
54
|
|
|
def parse(cls, lines: Sequence[str]) -> Set[DagTargetLinkType]: |
|
|
|
|
|
|
55
|
|
|
pat_type = re.compile(r"([a-z_]+)") |
|
56
|
|
|
pat_rel = re.compile(r"((?:->)|(?:<-)|(?:==)|(?:~~))") |
|
57
|
|
|
pat_words = re.compile(r"(?:words: *([^|]+(?:\|[^|]+)+))?") |
|
58
|
|
|
pat = re.compile(f"^ *{pat_type} *{pat_rel} *{pat_type} *{pat_words} *$") |
|
59
|
|
|
to_rel = { |
|
60
|
|
|
"->": TargetRelationshipType.superset_of, |
|
61
|
|
|
"<-": TargetRelationshipType.subset_of, |
|
62
|
|
|
"~~": TargetRelationshipType.overlaps_with, |
|
63
|
|
|
"==": TargetRelationshipType.equivalent_to, |
|
64
|
|
|
} |
|
65
|
|
|
edges = set() |
|
66
|
|
|
for line in lines: |
|
67
|
|
|
match = pat.fullmatch(line) |
|
68
|
|
|
if match is None: |
|
69
|
|
|
raise AssertionError(f"Could not parse line '{line}'") |
|
70
|
|
|
try: |
|
71
|
|
|
source = TargetType[match.group(1).lower()] |
|
72
|
|
|
rel = TargetRelationshipType[to_rel[match.group(2).lower()]] |
|
73
|
|
|
target = TargetType[match.group(3).lower()] |
|
74
|
|
|
words = None if match.group(4) == "" else match.group(4).split("|") |
|
75
|
|
|
except KeyError: |
|
76
|
|
|
raise AssertionError(f"Could not parse line '{line}'") |
|
77
|
|
|
edges.add(DagTargetLinkType(source, rel, target, words)) |
|
78
|
|
|
return edges |
|
79
|
|
|
|
|
80
|
|
|
def __call__(self, target: ChemblTarget) -> Sequence[ChemblTarget]: |
|
81
|
|
|
""" |
|
82
|
|
|
Returns: |
|
83
|
|
|
""" |
|
84
|
|
|
if not target.type.is_traversable: |
|
85
|
|
|
return [target] |
|
86
|
|
|
found = target.traverse(self.edges) |
|
87
|
|
|
return [f.target for f in found if f.is_end] |
|
88
|
|
|
|
|
89
|
|
|
|
|
90
|
|
|
class TargetTraversalStrategy0(StandardTargetTraversalStrategy, metaclass=abc.ABCMeta): |
|
|
|
|
|
|
91
|
|
|
"""""" |
|
92
|
|
|
|
|
93
|
|
|
@classmethod |
|
94
|
|
|
def edges(cls) -> Set[DagTargetLinkType]: |
|
95
|
|
|
return cls.read(MandosResources.path("strategies", "strategy0.txt")) |
|
96
|
|
|
|
|
97
|
|
|
|
|
98
|
|
|
class TargetTraversalStrategy1(StandardTargetTraversalStrategy, metaclass=abc.ABCMeta): |
|
|
|
|
|
|
99
|
|
|
"""""" |
|
100
|
|
|
|
|
101
|
|
|
@classmethod |
|
102
|
|
|
def edges(cls) -> Set[DagTargetLinkType]: |
|
103
|
|
|
return cls.read(MandosResources.path("strategies", "strategy1.txt")) |
|
104
|
|
|
|
|
105
|
|
|
|
|
106
|
|
|
class TargetTraversalStrategy2(StandardTargetTraversalStrategy, metaclass=abc.ABCMeta): |
|
107
|
|
|
""" |
|
108
|
|
|
Traverse the DAG up and down, following only desired links |
|
109
|
|
|
Some links from complex to complex group are "overlaps with" |
|
110
|
|
|
ex: CHEMBL4296059 |
|
111
|
|
|
it's also rare to need going from a selectivity group "down" to complex group / family / etc. |
|
112
|
|
|
usually they have a link upwards |
|
113
|
|
|
so... |
|
114
|
|
|
If it's a single protein, it's too risk to traverse up into complexes |
|
115
|
|
|
That's because lots of proteins *occasionally* make complexes, and there are some weird ones |
|
116
|
|
|
BUT We want to catch some obvious cases like GABA A subunits |
|
117
|
|
|
ChEMBL calls many of these "something subunit something" |
|
118
|
|
|
This is the only time we'll allow going directly from protein to complex |
|
119
|
|
|
In this case, we'll also disallow links form protein to family, |
|
120
|
|
|
just because we're pretty sure it's a subunit |
|
121
|
|
|
But we can go from single protein to complex to complex group to family |
|
122
|
|
|
""" |
|
123
|
|
|
|
|
124
|
|
|
@classmethod |
|
125
|
|
|
def edges(cls) -> Set[DagTargetLinkType]: |
|
126
|
|
|
return cls.read(MandosResources.path("strategies", "strategy2.txt")) |
|
127
|
|
|
|
|
128
|
|
|
|
|
129
|
|
|
class TargetTraversalStrategies: |
|
130
|
|
|
""" |
|
131
|
|
|
Factory. |
|
132
|
|
|
""" |
|
133
|
|
|
|
|
134
|
|
|
@classmethod |
|
135
|
|
|
def by_name(cls, fully_qualified: str, api: ChemblApi) -> TargetTraversalStrategy: |
|
136
|
|
|
""" |
|
137
|
|
|
For dependency injection. |
|
138
|
|
|
|
|
139
|
|
|
Args: |
|
140
|
|
|
fully_qualified: |
|
141
|
|
|
api: |
|
142
|
|
|
|
|
143
|
|
|
Returns: |
|
144
|
|
|
|
|
145
|
|
|
""" |
|
146
|
|
|
s = fully_qualified |
|
|
|
|
|
|
147
|
|
|
mod = s[: s.rfind(".")] |
|
148
|
|
|
clz = s[s.rfind(".") :] |
|
149
|
|
|
x = getattr(sys.modules[mod], clz) |
|
|
|
|
|
|
150
|
|
|
return cls.create(x, api) |
|
151
|
|
|
|
|
152
|
|
|
@classmethod |
|
153
|
|
|
def strategy0(cls, api: ChemblApi) -> TargetTraversalStrategy: |
|
|
|
|
|
|
154
|
|
|
return cls.create(TargetTraversalStrategy0, api) |
|
155
|
|
|
|
|
156
|
|
|
@classmethod |
|
157
|
|
|
def strategy1(cls, api: ChemblApi) -> TargetTraversalStrategy: |
|
|
|
|
|
|
158
|
|
|
return cls.create(TargetTraversalStrategy1, api) |
|
159
|
|
|
|
|
160
|
|
|
@classmethod |
|
161
|
|
|
def strategy2(cls, api: ChemblApi) -> TargetTraversalStrategy: |
|
|
|
|
|
|
162
|
|
|
return cls.create(TargetTraversalStrategy2, api) |
|
163
|
|
|
|
|
164
|
|
|
@classmethod |
|
165
|
|
|
def create(cls, clz: Type[TargetTraversalStrategy], api: ChemblApi) -> TargetTraversalStrategy: |
|
166
|
|
|
""" |
|
167
|
|
|
Factory method. |
|
168
|
|
|
|
|
169
|
|
|
Args: |
|
170
|
|
|
clz: |
|
171
|
|
|
api: |
|
172
|
|
|
|
|
173
|
|
|
Returns: |
|
174
|
|
|
|
|
175
|
|
|
""" |
|
176
|
|
|
|
|
177
|
|
|
class X(clz): |
|
|
|
|
|
|
178
|
|
|
@classmethod |
|
179
|
|
|
def api(cls) -> ChemblApi: |
|
|
|
|
|
|
180
|
|
|
return api |
|
181
|
|
|
|
|
182
|
|
|
X.__name__ = clz.__name__ |
|
183
|
|
|
return X() |
|
184
|
|
|
|
|
185
|
|
|
|
|
186
|
|
|
__all__ = ["TargetTraversalStrategy", "TargetTraversalStrategies"] |
|
187
|
|
|
|