mandos.search.chembl._activity_search

Complexity

Total Complexity

25

Size/Duplication

Total Lines	141
Duplicated Lines	0 %

Importance

Changes

0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	_ActivitySearch.allowed_assay_types()	0	3	1
A	_ActivitySearch.query()	0	12	3
A	_ActivitySearch.__init__()	0	17	1
A	_ActivitySearch._extract()	0	31	4
F	_ActivitySearch.should_include()	0	35	16

import logging

Missing module docstring

import abc
from dataclasses import dataclass
from typing import Sequence, Set, Optional

from pocketutils.core.dot_dict import NestedDotDict

Unable to import 'pocketutils.core.dot_dict'

from mandos.model.chembl_api import ChemblApi
from mandos.model.chembl_support import ChemblCompound, AssayType
from mandos.model.chembl_support.chembl_target_graphs import ChemblTargetGraph
from mandos.model.taxonomy import Taxonomy
from mandos.search.chembl._protein_search import ProteinHit, ProteinSearch, H

logger = logging.getLogger("mandos")


@dataclass(frozen=True, order=True, repr=True)
class _ActivityHit(ProteinHit):
    taxon_id: int
    taxon_name: str
    src_id: str


class _ActivitySearch(ProteinSearch[H], metaclass=abc.ABCMeta):
    """
    Search for ``activity``.
    """

    def __init__(

Too many arguments (10/5)

        self,

Wrong hanging indentation before block (add 4 spaces).

        key: str,

Wrong hanging indentation before block (add 4 spaces).

        api: ChemblApi,

Wrong hanging indentation before block (add 4 spaces).

        taxa: Sequence[Taxonomy],

Wrong hanging indentation before block (add 4 spaces).

        traversal_strategy: str,

Wrong hanging indentation before block (add 4 spaces).

        allowed_target_types: Set[str],

Wrong hanging indentation before block (add 4 spaces).

        min_confidence_score: Optional[int],

Wrong hanging indentation before block (add 4 spaces).

        allowed_relations: Set[str],

Wrong hanging indentation before block (add 4 spaces).

        min_pchembl: Optional[float],

Wrong hanging indentation before block (add 4 spaces).

        banned_flags: Set[str],

Wrong hanging indentation before block (add 4 spaces).

    ):
        super().__init__(key, api, taxa, traversal_strategy, allowed_target_types)
        self.min_confidence_score = min_confidence_score
        self.allowed_relations = allowed_relations
        self.min_pchembl = min_pchembl
        self.banned_flags = banned_flags

    @classmethod
    def allowed_assay_types(cls) -> Set[str]:

Missing function or method docstring

        raise NotImplementedError()

    def query(self, parent_form: ChemblCompound) -> Sequence[NestedDotDict]:

Missing function or method docstring

        filters = dict(
            parent_molecule_chembl_id=parent_form.chid,
            assay_type__iregex=self._set_to_regex(self.allowed_assay_types()),
            standard_relation__iregex=self._set_to_regex(self.allowed_relations),
            pchembl_value__isnull=None if self.min_pchembl is None else False,
            target_organism__isnull=None if len(self.taxa) == 0 else False,
        )
        # I'd rather not figure out how the API interprets None, so remove them
        filters = {k: v for k, v in filters.items() if v is not None}
        return list(self.api.activity.filter(**filters))

    def should_include(

Missing function or method docstring

        self, lookup: str, compound: ChemblCompound, data: NestedDotDict, target: ChemblTargetGraph

The argument compound seems to be unused.

Wrong hanging indentation before block (add 4 spaces).

    ) -> bool:
        if (
            (

Too many boolean expressions in if statement (9/5)

Wrong hanging indentation before block (add 4 spaces).

                data.get_as("data_validity_comment", lambda s: s.lower())
                in {s.lower() for s in self.banned_flags}
            )
            or (data.req_as("standard_relation", str) not in self.allowed_relations)

Wrong hanging indentation before block (add 4 spaces).

            or (data.req_as("assay_type", str) not in self.allowed_assay_types())

Wrong hanging indentation before block (add 4 spaces).

            or (len(self.taxa) > 0 and not self.is_in_taxa(data.get_as("target_tax_id", int)))

Wrong hanging indentation before block (add 4 spaces).

            or (self.min_pchembl is not None and data.get("pchembl_value") is None)

Wrong hanging indentation before block (add 4 spaces).

            or self.min_pchembl is not None

Wrong hanging indentation before block (add 4 spaces).

            and data.req_as("pchembl_value", float) < self.min_pchembl

Wrong hanging indentation before block (add 4 spaces).

        ):
            return False
        if data.get("data_validity_comment") is not None:
            logger.warning(

Use lazy % formatting in logging functions

                f"Activity annotation for {lookup} has flag '{data.get('data_validity_comment')} (ok)"

This line is too long as per the coding-style (102/100).

            )
        # The `target_organism` doesn't always match the `assay_organism`
        # Ex: see assay CHEMBL823141 / document CHEMBL1135642 for homo sapiens in xenopus laevis
        # However, it's often something like yeast expressing a human / mouse / etc receptor
        # So there's no need to filter by it
        assay = self.api.assay.get(data.req_as("assay_chembl_id", str))
        if target.type.name.lower() not in {s.lower() for s in self.allowed_target_types}:
            logger.warning(f"Excluding {target} with type {target.type}")

Use lazy % formatting in logging functions

            return False
        confidence_score = assay.get("confidence_score")
        if self.min_confidence_score is not None:
            if confidence_score is None or confidence_score < self.min_confidence_score:
                return False
        # Some of these are non-protein types
        # And if it's unknown, we don't know what to do with it
        return True

    def _extract(self, lookup: str, compound: ChemblCompound, data: NestedDotDict) -> NestedDotDict:
        # we know these exist from the query
        organism = data.req_as("target_organism", str)
        tax_id = data.req_as("target_tax_id", int)
        if len(self.taxa) == 0:
            tax_id, tax_name = tax_id, organism

Assigning the same variable 'tax_id' to itself

        else:
            taxes = {tax.req(tax_id) for tax in self.taxa if tax.contains(tax_id)}
            tax = next(iter(taxes))
            if len(taxes) > 1:
                logger.warning(f"Multiple matches for taxon {tax_id}: {taxes}; using {tax}")

Use lazy % formatting in logging functions

            if organism != tax.name:
                logger.warning(f"Target organism {organism} is not {tax.name}")

Use lazy % formatting in logging functions

            tax_id = tax.id
            tax_name = tax.name
        return NestedDotDict(
            dict(
                record_id=data.req_as("activity_id", str),
                origin_inchikey=lookup,
                matched_inchikey=compound.inchikey,
                compound_id=compound.chid,
                compound_name=compound.name,
                taxon_id=tax_id,
                taxon_name=tax_name,
                pchembl=data.req_as("pchembl_value", float),
                std_type=data.req_as("standard_type", str),
                src_id=data.req_as("src_id", str),
                exact_target_id=data.req_as("target_chembl_id", str),
                tissue=data.get_as("tissue", str),
                cell_type=data.get_as("cell_type", str),
                subcellular_region=data.get("subcellular_region", str),
            )
        )

    assay_type: AssayType
    tissue: Optional[str]
    cell_type: Optional[str]
    subcellular_region: Optional[str]


__all__ = ["_ActivitySearch", "_ActivityHit"]