mandos.entry.multi_searches

Complexity

Total Complexity

28

Size/Duplication

Total Lines	188
Duplicated Lines	0 %

Importance

Changes

0

14 Methods

Rating	Name	Duplication	Size	Complexity
A	CmdRunner.key()	0	3	1
A	MultiSearch.run()	0	18	3
A	MultiSearch.final_path()	0	7	2
A	MultiSearch.build()	0	6	1
A	CmdRunner.done_path()	0	3	1
A	CmdRunner.run()	0	2	1
A	CmdRunner.build()	0	29	3
A	MultiSearch.__post_init__()	0	7	5
A	MultiSearch._build_commands()	0	15	5
A	CmdRunner.output_path()	0	3	1
A	CmdRunner.test()	0	2	1
A	CmdRunner.was_run()	0	3	1
A	MultiSearch.explain_path()	0	3	1
A	MultiSearch.to_table()	0	11	2

"""
Runner.
"""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Sequence, Type, Union, Optional, MutableMapping

Unused Dict imported from typing

import pandas as pd

Unable to import 'pandas'

import typer

Unable to import 'typer'

from typeddfs import TypedDfs

Unable to import 'typeddfs'

from pocketutils.core.dot_dict import NestedDotDict

Unable to import 'pocketutils.core.dot_dict'

from mandos import logger
from mandos.entry.api_singletons import Apis
from mandos.entry.entry_commands import Entries
from mandos.entry.abstract_entries import Entry
from mandos.model.utils.reflection_utils import InjectionError
from mandos.model.hits import HitFrame
from mandos.model.settings import MANDOS_SETTINGS

cli = typer.Typer()
Apis.set_default()
Chembl, Pubchem = Apis.Chembl, Apis.Pubchem

EntriesByCmd: MutableMapping[str, Type[Entry]] = {e.cmd(): e for e in Entries}

# these are only permitted in 'meta', not individual searches
meta_keys = {"verbose", "quiet", "check", "log", "to"}
forbidden_keys = {"dir", "out-dir", "out_dir"}

SearchExplainDf = (
    TypedDfs.typed("SearchExplainDf")
    .require("key", "search", "source", dtype=str)
    .require("category", "desc", "args", dtype=str)
    .strict()
    .secure()
).build()


@dataclass(frozen=True, repr=True)

Missing class docstring

class MultiSearch:
    # 'meta' allows us to set defaults for things like --to
    meta: NestedDotDict
    searches: Sequence[NestedDotDict]
    toml_path: Path
    input_path: Path
    out_dir: Path

    @property
    def final_path(self) -> Path:

Missing function or method docstring

        if "to" in self.meta:
            fmt = Path(self.meta["to"]).suffix
        else:
            fmt = MANDOS_SETTINGS.default_table_suffix
        return self.out_dir / ("search_" + self.input_path.name + "_" + self.toml_path.name + fmt)

    @property
    def explain_path(self) -> Path:

Missing function or method docstring

        return Path(str(self.final_path.with_suffix("")) + "_explain.tsv")

    def __post_init__(self):
        to = self.meta.get_as("to", str)

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

        if to is not None and not to.startswith("."):
            raise ValueError(f"Argument 'to' ({to})' must start with '.'.")
        for key, value in self.meta.items():

The variable value seems to be unused.

            if key not in meta_keys:
                raise ValueError(f"{key} in 'meta' not supported.")

    @classmethod
    def build(cls, input_path: Path, out_dir: Path, toml_path: Path) -> MultiSearch:

Missing function or method docstring

        toml = NestedDotDict.read_toml(toml_path)
        searches = toml.get_as("search", list, [])
        meta = toml.sub("meta")
        return MultiSearch(meta, searches, toml_path, input_path, out_dir)

    def to_table(self) -> SearchExplainDf:

Missing function or method docstring

        rows = []
        for cmd in self._build_commands():
            name = cmd.cmd.get_search_type().search_name
            cat = cmd.category
            src = cmd.cmd.get_search_type()(cmd.key).data_source
            desc = cmd.cmd.describe()
            args = ", ".join([f"{k}={v}" for k, v in cmd.params.items()])
            ser = dict(key=cmd.key, search=name, category=cat, source=src, desc=desc, args=args)
            rows.append(pd.Series(ser))
        return SearchExplainDf(rows)

    def run(self) -> None:

Missing function or method docstring

        # build up the list of Entry classes first, and run ``test`` on each one
        # that's to check that the parameters are correct before running anything
        commands = self._build_commands()
        # write a metadata file describing all of the searches
        explain = self.to_table()
        explain.write_file(self.explain_path)
        for cmd in commands:
            cmd.test()
            logger.info(f"Search {cmd.key} looks ok.")
        logger.notice("All searches look ok.")
        for cmd in commands:
            cmd.run()
        logger.notice("Done with all searches!")
        # write the final file
        df = HitFrame(pd.concat([HitFrame.read_file(cmd.output_path) for cmd in commands]))

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.write_file(self.final_path)
        logger.notice(f"Concatenated file to {self.final_path}")

    def _build_commands(self) -> Sequence[CmdRunner]:
        commands = {}
        skipping = []
        for search in self.searches:
            cmd = CmdRunner.build(search, self.meta, self.input_path, self.out_dir)
            if cmd.was_run:
                skipping += [cmd]
            else:
                commands[cmd.key] = cmd
                if cmd.key in commands:
                    raise ValueError(f"Repeated search key '{cmd.key}'")
        if len(skipping) > 0:
            skipping = ", ".join([c.key for c in skipping])
            logger.notice(f"Skipping searches {skipping} (already run).")
        return list(commands.values())


@dataclass(frozen=True, repr=True)

Missing class docstring

class CmdRunner:
    cmd: Type[Entry]
    params: MutableMapping[str, Union[int, str, float]]
    input_path: Path
    category: Optional[str]

    @property
    def key(self) -> str:

Missing function or method docstring

        return self.params["key"]

    @property
    def output_path(self) -> Path:

Missing function or method docstring

        return Path(self.params["to"])

    @property
    def was_run(self) -> bool:

Missing function or method docstring

        return self.done_path.exists()

    @property
    def done_path(self) -> Path:

Missing function or method docstring

        return self.output_path.with_suffix(self.output_path.suffix + ".done")

    def test(self) -> None:

Missing function or method docstring

        self.cmd.test(self.input_path, **{**self.params, **dict(quiet=True)})

    def run(self) -> None:

Missing function or method docstring

        self.cmd.run(self.input_path, **{**self.params, **dict(no_setup=True, check=False)})

    @classmethod
    def build(cls, e: NestedDotDict, meta: NestedDotDict, input_path: Path, out_dir: Path):

Missing function or method docstring

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

        cmd = e.req_as("source", str)
        key = e.get_as("key", str, cmd)
        to = meta.get_as("to", str, MANDOS_SETTINGS.default_table_suffix)

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

        try:
            cmd = EntriesByCmd[cmd]
        except KeyError:
            raise InjectionError(f"Search command {cmd} (key {key}) does not exist")
        # use defaults
        params = dict(meta)
        # they shouldn't pass any of these args
        bad = {b for b in {*meta_keys, "path", "no_setup", "out_dir"} if b in e}
        if len(bad) > 0:
            raise ValueError(f"Forbidden keys in [[search]] ({cmd}): {','.join(bad)}")
        # update the defaults from 'meta' (e.g. 'verbose')
        # skip the source -- it's the command name
        # stupidly, we need to explicitly add the defaults from the OptionInfo instances
        params.update(cmd.default_param_values().items())
        # do this after: the defaults had path, key, and to
        params["key"] = key
        params["path"] = e.req_as("path", Path)
        params["out_dir"] = out_dir
        params.setdefault("to", to)
        # now add the params we got for this command's section
        params.update({k: v for k, v in e.items() if k != "source" and k != "category"})

Consider merging these comparisons with "in" to "k not in ('source', 'category')"

        del params["check"]
        category = e.get_as("category", str)
        return CmdRunner(cmd, params, input_path, category)


__all__ = ["MultiSearch"]