mandos.model.taxonomy_caches

Complexity

Total Complexity

42

Size/Duplication

Total Lines	238
Duplicated Lines	0 %

Importance

Changes

0

21 Methods

Rating	Name	Duplication	Size	Complexity
A	TaxonomyFactory.load()	0	2	1
A	FixedTaxonomyFactory.__init__()	0	2	1
A	CachedTaxonomyCache._check_has()	0	8	2
A	CachedTaxonomyCache._determine_name()	0	6	2
A	TaxonomyFactories.main()	0	7	1
A	CachedTaxonomyCache._resolve_non_vertebrate_final()	0	2	1
A	CachedTaxonomyCache._get_resource()	0	6	2
A	CachedTaxonomyCache.load()	0	19	3
A	CachedTaxonomyCache._download_raw()	0	7	3
A	CachedTaxonomyCache.resolve_path()	0	2	1
A	CachedTaxonomyCache._resolve_non_vertebrate_raw()	0	4	1
A	CachedTaxonomyCache.load_exact()	0	5	4
A	TaxonomyFactories.get_smart_taxonomy()	0	24	3
A	TaxonomyFactories.list_cached_files()	0	7	1
A	CachedTaxonomyCache.load_vertebrate()	0	4	2
A	CachedTaxonomyCache.__init__()	0	3	1
A	CachedTaxonomyCache.rebuild()	0	9	5
A	FixedTaxonomyFactory.load()	0	4	2
A	CachedTaxonomyCache._fix()	0	23	1
A	CachedTaxonomyCache.delete_exact()	0	10	2
A	CachedTaxonomyCache._load_or_dl()	0	12	3

"""
Caching.
"""

from __future__ import annotations

import abc
import shutil
from pathlib import Path
from typing import AbstractSet, Collection, Iterable, Mapping, Optional, Set, Union

Unused AbstractSet imported from typing

import pandas as pd

Unable to import 'pandas'

import requests

Unable to import 'requests'

from pocketutils.core.exceptions import LookupFailedError, XValueError

Unable to import 'pocketutils.core.exceptions'

Unused LookupFailedError imported from pocketutils.core.exceptions

from typeddfs import TypedDfs

Unable to import 'typeddfs'

from typeddfs.checksums import Checksums

Unable to import 'typeddfs.checksums'

from mandos.model.settings import SETTINGS, Globals
from mandos.model.taxonomy import Taxonomy, TaxonomyDf
from mandos.model.utils.resources import MandosResources
from mandos.model.utils.setup import logger


class TaxonomyFactory(metaclass=abc.ABCMeta):

Missing class docstring

    def load(self, taxon: Union[int, str]) -> Taxonomy:

Missing function or method docstring

        raise NotImplementedError()


class CachedTaxonomyCache(TaxonomyFactory, metaclass=abc.ABCMeta):
    """
    Preps a new taxonomy file for use in mandos.
    Just returns if a corresponding file already exists in the resources dir or mandos cache (``~/.mandos``).

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

    Otherwise, downloads a tab-separated file from UniProt.
    (To find manually, follow the ``All lower taxonomy nodes`` link and click ``Download``.)
    Then applies fixes and reduces the file size, creating a new file alongside.
    Puts both the raw data and fixed data in the cache under ``~/.mandos/taxonomy/``.
    """

    def __init__(self, *, cache_dir: Path = SETTINGS.taxonomy_cache_path, local_only: bool):
        self.cache_dir = cache_dir
        self.local_only = local_only

    def load(self, taxon: Union[int, str]) -> Taxonomy:
        """
        Tries, in order:

            1. A cached file exactly matching the taxon ID
            2. A taxon ID under vertebrata
            3. The UNIQUE name of a taxon under vertebrata
            4. Downloads the taxonomy with the specified ID
        """
        tree = self.load_exact(taxon)
        if tree is None:
            vert = self.load_vertebrate(Globals.vertebrata)
            if taxon in vert:
                tree = vert.subtrees_by_ids_or_names(taxon)
            else:
                logger.info(f"Taxon {taxon} found in the vertebrata cache")
                tree = self._load_or_dl(taxon)
        logger.info(f"Taxonomy has {len(tree)} taxa with {len(tree.roots)} roots")
        return tree

    def load_exact(self, taxon: int) -> Optional[Taxonomy]:

Missing function or method docstring

        path = self._resolve_non_vertebrate_final(taxon)
        if (self._check_has(taxon, path) or self.local_only) and path.exists():
            return Taxonomy.from_path(path)
        return None

    def load_vertebrate(self, taxon: Union[int, str]) -> Optional[Taxonomy]:

Missing function or method docstring

        vertebrata = self._load_or_dl(Globals.vertebrata)
        vertebrate = vertebrata.subtrees_by_ids_or_names([taxon])
        return vertebrate if vertebrate.n_taxa() > 0 else None

    def _check_has(self, taxon: Union[str, int], path: Path) -> bool:

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

        if path.exists():
            return not MandosResources.check_expired(
                path,
                max_sec=SETTINGS.taxon_expire_sec,
                what=f"Cached taxa under {taxon}",
            )
        return False

    def _load_or_dl(self, taxon: Union[int, str]) -> Taxonomy:
        path = self._resolve_non_vertebrate_final(taxon)
        raw_path = self._resolve_non_vertebrate_raw(taxon)
        if self._check_has(taxon, path) or self.local_only:

Unnecessary "else" after "return"

            return Taxonomy.from_path(path)
        else:
            logger.notice(f"Downloading new taxonomy file for taxon {taxon}")
            self._download_raw(raw_path, taxon)
            path = self._resolve_non_vertebrate_final(taxon)
            df = self._fix(raw_path, taxon, path)

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)

            logger.notice(f"Cached taxonomy at {path} .")
            return df

    def rebuild(self, *taxa: int, replace: bool) -> None:

Missing function or method docstring

        if self.local_only:
            logger.error(f"Cannot rebuild -- local_only is set")

Using an f-string that does not have any interpolated variables

        for taxon in taxa:
            path = self.resolve_path(taxon)
            if replace or not path.exists():
                self.delete_exact(taxon)
                self._load_or_dl(taxon)
                logger.notice(f"Regenerated {taxon} taxonomy")

    def delete_exact(self, taxon: int) -> None:

Missing function or method docstring

        raw = self._resolve_non_vertebrate_raw(taxon)
        raw.unlink(missing_ok=True)

The keyword missing_ok does not seem to exist for the method call.

        p = self._resolve_non_vertebrate_raw(taxon)

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

        if p.exists():
            p.unlink()
            logger.warning(f"Deleted cached taxonomy file {p}")
        # delete either way:
        checksum_file = Checksums.get_hash_file(p, algorithm=SETTINGS.checksum_algorithm)

The Instance of Settings does not seem to have a member named checksum_algorithm.

        checksum_file.unlink(missing_ok=True)

    def resolve_path(self, taxon: int) -> Path:

Missing function or method docstring

        return self._resolve_non_vertebrate_final(taxon)

    def _resolve_non_vertebrate_final(self, taxon: int) -> Path:
        return self._get_resource(f"{taxon}{SETTINGS.archive_filename_suffix}")

    def _resolve_non_vertebrate_raw(self, taxon: int) -> Path:
        # this is what is downloaded from PubChem
        # the filename is the same
        return self._get_resource(f"taxonomy-ancestor_{taxon}.tsv.gz")

    def _get_resource(self, *nodes: Union[Path, str]) -> Path:
        path = MandosResources.path(*nodes)
        if path.exists():
            return path
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        return Path(self.cache_dir, *nodes)

    def _download_raw(self, raw_path: Path, taxon: int) -> None:

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

        # this is faster and safer than using pd.read_csv(url)
        # https://uniprot.org/taxonomy/?query=ancestor:7742&format=tab&force=true&columns=id&compress=yes
        url = f"https://uniprot.org/taxonomy/?query=ancestor:{taxon}&format=tab&force=true&columns=id&compress=yes"

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

        with requests.get(url, stream=True) as r:

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

            with raw_path.open("wb") as f:

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

                shutil.copyfileobj(r.raw, f)

    def _fix(self, raw_path: Path, taxon: int, final_path: Path) -> TaxonomyDf:
        # now process it!
        # unfortunately it won't include an entry for the root ancestor (`taxon`)
        # so, we'll add it in (in ``df.append`` below)
        # noinspection PyPep8Naming
        raw_type = TypedDfs.untyped("Raw")
        df = raw_type.read_file(raw_path)

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)

        # find the scientific name of the parent
        scientific_name = self._determine_name(df, taxon)
        # now fix the columns
        df = df[["Taxon", "Mnemonic", "Scientific name", "Common name", "Parent"]]

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.columns = ["taxon", "mnemonic", "scientific_name", "common_name", "parent"]
        # now add the ancestor back in
        df = df.append(

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)

            pd.Series(dict(taxon=taxon, scientific_name=scientific_name, parent=0)),
            ignore_index=True,
        )
        df["parent"] = df["parent"].fillna(0).astype(int)
        # write it to a feather / csv / whatever
        df = TaxonomyDf.convert(df)

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(final_path, dir_hash=True)
        raw_path.unlink()
        return df

    def _determine_name(self, df: pd.DataFrame, taxon: int) -> str:

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 method could be written as a function/class method.

        got = df[df["Parent"] == taxon]
        if len(got) == 0:
            raise XValueError(f"Could not infer scientific name for {taxon}")
        z = str(list(got["Lineage"])[0])

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

        return z.split("; ")[-1].strip()


class FixedTaxonomyFactory(TaxonomyFactory):
    """
    Mostly for testing.
    """

    def __init__(self, tax: Taxonomy):
        self._tax = tax

    def load(self, taxon: Union[int, str]) -> Taxonomy:
        if isinstance(taxon, str):
            taxon = self._tax.req_only_by_name(taxon).id
        return self._tax.subtree(taxon)


class TaxonomyFactories:
    """
    Collection of static factory methods.
    """

    @classmethod
    def list_cached_files(cls) -> Mapping[int, Path]:

Missing function or method docstring

        suffix = SETTINGS.archive_filename_suffix
        return {
            int(p.scientific_name.replace(suffix, "")): p
            for p in SETTINGS.taxonomy_cache_path.iterdir()
            if p.suffix.endswith(suffix)
        }

    @classmethod
    def main(

Missing function or method docstring

        cls,

Wrong hanging indentation before block (add 4 spaces).

        cache_dir: Path = SETTINGS.taxonomy_cache_path,

Wrong hanging indentation before block (add 4 spaces).

        local_only: bool = False,

Wrong hanging indentation before block (add 4 spaces).

    ):
        return CachedTaxonomyCache(local_only=local_only, cache_dir=cache_dir)

    @classmethod
    def get_smart_taxonomy(

Missing function or method docstring

        cls,

Wrong hanging indentation before block (add 4 spaces).

        *,

Wrong hanging indentation before block (add 4 spaces).

        allow: Iterable[Union[int, str]],

Wrong hanging indentation before block (add 4 spaces).

        forbid: Iterable[Union[int, str]],

Wrong hanging indentation before block (add 4 spaces).

        ancestors: Union[int, Collection[int]] = Globals.cellular_taxon,

Wrong hanging indentation before block (add 4 spaces).

        cache_dir: Path = SETTINGS.taxonomy_cache_path,

Wrong hanging indentation before block (add 4 spaces).

        local_only: bool,

Wrong hanging indentation before block (add 4 spaces).

    ) -> Taxonomy:
        cache = CachedTaxonomyCache(local_only=local_only, cache_dir=cache_dir)
        vertebrata = cache.load_vertebrate(Globals.vertebrata)
        return vertebrata.subtrees_by_ids_or_names(allow).exclude_subtrees_by_ids_or_names(forbid)
        # TODO:

TODO and FIXME comments should generally be avoided.

        vertebrates: Set[Union[int, str]] = {t for t in allow if t in vertebrata}

This code does not seem to be reachable.

        invertebrates: Set[Union[int, str]] = {t for t in allow if t not in vertebrata}
        trees: Set[Taxonomy] = {cache.load(t) for t in vertebrates}
        if len(invertebrates) > 0:
            if len(ancestors) == 0:
                new = {cache.load(t) for t in invertebrates}
            else:
                new = Taxonomy.from_trees({cache.load(t) for t in ancestors})
            trees.add(new.subtrees_by_ids_or_names(invertebrates))
        return Taxonomy.from_trees(trees).exclude_subtrees_by_ids_or_names(forbid)


__all__ = ["TaxonomyFactory", "TaxonomyFactories"]