dmyersturnbull /
mandos
| Total Complexity | 58 |
| Total Lines | 379 |
| Duplicated Lines | 0 % |
| Changes | 0 | ||
Complex classes like mandos.model.apis.hmdb_api often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
| 1 | import abc |
||
|
introduced
by
|
|||
| 2 | import math |
||
| 3 | import time |
||
| 4 | import urllib |
||
| 5 | from dataclasses import dataclass |
||
| 6 | from datetime import datetime |
||
| 7 | from functools import cached_property |
||
| 8 | from pathlib import Path |
||
| 9 | from typing import Mapping, NamedTuple, Optional, Sequence |
||
| 10 | from urllib import request |
||
| 11 | |||
| 12 | import decorateme |
||
| 13 | import regex |
||
| 14 | from pocketutils.core.chars import Chars |
||
| 15 | from pocketutils.core.dot_dict import NestedDotDict |
||
| 16 | from pocketutils.core.enums import FlagEnum |
||
| 17 | from pocketutils.core.query_utils import QueryExecutor |
||
| 18 | from pocketutils.tools.common_tools import CommonTools |
||
| 19 | |||
| 20 | from mandos.model import Api, CompoundNotFoundError |
||
| 21 | from mandos.model.settings import QUERY_EXECUTORS, SETTINGS |
||
| 22 | from mandos.model.utils.setup import logger |
||
| 23 | |||
| 24 | |||
| 25 | class _Prop(NamedTuple): |
||
| 26 | kind: str |
||
| 27 | source: str |
||
| 28 | |||
| 29 | |||
| 30 | _prefixes = dict(M=1e6, mM=1e3, µM=1, uM=1, nM=1e-3, pM=1e-6, fM=1e-9) |
||
| 31 | |||
| 32 | _PREDICTED_PROPERTIES = [ |
||
| 33 | _Prop("average_mass", "ChemAxon"), |
||
| 34 | _Prop("logp", "ALOGPS"), |
||
| 35 | _Prop("logs", "ALOGPS"), |
||
| 36 | _Prop("solubility", "ALOGPS"), |
||
| 37 | _Prop("pka_strongest_acidic", "ChemAxon"), |
||
| 38 | _Prop("polar_surface_area", "ChemAxon"), |
||
| 39 | _Prop("polarizability", "ChemAxon"), |
||
| 40 | _Prop("physiological_charge", "ChemAxon"), |
||
| 41 | ] |
||
| 42 | |||
| 43 | _RULES = [ |
||
| 44 | _Prop("rule_of_five", "ChemAxon"), |
||
| 45 | _Prop("ghose_filter", "ChemAxon"), |
||
| 46 | _Prop("veber_rule", "ChemAxon"), |
||
| 47 | _Prop("mddr_like_rule", "ChemAxon"), |
||
| 48 | ] |
||
| 49 | |||
| 50 | _p1 = regex.compile(r"^([0-9.]+ +\(([0-9.]+) *\- *([0-9.]+)\)$", flags=regex.V1) |
||
| 51 | _p2 = regex.compile(r"^([0-9.]+) +\+\/\- +([0-9.]+)$", flags=regex.V1) |
||
| 52 | |||
| 53 | |||
| 54 | class HmdbCompoundLookupError(CompoundNotFoundError): |
||
| 55 | """ """ |
||
| 56 | |||
| 57 | |||
| 58 | class ConcentrationBound(NamedTuple): |
||
| 59 | mean: float |
||
| 60 | lower: float |
||
| 61 | upper: float |
||
| 62 | |||
| 63 | @property |
||
| 64 | def std(self) -> float: |
||
| 65 | return self.upper / 2 - self.lower / 2 |
||
| 66 | |||
| 67 | @property |
||
| 68 | def is_symmetric(self) -> bool: |
||
| 69 | return math.isclose(self.upper - self.mean, self.mean - self.lower) |
||
| 70 | |||
| 71 | |||
| 72 | @dataclass(frozen=True, repr=True, order=True) |
||
| 73 | class HmdbProperty: |
||
| 74 | kind: str |
||
| 75 | source: str |
||
| 76 | value: str |
||
| 77 | |||
| 78 | |||
| 79 | @dataclass(frozen=True, repr=True, order=True) |
||
| 80 | class HmdbDisease: |
||
| 81 | name: str |
||
| 82 | omim_id: str |
||
| 83 | n_refs: int |
||
| 84 | |||
| 85 | |||
| 86 | class PersonAge(FlagEnum): |
||
| 87 | unknown = () |
||
| 88 | adults = () |
||
| 89 | children = () |
||
| 90 | |||
| 91 | |||
| 92 | class PersonSex(FlagEnum): |
||
| 93 | unknown = () |
||
| 94 | male = () |
||
| 95 | female = () |
||
| 96 | |||
| 97 | |||
| 98 | @dataclass(frozen=True, repr=True, order=True) |
||
| 99 | class HmdbConcentration: |
||
| 100 | specimen: str |
||
| 101 | ages: PersonAge |
||
| 102 | sexes: PersonSex |
||
| 103 | condition: Optional[str] |
||
| 104 | micromolar: Optional[ConcentrationBound] |
||
| 105 | mg_per_kg: Optional[ConcentrationBound] |
||
| 106 | |||
| 107 | def __post_init__(self): |
||
| 108 | if (self.mg_per_kg is None) + (self.micromolar is None) != 1: |
||
| 109 | raise AssertionError( |
||
| 110 | f"Provided both micromolar ({self.micromolar})" |
||
| 111 | + f" and mg/kg ({self.mg_per_kg}), or neither" |
||
| 112 | ) |
||
| 113 | |||
| 114 | @cached_property |
||
| 115 | def format_value(self) -> str: |
||
| 116 | return f"{self._value}{Chars.narrownbsp}{self._unit}" |
||
| 117 | |||
| 118 | @cached_property |
||
| 119 | def format_value_pm(self) -> str: |
||
| 120 | v, u, s = self._value, self._unit, Chars.narrownbsp |
||
| 121 | return f"{v.mean}{Chars.plusminus}{v.std}{s}{u}" |
||
| 122 | |||
| 123 | @cached_property |
||
| 124 | def format_value_range(self) -> str: |
||
| 125 | v, u, s = self._value, self._unit, Chars.narrownbsp |
||
| 126 | return f"{v.mean}{s}({v.lower}{Chars.en}{v.upper}){s}{u}" |
||
| 127 | |||
| 128 | @property |
||
| 129 | def _value(self) -> ConcentrationBound: |
||
| 130 | if self.mg_per_kg is not None: |
||
| 131 | return self.mg_per_kg |
||
| 132 | return self.micromolar |
||
| 133 | |||
| 134 | @property |
||
| 135 | def _unit(self) -> str: |
||
| 136 | if self.mg_per_kg is not None: |
||
| 137 | return " mg/kg" |
||
| 138 | return " µmol/L" |
||
| 139 | |||
| 140 | |||
| 141 | class HmdbData: |
||
| 142 | def __init__(self, data: NestedDotDict): |
||
| 143 | self._data = data |
||
| 144 | |||
| 145 | @property |
||
| 146 | def cid(self) -> str: |
||
| 147 | return self._data.req_as("metabolite.accession", str) |
||
| 148 | |||
| 149 | @property |
||
| 150 | def inchi(self) -> str: |
||
| 151 | return self._data.req_as("metabolite.inchi", str) |
||
| 152 | |||
| 153 | @property |
||
| 154 | def inchikey(self) -> str: |
||
| 155 | return self._data.req_as("metabolite.inchikey", str) |
||
| 156 | |||
| 157 | @property |
||
| 158 | def smiles(self) -> str: |
||
| 159 | return self._data.req_as("metabolite.smiles", str) |
||
| 160 | |||
| 161 | @property |
||
| 162 | def cas(self) -> str: |
||
| 163 | return self._data.req_as("metabolite.cas_registry_number", str) |
||
| 164 | |||
| 165 | @property |
||
| 166 | def drugbank_id(self) -> Optional[str]: |
||
| 167 | return self._data.get_as("metabolite.inchikey", str) |
||
| 168 | |||
| 169 | @property |
||
| 170 | def pubchem_id(self) -> Optional[str]: |
||
| 171 | return self._data.get_as("metabolite.pubchem_compound_id", str) |
||
| 172 | |||
| 173 | @property |
||
| 174 | def create_date(self) -> datetime: |
||
| 175 | return datetime.fromisoformat(self._data.req_as("metabolite.creation_date", str)) |
||
| 176 | |||
| 177 | @property |
||
| 178 | def mod_date(self) -> datetime: |
||
| 179 | return datetime.fromisoformat(self._data.req_as("metabolite.update_date", str)) |
||
| 180 | |||
| 181 | @cached_property |
||
| 182 | def predicted_properties(self) -> Sequence[HmdbProperty]: |
||
| 183 | data = self._data.get("metabolite.predicted_properties", []) |
||
| 184 | return [ |
||
| 185 | HmdbProperty(kind=x["kind"], source=x["source"], value=x["value"]) |
||
| 186 | for x in data |
||
| 187 | if _Prop(x["kind"], x["source"]) in _PREDICTED_PROPERTIES |
||
| 188 | ] |
||
| 189 | |||
| 190 | @cached_property |
||
| 191 | def rules(self) -> Mapping[str, bool]: |
||
| 192 | data = self._data.get("metabolite.predicted_properties", []) |
||
| 193 | return { |
||
| 194 | r["kind"]: CommonTools.parse_bool_flex(r["value"]) |
||
| 195 | for r in data |
||
| 196 | if (r["kind"], r["source"]) in _RULES |
||
| 197 | } |
||
| 198 | |||
| 199 | @cached_property |
||
| 200 | def diseases(self) -> Sequence[HmdbDisease]: |
||
| 201 | data = self._data.get_list_as("metabolite.diseases", NestedDotDict) |
||
| 202 | return [HmdbDisease(d["name"], d["omim_id"], len(d.get("references", []))) for d in data] |
||
| 203 | |||
| 204 | @cached_property |
||
| 205 | def specimens(self) -> Sequence[str]: |
||
| 206 | return self._data.get_list_as("metabolite.biological_properties.biospecimen_locations", str) |
||
| 207 | |||
| 208 | @cached_property |
||
| 209 | def tissue_locations(self) -> Sequence[str]: |
||
| 210 | return self._data.get_list_as("metabolite.biological_properties.tissue_locations", str) |
||
| 211 | |||
| 212 | @cached_property |
||
| 213 | def normal_concentrations(self) -> Sequence[HmdbConcentration]: |
||
| 214 | data = self._data.get_list_as("metabolite.normal_concentrations", NestedDotDict, []) |
||
| 215 | results = [] |
||
| 216 | for d in data: |
||
| 217 | x = self._new_conc(d) |
||
| 218 | if x is not None: |
||
| 219 | results.append(x) |
||
| 220 | return results |
||
| 221 | |||
| 222 | def _new_conc(self, x: NestedDotDict) -> Optional[HmdbConcentration]: |
||
| 223 | specimen = x["biospecimen"] |
||
| 224 | # both can be "Not Specified" |
||
| 225 | ages = { |
||
| 226 | "Adult": PersonAge.adults, |
||
| 227 | "Children": PersonAge.children, |
||
| 228 | "Both": PersonAge.adults | PersonAge.children, |
||
| 229 | }.get(x.get_as("subject_age", str, "").split(" ")[0], PersonAge.unknown) |
||
| 230 | sexes = { |
||
| 231 | "Male": PersonSex.male, |
||
| 232 | "Female": PersonSex.female, |
||
| 233 | "Both": PersonSex.female | PersonSex.male, |
||
| 234 | }.get(x.get_as("subject_sex", str, ""), PersonSex.unknown) |
||
| 235 | condition = ( |
||
| 236 | None |
||
| 237 | if x.get("subject_condition") == "Normal" |
||
| 238 | else x.get_as("patient_information", str, "") |
||
| 239 | ) |
||
| 240 | value, units = x.get_as("concentration_value", str), x.get_as("concentration_units", str) |
||
| 241 | if value is None or len(value) == 0: |
||
| 242 | logger.trace(f"Discarding {x} with empty value") |
||
| 243 | return None |
||
| 244 | if units not in ["uM", "mg/kg"]: |
||
| 245 | logger.trace(f"Discarding {x} with units '{units}'") |
||
| 246 | return None |
||
| 247 | bound = self._parse_conc(value) |
||
| 248 | if bound is None: |
||
| 249 | logger.warning(f"Could not parse concentration {value} (units: {units})") |
||
| 250 | logger.trace(f"Full data: {x}") |
||
| 251 | return None |
||
| 252 | return HmdbConcentration( |
||
| 253 | specimen=specimen, |
||
| 254 | ages=ages, |
||
| 255 | sexes=sexes, |
||
| 256 | condition=condition, |
||
| 257 | micromolar=bound if units == "uM" else None, |
||
| 258 | mg_per_kg=bound if units == "mg/kg" else None, |
||
| 259 | ) |
||
| 260 | |||
| 261 | def _parse_conc(self, value: str) -> Optional[ConcentrationBound]: |
||
| 262 | m: regex.Match = _p1.fullmatch(value) |
||
| 263 | if m is not None: |
||
| 264 | return ConcentrationBound(*m.groups()) |
||
| 265 | m: regex.Match = _p2.fullmatch(value) |
||
| 266 | if m is not None: |
||
| 267 | v, std = m.groups() |
||
| 268 | return ConcentrationBound(v, v - std, v + std) |
||
| 269 | return None |
||
| 270 | |||
| 271 | @cached_property |
||
| 272 | def abnormal_concentrations(self) -> Sequence[HmdbConcentration]: |
||
| 273 | return self._data.get("metabolite.normal_concentrations", []) |
||
| 274 | |||
| 275 | |||
| 276 | @decorateme.auto_repr_str() |
||
| 277 | class HmdbApi(Api, metaclass=abc.ABCMeta): |
||
| 278 | def fetch(self, hmdb_id: str) -> HmdbData: |
||
| 279 | raise NotImplementedError() |
||
| 280 | |||
| 281 | |||
| 282 | @decorateme.auto_repr_str() |
||
| 283 | class QueryingHmdbApi(HmdbApi): |
||
| 284 | def __init__(self, executor: QueryExecutor = QUERY_EXECUTORS.hmdb): |
||
| 285 | self._executor = executor |
||
| 286 | |||
| 287 | def fetch(self, inchikey_or_hmdb_id: str) -> HmdbData: |
||
| 288 | logger.debug(f"Downloading HMDB data for {inchikey_or_hmdb_id}") |
||
| 289 | # e.g. https://hmdb.ca/metabolites/HMDB0001925.xml |
||
| 290 | cid = None |
||
| 291 | if inchikey_or_hmdb_id.startswith("HMDB"): |
||
| 292 | cid = inchikey_or_hmdb_id |
||
| 293 | else: |
||
| 294 | time.sleep(SETTINGS.hmdb_query_delay_min) # TODO |
||
| 295 | url = f"https://hmdb.ca/unearth/q?query={inchikey_or_hmdb_id}&searcher=metabolites" |
||
| 296 | try: |
||
| 297 | res = urllib.request.urlopen(url) |
||
| 298 | url_ = res.geturl() |
||
| 299 | logger.trace(f"Got UR {url_} from {url}") |
||
| 300 | cid = url_.split("/")[-1] |
||
| 301 | if not cid.startswith("HMDB"): |
||
| 302 | raise ValueError(f"Invalid CID {cid} from URL {url_}") |
||
| 303 | except Exception: |
||
| 304 | raise HmdbCompoundLookupError(f"No HMDB match for {inchikey_or_hmdb_id}") |
||
| 305 | url = f"https://hmdb.ca/metabolites/{cid}.xml" |
||
| 306 | try: |
||
| 307 | data = self._executor(url) |
||
| 308 | except Exception: |
||
| 309 | raise HmdbCompoundLookupError(f"No HMDB match for {inchikey_or_hmdb_id} ({cid})") |
||
| 310 | return HmdbData(self._to_json(data)) |
||
| 311 | |||
| 312 | def _to_json(self, xml) -> NestedDotDict: |
||
| 313 | response = {} |
||
| 314 | for child in list(xml): |
||
| 315 | if len(list(child)) > 0: |
||
| 316 | response[child.tag] = self._to_json(child) |
||
| 317 | else: |
||
| 318 | response[child.tag] = child.text or "" |
||
| 319 | return NestedDotDict(response) |
||
| 320 | |||
| 321 | def _query(self, url: str) -> str: |
||
| 322 | data = self._executor(url) |
||
| 323 | tt = self._executor.last_time_taken |
||
| 324 | wt, qt = tt.wait.total_seconds(), tt.query.total_seconds() |
||
| 325 | bts = int(len(data) * 8 / 1024) |
||
| 326 | logger.trace(f"Queried {bts} kb from {url} in {qt:.1} s with {wt:.1} s of wait") |
||
| 327 | return data |
||
| 328 | |||
| 329 | |||
| 330 | @decorateme.auto_repr_str() |
||
| 331 | class CachingHmdbApi(HmdbApi): |
||
| 332 | def __init__( |
||
| 333 | self, query: Optional[QueryingHmdbApi], cache_dir: Path = SETTINGS.hmdb_cache_path |
||
| 334 | ): |
||
| 335 | self._query = query |
||
| 336 | self._cache_dir = cache_dir |
||
| 337 | |||
| 338 | def path(self, inchikey_or_hmdb_id: str) -> Path: |
||
| 339 | return self._cache_dir / f"{inchikey_or_hmdb_id}.json.gz" |
||
| 340 | |||
| 341 | def fetch(self, inchikey_or_hmdb_id: str) -> HmdbData: |
||
| 342 | path = self.path(inchikey_or_hmdb_id) |
||
| 343 | if path.exists(): |
||
| 344 | return HmdbData(NestedDotDict.read_json(path)) |
||
| 345 | else: |
||
| 346 | data = self._query.fetch(inchikey_or_hmdb_id) |
||
| 347 | path = self.path(data.cid) |
||
| 348 | data._data.write_json(path, mkdirs=True) |
||
| 349 | logger.info(f"Saved HMDB metabolite {data.cid}") |
||
| 350 | self._write_links(data) |
||
| 351 | return data |
||
| 352 | |||
| 353 | def _write_links(self, data: HmdbData) -> None: |
||
| 354 | path = self.path(data.cid) |
||
| 355 | # these all have different prefixes, so it's ok |
||
| 356 | aliases = [ |
||
| 357 | data.inchikey, |
||
| 358 | *[ell for ell in [data.cas, data.pubchem_id, data.drugbank_id] if ell is not None], |
||
| 359 | ] |
||
| 360 | for alias in aliases: |
||
| 361 | link = self.path(alias) |
||
| 362 | link.unlink(missing_ok=True) |
||
| 363 | path.link_to(link) |
||
| 364 | logger.debug(f"Added aliases {','.join([str(s) for s in aliases])} ⇌ {data.cid} ({path})") |
||
| 365 | |||
| 366 | |||
| 367 | __all__ = [ |
||
| 368 | "HmdbApi", |
||
| 369 | "QueryingHmdbApi", |
||
| 370 | "CachingHmdbApi", |
||
| 371 | "HmdbProperty", |
||
| 372 | "ConcentrationBound", |
||
| 373 | "HmdbData", |
||
| 374 | "PersonSex", |
||
| 375 | "PersonAge", |
||
| 376 | "HmdbConcentration", |
||
| 377 | "HmdbDisease", |
||
| 378 | "HmdbCompoundLookupError", |
||
| 379 | ] |
||
| 380 |
Loading history...