mandos.model.apis.g2p_api

Complexity

Total Complexity

24

Size/Duplication

Total Lines	134
Duplicated Lines	0 %

Importance

Changes

0

1 Function

Rating	Name	Duplication	Size	Complexity
A	_oint()	0	4	4

import abc

Missing module docstring

import time

The import time seems to be unused.

from datetime import datetime
from pathlib import Path
from typing import Optional, Type

import numpy as np

Unable to import 'numpy'

import orjson

Unable to import 'orjson'

import pandas as pd

Unable to import 'pandas'

from typeddfs import TypedDfs, TypedDf

Unable to import 'typeddfs'

from pocketutils.tools.common_tools import CommonTools

Unable to import 'pocketutils.tools.common_tools'

from mandos.model import Api, CompoundNotFoundError
from mandos.model.apis.g2p_data import G2pData, G2pInteraction, TrueFalseUnknown

LIGANDS_URL = "https://www.guidetopharmacology.org/DATA/ligand_id_mapping.tsv"
INTERACTIONS_URL = "https://www.guidetopharmacology.org/DATA/interactions.tsv"


def _oint(x: str) -> Optional[int]:

Argument name "x" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

    if x is None or isinstance(x, str) and x.strip() == "":
        return None
    return int(x)


LigandDf = (
    TypedDfs.typed("LigandDf")
    .require("Ligand id", dtype=int)
    .require("Name", "Type", "Approved", "PubChem CID", "InChIKey", dtype=str)
).build()


InteractionDf = (
    TypedDfs.typed("InteractionDf")
    .require(
        "target", "target_id", "target_gene_symbol", "target_uniprot", "target_species", dtype=str
    )
    .require("ligand", dtype=str)
    .require("ligand_id", dtype=int)
    .require("type", "action", dtype=str)
    .require("selectivity", "endogenous", "primary_target", dtype=str)
    .require("affinity_units", dtype=str)
    .require("affinity_median", dtype=np.float64)
).build()


class G2pApi(Api, metaclass=abc.ABCMeta):

Missing class docstring

    def fetch(self, inchikey: str) -> G2pData:

Missing function or method docstring

        raise NotImplementedError()


class CachedG2pApi(G2pApi, metaclass=abc.ABCMeta):

Missing class docstring

    def __init__(self, cache_path: Path):
        self.cache_path = Path(cache_path)
        self.ligands: LigandDf = None
        self.interactions: InteractionDf = None

    def fetch(self, inchikey: str) -> G2pData:

Empty method docstring

        """ """
        series = self.ligands[self.ligands["inchikey"] == inchikey]
        if len(series) == 0:
            raise CompoundNotFoundError(f"G2P ligand {inchikey} not found")
        basic = dict(CommonTools.only(series).to_dict())
        g2pid = int(basic["Ligand id"])
        interactions = [
            self._convert_interaction(series)
            for series in self.interactions[self.interactions["ligand_id"] == g2pid]
        ]
        return G2pData(
            inchikey=basic["InChIKey"],
            g2pid=g2pid,
            name=basic["ligand"],
            type=basic["Type"],
            approved=TrueFalseUnknown.parse(basic["Approved"]),
            pubchem_id=_oint(basic["PubChem ID"]),
            interactions=interactions,
        )

    def download(self, force: bool = False) -> None:

Missing function or method docstring

        if self.ligands is None or self.interactions is None or force:
            # always download both together -- we don't want them non-synced
            if self.ligands_path.exists() and self.interactions_path.exists() and not force:
                self.ligands = LigandDf.read_file(self.ligands_path)
                self.interactions = InteractionDf.read_file(self.ligands_path)
            else:
                self.ligands = LigandDf.read_file(LIGANDS_URL, sep="\t")
                self.ligands.write_file(self.ligands_path)
                self.interactions = InteractionDf.read_file(INTERACTIONS_URL, sep="\t")
                self.interactions.write_file(self.interactions_path)
                info = dict(dt_downloaded=datetime.now().isoformat())
                info = orjson.dumps(info).decode(encoding="utf8")
                (self.cache_path / "info.json").write_text(info)

    @property
    def ligands_path(self) -> Path:

Missing function or method docstring

        return self.cache_path / "ligands.feather"

    @property
    def interactions_path(self) -> Path:

Missing function or method docstring

        return self.cache_path / "interactions.feather"

    def _load_file(self, clazz: Type[TypedDf], path: Path, url: str) -> pd.DataFrame:
        if path.exists():

Unnecessary "else" after "return"

            return clazz.read_file(self.ligands_path, sep="\t")
        else:
            df = clazz.read_file(url, sep="\t")

Variable name "df" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            df.to_csv(self.ligands_path, sep="\t")
            return df

    def _convert_interaction(self, series: pd.Series) -> G2pInteraction:

This method could be written as a function/class method.

        d = dict(series.to_dict())

Variable name "d" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        sel_map = {
            "Selective": TrueFalseUnknown.true,
            "Non-selective": TrueFalseUnknown.false,
            "Not Determined": TrueFalseUnknown.unknown,
        }
        d["selectivity"] = sel_map.get(d["selectivity"], TrueFalseUnknown.unknown)
        d["primary_target"] = TrueFalseUnknown.parse(d["primary_target"])
        d["endogenous"] = TrueFalseUnknown.parse(d["endogenous"])
        return G2pInteraction(**d)

    def __repr__(self):
        loaded = "not loaded" if self.ligands is None else f"n={len(self.ligands)}"
        return f"{self.__class__.__name__}({self.cache_path} : {loaded})"

    def __str__(self):
        return repr(self)

    def __eq__(self, other):
        raise NotImplementedError(f"Cannot compare {self.__class__.__name__}")


_all__ = ["G2pApi", "CachedG2pApi"]