mandos.model.correlation_math

Complexity

Total Complexity

31

Size/Duplication

Total Lines	241
Duplicated Lines	0 %

Importance

Changes

0

17 Methods

Rating	Name	Duplication	Size	Complexity
A	AffinityFunctions.identity()	0	7	1
A	AffinityFunctions.rho_cm()	0	13	1
A	AffinityMatrix.index_names()	0	3	1
A	AffinityMatrix.all_pairs()	0	16	3
A	AffinityMatrix.from_function()	0	21	4
A	AffinityMatrix.restrict()	0	4	1
A	AffinityFunctions.pearson()	0	23	3
A	AffinityFunctions.random_uniform()	0	7	1
A	ConfusionMatrix.cols()	0	4	1
A	AffinityMatrix.cols()	0	4	1
A	AffinityFunctions.jaccard()	0	9	2
A	AffinityMatrix.non_self_pairs()	0	9	1
A	AffinityMatrix.rows()	0	4	1
A	AffinityMatrix.from_triples()	0	7	2
A	AffinityFunctions.bit_string_jaccard()	0	11	2
A	ConfusionMatrix.rows()	0	4	1
A	AffinityFunctions.negative_minkowski()	0	32	5

"""
TODO This module will probably be deleted.
"""

from __future__ import annotations
from collections import defaultdict
import logging
from typing import Sequence, Mapping, Set, TypeVar, Callable, SupportsFloat, Optional, Type, List

Unused Type imported from typing

from typing import Tuple as Tup

import numpy as np

Unable to import 'numpy'

import pandas as pd

Unable to import 'pandas'

from typeddfs import TypedDfs, TypedDf

Unable to import 'typeddfs'

Unused TypedDfs imported from typeddfs

logger = logging.getLogger("mandos")
CompoundCompoundPair = Tup[str, str]


T = TypeVar("T")

Class name "T" doesn't conform to PascalCase naming style ('[^\\W\\da-z][^\\W_]+$' pattern)

Z = TypeVar("Z")

Class name "Z" doesn't conform to PascalCase naming style ('[^\\W\\da-z][^\\W_]+$' pattern)

class ConfusionMatrix(TypedDf):

Missing class docstring

    @property
    def rows(self):

Empty method docstring

        """"""
        return [str(s) for s in self.index.tolist()]

    @property
    def cols(self):

Empty method docstring

        """"""
        return [str(s) for s in self.columns.tolist()]


class AffinityFunctions:
    """
    These are functions that can be passed into ``AffinityMatrix.from_function``.
    """

    @classmethod
    def jaccard(cls) -> Callable[[Set[Z], Set[Z]], float]:

Missing function or method docstring

        def x(a_set: Set[Z], b_set: Set[Z]):

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

            if len(a_set) == len(b_set) == 0:
                return float("NaN")
            return len(a_set.intersection(b_set)) / len(a_set.union(b_set))

        x.__name__ = "jaccard"
        return x

    @classmethod
    def bit_string_jaccard(cls) -> Callable[[str, str], float]:

Missing function or method docstring

        def x(a: str, b: str):

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

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

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

            if len(a) == len(b) == 0:
                return float("NaN")
            on_bits_a = {i for i, v in enumerate(a) if int(v) == 1}
            on_bits_b = {i for i, v in enumerate(b) if int(v) == 1}
            return len(on_bits_a.intersection(on_bits_b)) / len(on_bits_a.union(on_bits_b))

        x.__name__ = "bit_string_jaccard"
        return x

    @classmethod
    def identity(cls) -> Callable[[str, str], float]:

Missing function or method docstring

        def x(a: str, b: str):

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

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

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

            return float(a == b)

        x.__name__ = "identity"
        return x

    @classmethod
    def random_uniform(cls, state: np.random.RandomState) -> Callable[[Z, Z], float]:

Missing function or method docstring

        def x(a: Z, b: Z):

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

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

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

The argument a seems to be unused.

The argument b seems to be unused.

            return state.uniform(0, 1)

        x.__name__ = "random_uniform"
        return x

    @classmethod
    def rho_cm(cls, df: ConfusionMatrix) -> Callable[[str, str], float]:

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)

Missing function or method docstring

        def x(d: ConfusionMatrix, a: str, b: str):

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

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

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

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

            ij = d.at[a, b]

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

            a_off_diag = np.sum(d.at[a, j] for j in d.cols if j != a)
            b_off_diag = np.sum(d.at[i, b] for i in d.rows if i != b)
            return ij / a_off_diag / b_off_diag

        def qq(a: str, b: str):

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

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

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

            return 0.5 * x(df, a, b) + 0.5 * x(ConfusionMatrix(df.transpose()), a, b)

        x.__name__ = "rho_cm"
        return qq

    @classmethod
    def pearson(
        cls, weights: Optional[Sequence[SupportsFloat]] = None

Wrong hanging indentation before block (add 4 spaces).

    ) -> Callable[[Sequence[SupportsFloat], Sequence[SupportsFloat]], float]:
        """
        Pearson correlation coefficient, possibly weighted.

        Args:
            weights:

        Returns:

        """
        weights = 1.0 if weights is None else np.array(weights, dtype=np.float64)

        def x(a: Sequence[SupportsFloat], b: Sequence[SupportsFloat]):

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

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

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

            a, b = np.array(a, dtype=np.float64), np.array(b, dtype=np.float64)
            a = (a - a.mean()) / a.std()
            b = (b - b.mean()) / b.std()
            return np.average(a * b, weights=weights)

        x.__name__ = "pearson" if weights == 1.0 else "weighted-pearson"
        return x

    @classmethod
    def negative_minkowski(
        cls, order: float, weights: Optional[Sequence[SupportsFloat]] = None

Wrong hanging indentation before block (add 4 spaces).

    ) -> Callable[[Sequence[SupportsFloat], Sequence[SupportsFloat]], float]:
        """

        If ``order==0``, this is defined to return the number of elements that differ (number of nonzero).

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

        Args:
            order:
            weights:

        Returns:

        """
        weights = 1.0 if weights is None else np.array(weights, dtype=np.float64)

        def x(a: Sequence[SupportsFloat], b: Sequence[SupportsFloat]):

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

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

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

            a, b = np.array(a, dtype=np.float64), np.array(b, dtype=np.float64)
            if np.isposinf(order):

Unnecessary "elif" after "return"

                return -np.max(np.abs(a - b) * weights)
            elif order == 0:
                return -float(np.count_nonzero(a - b))
            else:
                return -np.float_power(
                    np.sum(np.float_power(np.abs(a - b) * weights, order)), 1 / order
                )

        x.__name__ = (
            "minkowski" + str(order) if weights == 1.0 else "weighted-minkowski" + str(order)
        )
        return x


class AffinityMatrix(TypedDf):
    """
    An affinity matrix of compound by compound.
    It has a single index, which is compound A,
    and the column labels are the compound B.
    """

    @classmethod
    def index_names(cls) -> List[str]:

Missing function or method docstring

        return ["name"]

    # @classmethod
    # def must_be_symmetric(cls) -> bool:
    #    return True

    @property
    def rows(self):

Empty method docstring

        """"""
        return [str(s) for s in self.index.tolist()]

    @property
    def cols(self):

Empty method docstring

        """"""
        return [str(s) for s in self.columns.tolist()]

    def restrict(self, to: Set[str]) -> AffinityMatrix:

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)

Missing function or method docstring

        df = self[self["name"].isin(to)]

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 = df[[v for v in df.columns if v in to]]

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)

        return AffinityMatrix.convert(df)

    def non_self_pairs(self) -> Mapping[CompoundCompoundPair, float]:
        """
        Gets the STRICT lower-triangular matrix, which excludes comparisons where the two labels are the same.

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

        The result is (n choose 2) matrix elements.

        Returns:
            A dict from comparison labels to their values in the matrix (i.e. (i, j)).
        """
        return {(a, b): value for (a, b), value in self.all_pairs().items() if a != b}

    def all_pairs(self) -> Mapping[CompoundCompoundPair, float]:
        """
        Gets the (non-strict) lower-triangular matrix.
        The result is (n(n+1)/2) matrix elements.

        Returns:
            A dict from comparison labels to their values in the matrix (i.e. (i, j)).
        """
        values = {}
        for i in range(0, len(self)):
            a = self.index.values[i]

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

            for j in range(0, i + 1):
                b = self.columns.values[j]

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

                # use (b, a) to get lower triangle
                values[(str(b).strip(), str(a).strip())] = self.iat[i, j]
        return values

    @classmethod
    def from_triples(cls, df: pd.DataFrame) -> AffinityMatrix:

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)

Missing function or method docstring

        # TODO this function doesn't belong here

TODO and FIXME comments should generally be avoided.

        dct = defaultdict(set)
        for row in df.itertuples():
            dct[row.compound_lookup].add(row.object_id + "," + row.predicate)
        return AffinityMatrix.from_function(dct, AffinityFunctions.jaccard())

    @classmethod
    def from_function(cls, items: Mapping[str, T], fn: Callable[[T, T], float]) -> AffinityMatrix:

Argument name "fn" 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

        rows = []
        for a_label, a_value in items.items():
            rows.append(
                pd.Series(
                    {
                        "name": str(a_label),
                        **{
                            str(b_label): fn(a_value, b_value) for b_label, b_value in items.items()
                        },
                    }
                )
            )
        df = pd.DataFrame(rows)  # .astype(np.float64)

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["name"] = df["name"].map(lambda v: str(v).replace(".0", "")).astype(str)
        df = df.set_index("name")

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)

        afm = AffinityMatrix.convert(df)
        for c in afm.columns:

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

            afm[c] = afm[c].astype(np.float64)
        return afm


__all__ = [
    "AffinityMatrix",
    "AffinityFunctions",
]