mandos.analysis.distances

Complexity

Total Complexity

39

Size/Duplication

Total Lines	242
Duplicated Lines	0 %

Importance

Changes

0

18 Methods

Rating	Name	Duplication	Size	Complexity
A	JPrimeMatrixCalculator._jx()	0	11	2
A	_Inf.__repr__()	0	2	1
A	MatrixAlg.clazz()	0	3	1
A	_Inf.log()	0	6	1
A	JPrimeMatrixCalculator._path_of()	0	2	1
A	JPrimeMatrixCalculator._vee()	0	2	1
A	JPrimeMatrixCalculator.calc_one()	0	14	5
A	_Inf.got()	0	5	2
A	MatrixCalculator.calc_all()	0	2	1
A	MatrixCalculator.__init__()	0	4	1
A	_Inf.i()	0	3	1
A	_Inf.__str__()	0	2	1
A	MatrixCalculation.create()	0	11	1
A	_Inf.is_used()	0	2	1
A	JPrimeMatrixCalculator._j_prime()	0	17	4
A	_Inf.__init__()	0	3	1
A	JPrimeMatrixCalculator._wedge()	0	2	1
F	JPrimeMatrixCalculator.calc_all()	0	83	13

"""
Calculations of overlap (similarity) between annotation sets.
"""
import abc
import enum
import math
import time
from collections import defaultdict
from pathlib import Path
from typing import Collection, Sequence, Type, Union

import decorateme

Unable to import 'decorateme'

import numpy as np

Unable to import 'numpy'

import pandas as pd

Unable to import 'pandas'

from pocketutils.core.chars import Chars

Unable to import 'pocketutils.core.chars'

from pocketutils.core.enums import CleverEnum

Unable to import 'pocketutils.core.enums'

from pocketutils.tools.unit_tools import UnitTools

Unable to import 'pocketutils.tools.unit_tools'

from typeddfs.df_errors import HashFileMissingError

Unable to import 'typeddfs.df_errors'

from mandos.analysis import AnalysisUtils as Au
from mandos.analysis.io_defns import SimilarityDfLongForm, SimilarityDfShortForm
from mandos.model.hit_dfs import HitDf
from mandos.model.hits import AbstractHit
from mandos.model.utils import unlink

# note that most of these math functions are much faster than their numpy counterparts
# if we're not broadcasting, it's almost always better to use them
# some are more accurate, too
# e.g. we're using fsum rather than sum
from mandos.model.utils.setup import logger


@decorateme.auto_repr_str()

Missing class docstring

class MatrixCalculator(metaclass=abc.ABCMeta):
    def __init__(self, *, min_compounds: int, min_nonzero: int, min_hits: int):
        self.min_compounds = min_compounds
        self.min_nonzero = min_nonzero
        self.min_hits = min_hits

    def calc_all(self, hits: Path, to: Path, *, keep_temp: bool = False) -> SimilarityDfLongForm:

Missing function or method docstring

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

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

        raise NotImplemented()

NotImplemented does not seem to be callable.

NotImplemented raised - should raise NotImplementedError

class _Inf:
    def __init__(self, n: int):
        self.n = n

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

        self.used, self.t0, self.nonzeros = set(), time.monotonic(), 0

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

    def is_used(self, c1: str, c2: str) -> bool:

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

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

Missing function or method docstring

        return (c1, c2) in self.used or (c2, c1) in self.used

    def got(self, c1: str, c2: str, z: float) -> None:

Missing function or method docstring

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

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

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

        self.used.add((c1, c2))
        self.nonzeros += int(c1 != c2 and not np.isnan(z) and 0 < z < 1)
        if self.i % 20000 == 0:
            self.log("info")

    @property
    def i(self) -> int:

Missing function or method docstring

        return len(self.used)

    def log(self, level: str) -> None:

Missing function or method docstring

        delta = UnitTools.delta_time_to_str(time.monotonic() - self.t0, space=Chars.narrownbsp)
        logger.log(
            level.upper(),
            f"Processed {self.i:,}/{self.n:,} pairs in {delta};"
            + f" {self.nonzeros:,} ({self.nonzeros / self.i * 100:.1f}%) are nonzero",
        )

    def __repr__(self):
        return f"{self.__class__.__name__}({self.i}/{self.n})"

    def __str__(self):
        return repr(self)


class JPrimeMatrixCalculator(MatrixCalculator):

Missing class docstring

    def calc_all(self, path: Path, to: Path, *, keep_temp: bool = False) -> SimilarityDfLongForm:

This function exceeds the maximum number of variables (23/15).

Parameters differ from overridden 'calc_all' method

        hits = HitDf.read_file(path).to_hits()
        key_to_hit = Au.hit_multidict(hits, "search_key")
        logger.notice(f"Calculating J on {len(key_to_hit):,} keys from {len(hits):,} hits")
        deltas, files, good_keys = [], [], {}
        for key, key_hits in key_to_hit.items():
            key: str = key
            key_hits: Sequence[AbstractHit] = key_hits
            n_compounds_0 = len({k.origin_inchikey for k in key_hits})
            part_path = self._path_of(path, key)
            n_compounds_in_mx = None
            n_nonzero = None
            df = None

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)

            if part_path.exists():
                try:
                    df = SimilarityDfLongForm.read_file(

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)

                        part_path, file_hash=False
                    )  # TODO: file_hash=True

TODO and FIXME comments should generally be avoided.

                    logger.warning(f"Results for key {key} already exist ({len(df):,} rows)")
                    n_compounds_in_mx = len(df["inchikey_1"].unique())
                except HashFileMissingError:
                    logger.error(f"Extant results for key {key} appear incomplete; restarting")
                    logger.opt(exception=True).debug(f"Hash error for {key}")
                    unlink(part_path)
                    # now let it go into the next block -- calculate from scratch
            if n_compounds_0 >= self.min_compounds:
                t1 = time.monotonic()

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)

                df: SimilarityDfShortForm = self.calc_one(key, key_hits)

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)

                t2 = time.monotonic()

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

                deltas.append(t2 - t1)
                df = df.to_long_form(kind="psi", key=key)

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)

                n_compounds_in_mx = len(df["inchikey_1"].unique())
                df.write_file(part_path)
                logger.debug(f"Wrote results for {key} to {part_path}")
            if df is not None:
                n_nonzero = len(df[df["value"] > 0])
            if n_compounds_in_mx < self.min_compounds:
                logger.warning(
                    f"Key {key} has {n_compounds_in_mx:,} < {self.min_compounds:,} compounds; skipping"

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

                )
            elif len(key_hits) < self.min_hits:
                logger.warning(
                    f"Key {key} has {len(key_hits):,} < {self.min_hits:,} hits; skipping"
                )
            elif n_nonzero is not None and n_nonzero < self.min_nonzero:
                logger.warning(
                    f"Key {key} has {n_nonzero:,} < {self.min_nonzero:,} nonzero pairs; skipping"
                )  # TODO: percent nonzero?

TODO and FIXME comments should generally be avoided.

            else:
                files.append(part_path)
                good_keys[key] = n_compounds_in_mx
            del df
        logger.debug(f"Concatenating {len(files):,} files")
        df = SimilarityDfLongForm(

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.concat(
                [SimilarityDfLongForm.read_file(self._path_of(path, k)) for k in good_keys.keys()]

Consider iterating the dictionary directly instead of calling .keys()

            )
        )
        logger.notice(f"Included {len(good_keys):,} keys: {', '.join(good_keys.keys())}")
        quartiles = {}
        for k, v in good_keys.items():

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

            vals = df[df["key"] == k]["value"]
            qs = {x: vals.quantile(x) for x in [0, 0.25, 0.5, 0.75, 1]}

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

            quartiles[k] = list(qs.values())
            logger.info(f"Key {k} has {v:,} compounds and {len(key_to_hit[k]):,} hits")
            logger.info(
                f"    {k} {Chars.fatright} unique values = {len(vals.unique())} unique values"
            )
            logger.info(f"    {k} {Chars.fatright} quartiles: " + " | ".join(qs.values()))
        df = df.set_attrs(

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)

            dict(
                keys={
                    k: dict(compounds=v, hits=len(key_to_hit[k]), quartiles=quartiles[k])
                    for k, v in good_keys.items()
                }
            )
        )
        df.write_file(to, attrs=True, file_hash=True)
        logger.notice(f"Wrote {len(df):,} rows to {to}")
        if not keep_temp:
            for k in key_to_hit.keys():
                unlink(self._path_of(path, k))
        return df

    def calc_one(self, key: str, hits: Sequence[AbstractHit]) -> SimilarityDfShortForm:

Missing function or method docstring

        ik2hits = Au.hit_multidict(hits, "origin_inchikey")
        logger.info(f"Calculating J on {key} for {len(ik2hits):,} compounds and {len(hits):,} hits")
        data = defaultdict(dict)
        inf = _Inf(n=int(len(ik2hits) * (len(ik2hits) - 1) / 2))
        for (c1, hits1) in ik2hits.items():

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

            for (c2, hits2) in ik2hits.items():

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

                if inf.is_used(c1, c2):
                    continue
                z = 1 if c1 == c2 else self._j_prime(key, hits1, hits2)

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)

                data[c1][c2] = z
                inf.got(c1, c2, z)
        inf.log("success")
        return SimilarityDfShortForm.from_dict(data)

    def _path_of(self, path: Path, key: str):

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

        return path.parent / f".{path.name}-{key}.tmp.feather"

    def _j_prime(
        self, key: str, hits1: Collection[AbstractHit], hits2: Collection[AbstractHit]

Wrong hanging indentation before block (add 4 spaces).

    ) -> float:
        if len(hits1) == 0 or len(hits2) == 0:
            return 0
        sources = {h.data_source for h in hits1}.intersection({h.data_source for h in hits2})
        if len(sources) == 0:
            return float("NaN")
        values = [
            self._jx(
                key,
                [h for h in hits1 if h.data_source == source],
                [h for h in hits2 if h.data_source == source],
            )
            for source in sources
        ]
        return float(math.fsum(values) / len(values))

    def _jx(
        self, key: str, hits1: Collection[AbstractHit], hits2: Collection[AbstractHit]

Wrong hanging indentation before block (add 4 spaces).

    ) -> float:
        # TODO -- for testing only

TODO and FIXME comments should generally be avoided.

        # TODO: REMOVE ME!

TODO and FIXME comments should generally be avoided.

        if key in ["core.chemidplus.effects", "extra.chemidplus.specific-effects"]:
            hits1 = [h.copy(weight=math.pow(10, -h.weight)) for h in hits1]
            hits2 = [h.copy(weight=math.pow(10, -h.weight)) for h in hits2]
        pair_to_weights = Au.weights_of_pairs(hits1, hits2)
        values = [self._wedge(ca, cb) / self._vee(ca, cb) for ca, cb in pair_to_weights.values()]
        return float(math.fsum(values) / len(values))

    def _wedge(self, ca: float, cb: float) -> float:

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

Argument name "cb" 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.

        return math.sqrt(Au.elle(ca) * Au.elle(cb))

    def _vee(self, ca: float, cb: float) -> float:

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

Argument name "cb" 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.

        return Au.elle(ca) + Au.elle(cb) - math.sqrt(Au.elle(ca) * Au.elle(cb))


class MatrixAlg(CleverEnum):

Missing class docstring

    j = enum.auto()

    @property
    def clazz(self) -> Type[MatrixCalculator]:

Missing function or method docstring

        return {MatrixAlg.j: JPrimeMatrixCalculator}[self]


@decorateme.auto_utils()

Missing class docstring

class MatrixCalculation:
    @classmethod
    def create(

Missing function or method docstring

        cls,

Wrong hanging indentation before block (add 4 spaces).

        algorithm: Union[str, MatrixAlg],

Wrong hanging indentation before block (add 4 spaces).

        *,

Wrong hanging indentation before block (add 4 spaces).

        min_compounds: int,

Wrong hanging indentation before block (add 4 spaces).

        min_nonzero: int,

Wrong hanging indentation before block (add 4 spaces).

        min_hits: int,

Wrong hanging indentation before block (add 4 spaces).

    ) -> MatrixCalculator:
        return MatrixAlg.of(algorithm).clazz(
            min_compounds=min_compounds, min_nonzero=min_nonzero, min_hits=min_hits
        )


__all__ = ["MatrixCalculator", "JPrimeMatrixCalculator", "MatrixCalculation", "MatrixAlg"]