mandos.entry.tools.searchers.MemoizedInputCompounds.read_file() - Code Metrics - Inspection of "fix: misc" - dmyersturnbull/mandos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 2e1b6b...3a0c28 )

by Douglas

created 2021-11-04 01:01 UTC

MemoizedInputCompounds.read_file() A

↳ Parent: mandos.entry.tools.searchers

Complexity

Conditions

Size

Total Lines	7
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	7
nop	2
dl	0
loc	7
rs	10
c	0
b	0
f	0

"""
Run searches and write files.
"""

from __future__ import annotations

import functools
import time
from dataclasses import dataclass
from datetime import timedelta
from pathlib import Path
from typing import Sequence

from pocketutils.core.exceptions import IllegalStateError

from typeddfs import Checksums, TypedDfs


from mandos.model import CompoundNotFoundError
from mandos.model.hit_dfs import HitDf
from mandos.model.hits import AbstractHit
from mandos.model.search_caches import SearchCache
from mandos.model.searches import Search, SearchError
from mandos.model.settings import SETTINGS
from mandos.model.utils.setup import logger


def _fix_cols(df):

    return df.rename(columns={s: s.lower() for s in df.columns})


InputCompoundsDf = (
    TypedDfs.typed("InputCompoundsDf")
    .require("inchikey")
    .reserve("inchi", "smiles", "compound_id", dtype=str)
    .post(_fix_cols)
    .strict(cols=False)
    .secure()
).build()


class MemoizedInputCompounds:

    @classmethod
    @functools.cache

    def read_file(cls, path: Path) -> InputCompoundsDf:

        logger.debug(f"Reading compounds from {path}")
        df = InputCompoundsDf.read_file(path)

        logger.info(f"Read {len(df)} compounds from {path}")
        return df


@dataclass(frozen=True, repr=True, order=True)

class SearchReturnInfo:
    n_kept: int
    n_processed: int
    n_errored: int
    time_taken: timedelta


@dataclass(frozen=True, repr=True)
class Searcher:
    """
    Executes one or more searches and saves the results.
    Create and use once.
    """

    what: Search
    input_df: InputCompoundsDf
    to: Path

    proceed: bool
    restart: bool

    def search(self) -> SearchReturnInfo:

        """
        Performs the search, and writes data.
        """
        inchikeys = self.input_df["inchikey"].unique()
        if self.is_complete:
            logger.info(f"{self.to} already complete")
            return SearchReturnInfo(
                n_kept=len(inchikeys), n_processed=0, n_errored=0, time_taken=timedelta(seconds=0)
            )
        logger.info(f"Will save every {SETTINGS.save_every} compounds")
        logger.info(f"Writing {self.what.key} to {self.to}")
        annotes = []
        compounds_run = set()
        cache = SearchCache(self.to, inchikeys, restart=self.restart, proceed=self.proceed)
        # refresh so we know it's (no longer) complete
        # this would only happen if we're forcing this -- which is not currently allowed
        (
            Checksums()
            .load_dirsum_of_file(self.to, missing_ok=True)
            .remove(self.to, missing_ok=True)
            .write(rm_if_empty=True)
        )
        t0, n0, n_proc, n_err, n_annot = time.monotonic(), cache.at, 0, 0, 0

        while True:
            try:
                compound = cache.next()

            except StopIteration:
                break
            try:
                with logger.contextualize(compound=compound):
                    x = self.what.find(compound)

                annotes.extend(x)
            except CompoundNotFoundError:
                logger.info(f"Compound {compound} not found for {self.what.key}")
                x = []

                n_err += 1
            except Exception:
                raise SearchError(
                    f"Failed {self.what.key} [{self.what.search_class}] on compound {compound}",
                    compound=compound,
                    search_key=self.what.key,
                    search_class=self.what.search_class,
                )
            compounds_run.add(compound)
            logger.debug(f"Found {len(x)} {self.what.search_name()} annotations for {compound}")
            n_annot += len(x)
            n_proc += 1
            # logging, caching, and such:
            on_nth = cache.at % SETTINGS.save_every == SETTINGS.save_every - 1
            is_last = cache.at == len(inchikeys) - 1
            if on_nth or is_last:
                logger.log(
                    "NOTICE" if is_last else "INFO",
                    f"Found {len(annotes)} {self.what.search_name()} annotations"
                    + f" for {cache.at} of {len(inchikeys)} compounds",
                )
                self._save(annotes, done=is_last)
            cache.save(*compounds_run)  # CRITICAL -- do this AFTER saving
        # done!
        i1, t1 = cache.at, time.monotonic()

        assert i1 == len(inchikeys)
        cache.kill()
        logger.success(f"Wrote {self.what.key} to {self.to}")
        return SearchReturnInfo(
            n_kept=n0, n_processed=n_proc, n_errored=n_err, time_taken=timedelta(seconds=t1 - t0)
        )

    @property
    def is_partial(self) -> bool:

        return self.to.exists() and not self.is_complete

    @property
    def is_complete(self) -> bool:

        done = self.to in Checksums().load_dirsum_of_file(self.to)
        if done and not self.to.exists():
            raise IllegalStateError(f"{self.to} marked complete but does not exist")
        return done

    def _save(self, hits: Sequence[AbstractHit], *, done: bool) -> None:
        df = HitDf.from_hits(hits)

        # keep all of the original extra columns from the input
        # e.g. if the user had 'inchi' or 'smiles' or 'pretty_name'
        # if "origin_inchikey" not in df.columns:
        for extra_col in [c for c in self.input_df.columns if c != "inchikey"]:
            extra_mp = self.input_df.set_index("inchikey")[extra_col].to_dict()
            df[extra_col] = df["origin_inchikey"].map(extra_mp.get)
        # write the file
        df: HitDf = HitDf.of(df)

        params = self.what.get_params()
        df = df.set_attrs(**params, key=self.what.key)

        df.write_file(self.to.resolve(), mkdirs=True, attrs=True, dir_hash=done)
        logger.debug(f"Saved {len(df)} rows to {self.to}")


__all__ = ["Searcher", "InputCompoundsDf", "SearchReturnInfo", "MemoizedInputCompounds"]


1			"""
2			Run searches and write files.
3			"""
4
5			from __future__ import annotations
6
7			import functools
8			import time
9			from dataclasses import dataclass
10			from datetime import timedelta
11			from pathlib import Path
12			from typing import Sequence
13
14			from pocketutils.core.exceptions import IllegalStateError
			0 ignored issues – show introduced 2021-08-02 23:39 UTC by Report Bug Copy Issue Report Unable to import 'pocketutils.core.exceptions' Loading history...
15			from typeddfs import Checksums, TypedDfs
			0 ignored issues – show introduced 2021-08-07 00:15 UTC by Report Bug Copy Issue Report Unable to import 'typeddfs' Loading history...
16
17			from mandos.model import CompoundNotFoundError
18			from mandos.model.hit_dfs import HitDf
19			from mandos.model.hits import AbstractHit
20			from mandos.model.search_caches import SearchCache
21			from mandos.model.searches import Search, SearchError
22			from mandos.model.settings import SETTINGS
23			from mandos.model.utils.setup import logger
24
25
26			def _fix_cols(df):
			0 ignored issues – show Coding Style Naming introduced 2021-08-02 23:39 UTC by Report Bug Copy Issue Report Argument name "df" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
27			return df.rename(columns={s: s.lower() for s in df.columns})
28
29
30			InputCompoundsDf = (
31			TypedDfs.typed("InputCompoundsDf")
32			.require("inchikey")
33			.reserve("inchi", "smiles", "compound_id", dtype=str)
34			.post(_fix_cols)
35			.strict(cols=False)
36			.secure()
37			).build()
38
39
40			class MemoizedInputCompounds:
			0 ignored issues – show introduced 2021-04-03 02:07 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
41			@classmethod
42			@functools.cache
			0 ignored issues – show Bug introduced 2021-11-04 01:02 UTC by Report Bug Copy Issue Report The Module `functools` does not seem to have a member named `cache`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
43			def read_file(cls, path: Path) -> InputCompoundsDf:
			0 ignored issues – show introduced 2021-10-28 02:14 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
44			logger.debug(f"Reading compounds from {path}")
45			df = InputCompoundsDf.read_file(path)
			0 ignored issues – show Coding Style Naming introduced 2021-03-21 02:08 UTC by Report Bug Copy Issue Report 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) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
46			logger.info(f"Read {len(df)} compounds from {path}")
47			return df
48
49
50			@dataclass(frozen=True, repr=True, order=True)
			0 ignored issues – show introduced 2021-10-02 18:29 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
51			class SearchReturnInfo:
52			n_kept: int
53			n_processed: int
54			n_errored: int
55			time_taken: timedelta
56
57
58			@dataclass(frozen=True, repr=True)
59			class Searcher:
60			"""
61			Executes one or more searches and saves the results.
62			Create and use once.
63			"""
64
65			what: Search
66			input_df: InputCompoundsDf
67			to: Path
			0 ignored issues – show Coding Style Naming introduced 2021-10-28 02:14 UTC by Report Bug Copy Issue Report Attribute name "to" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
68			proceed: bool
69			restart: bool
70
71			def search(self) -> SearchReturnInfo:
			0 ignored issues – show Comprehensibility introduced 2021-10-28 02:14 UTC by Report Bug Copy Issue Report This function exceeds the maximum number of variables (16/15). Loading history...
72			"""
73			Performs the search, and writes data.
74			"""
75			inchikeys = self.input_df["inchikey"].unique()
76			if self.is_complete:
77			logger.info(f"{self.to} already complete")
78			return SearchReturnInfo(
79			n_kept=len(inchikeys), n_processed=0, n_errored=0, time_taken=timedelta(seconds=0)
80			)
81			logger.info(f"Will save every {SETTINGS.save_every} compounds")
82			logger.info(f"Writing {self.what.key} to {self.to}")
83			annotes = []
84			compounds_run = set()
85			cache = SearchCache(self.to, inchikeys, restart=self.restart, proceed=self.proceed)
86			# refresh so we know it's (no longer) complete
87			# this would only happen if we're forcing this -- which is not currently allowed
88			(
89			Checksums()
90			.load_dirsum_of_file(self.to, missing_ok=True)
91			.remove(self.to, missing_ok=True)
92			.write(rm_if_empty=True)
93			)
94			t0, n0, n_proc, n_err, n_annot = time.monotonic(), cache.at, 0, 0, 0
			0 ignored issues – show Coding Style Naming introduced 2021-10-28 02:14 UTC by Report Bug Copy Issue Report Variable name "t0" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history... Coding Style Naming introduced 2021-10-28 02:14 UTC by Report Bug Copy Issue Report Variable name "n0" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
95			while True:
96			try:
97			compound = cache.next()
			0 ignored issues – show Bug introduced 2021-09-29 03:10 UTC by Report Bug Copy Issue Report `cache.next` does not seem to be callable. Loading history...
98			except StopIteration:
99			break
100			try:
101			with logger.contextualize(compound=compound):
102			x = self.what.find(compound)
			0 ignored issues – show Coding Style Naming introduced 2021-09-29 03:10 UTC by Report Bug Copy Issue Report Variable name "x" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
103			annotes.extend(x)
104			except CompoundNotFoundError:
105			logger.info(f"Compound {compound} not found for {self.what.key}")
106			x = []
			0 ignored issues – show Coding Style Naming introduced 2021-09-29 16:53 UTC by Report Bug Copy Issue Report Variable name "x" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
107			n_err += 1
108			except Exception:
109			raise SearchError(
110			f"Failed {self.what.key} [{self.what.search_class}] on compound {compound}",
111			compound=compound,
112			search_key=self.what.key,
113			search_class=self.what.search_class,
114			)
115			compounds_run.add(compound)
116			logger.debug(f"Found {len(x)} {self.what.search_name()} annotations for {compound}")
117			n_annot += len(x)
118			n_proc += 1
119			# logging, caching, and such:
120			on_nth = cache.at % SETTINGS.save_every == SETTINGS.save_every - 1
121			is_last = cache.at == len(inchikeys) - 1
122			if on_nth or is_last:
123			logger.log(
124			"NOTICE" if is_last else "INFO",
125			f"Found {len(annotes)} {self.what.search_name()} annotations"
126			+ f" for {cache.at} of {len(inchikeys)} compounds",
127			)
128			self._save(annotes, done=is_last)
129			cache.save(*compounds_run) # CRITICAL -- do this AFTER saving
130			# done!
131			i1, t1 = cache.at, time.monotonic()
			0 ignored issues – show Coding Style Naming introduced 2021-10-28 02:14 UTC by Report Bug Copy Issue Report Variable name "t1" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history... Coding Style Naming introduced 2021-10-28 02:14 UTC by Report Bug Copy Issue Report Variable name "i1" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
132			assert i1 == len(inchikeys)
133			cache.kill()
134			logger.success(f"Wrote {self.what.key} to {self.to}")
135			return SearchReturnInfo(
136			n_kept=n0, n_processed=n_proc, n_errored=n_err, time_taken=timedelta(seconds=t1 - t0)
137			)
138
139			@property
140			def is_partial(self) -> bool:
			0 ignored issues – show introduced 2021-10-28 02:14 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
141			return self.to.exists() and not self.is_complete
142
143			@property
144			def is_complete(self) -> bool:
			0 ignored issues – show introduced 2021-11-04 01:02 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
145			done = self.to in Checksums().load_dirsum_of_file(self.to)
146			if done and not self.to.exists():
147			raise IllegalStateError(f"{self.to} marked complete but does not exist")
148			return done
149
150			def _save(self, hits: Sequence[AbstractHit], *, done: bool) -> None:
151			df = HitDf.from_hits(hits)
			0 ignored issues – show Coding Style Naming introduced 2021-10-12 03:37 UTC by Report Bug Copy Issue Report 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) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
152			# keep all of the original extra columns from the input
153			# e.g. if the user had 'inchi' or 'smiles' or 'pretty_name'
154			# if "origin_inchikey" not in df.columns:
155			for extra_col in [c for c in self.input_df.columns if c != "inchikey"]:
156			extra_mp = self.input_df.set_index("inchikey")[extra_col].to_dict()
157			df[extra_col] = df["origin_inchikey"].map(extra_mp.get)
158			# write the file
159			df: HitDf = HitDf.of(df)
			0 ignored issues – show Coding Style Naming introduced 2021-10-28 02:14 UTC by Report Bug Copy Issue Report 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) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
160			params = self.what.get_params()
161			df = df.set_attrs(**params, key=self.what.key)
			0 ignored issues – show Coding Style Naming introduced 2021-11-04 01:02 UTC by Report Bug Copy Issue Report 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) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
162			df.write_file(self.to.resolve(), mkdirs=True, attrs=True, dir_hash=done)
163			logger.debug(f"Saved {len(df)} rows to {self.to}")
164
165
166			__all__ = ["Searcher", "InputCompoundsDf", "SearchReturnInfo", "MemoizedInputCompounds"]
167

dmyersturnbull / mandos

Push — main ( 2e1b6b...3a0c28 )

MemoizedInputCompounds.read_file() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like