mandos.analysis.regression

Complexity

Total Complexity

21

Size/Duplication

Total Lines	197
Duplicated Lines	0 %

Importance

Changes

0

13 Methods

Rating	Name	Duplication	Size	Complexity
A	EnrichmentCalculator.__init__()	0	2	1
A	EnrichmentAlg.clazz()	0	9	1
A	AlphaCalculator._calc_term()	0	6	1
A	FoldUnweightedCalc.for_pair()	0	6	2
A	EnrichmentAlg.description()	0	10	1
A	EnrichmentCalculator.calc()	0	6	2
A	EnrichmentAlg.symbol()	0	9	1
A	FoldWeightedCalc.for_pair()	0	8	2
A	SumUnweightedCalc.for_pair()	0	2	1
A	SumWeightedCalc.for_pair()	0	2	1
A	AlphaCalculator.for_pair()	0	6	2
A	EnrichmentCalculation.calc()	0	24	2
A	EnrichmentCalculator.for_pair()	0	2	1

"""
Scoring (regression and enrichment) calculations.
"""
import abc
import enum
import math
from typing import Generic, Mapping, Sequence, Type, TypeVar, Union

import pandas as pd

Unable to import 'pandas'

from typeddfs import TypedDfs

Unable to import 'typeddfs'

from mandos.analysis import AnalysisUtils as Au
from mandos.model import CleverEnum
from mandos.model.hits import AbstractHit, HitFrame, Pair

S = TypeVar("S", bound=Union[int, float, bool])

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

EnrichmentDf = (
    TypedDfs.typed("EnrichmentDf")
    .require("predicate", "object", dtype=str)
    .require("samples", dtype=int)
).build()
# extra cols are, e.g., alpha(score_1), alpha(is_hit)


ScoreDf = (
    TypedDfs.typed("ScoreDf")
    .require("inchikey", dtype=str, index=True)
    .reserve("score")
    .reserve("is_hit", dtype=bool)
).build()
# extra cols are, e.g., score_1, score_hello, is_lethal, is_good


def _score_cols(self: ScoreDf) -> Sequence[str]:
    return [
        col
        for col in self.columns
        if col == "score" or col.startswith("score_") or col.startswith("is_")
    ]


def _all_scores(self: ScoreDf) -> Mapping[str, Mapping[str, S]]:
    results = {}
    for c in self.score_cols:

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)

        results[c] = self[c].to_dict()
    return results


ScoreDf.score_cols = _score_cols
ScoreDf.all_scores = _all_scores


class EnrichmentCalculator(Generic[S], metaclass=abc.ABCMeta):

Missing class docstring

    def __init__(self, scores: Mapping[str, S]):
        self.scores = scores

    def calc(self, hits: Sequence[AbstractHit]) -> Mapping[Pair, float]:

Missing function or method docstring

        pair_to_hits = Au.hit_multidict(hits, "to_pair")
        results = {}
        for pair, the_hits in pair_to_hits.items():

The variable the_hits seems to be unused.

            results[pair] = self.for_pair(hits)
        return results

    def for_pair(self, hits: Sequence[AbstractHit]) -> float:

Missing function or method docstring

        raise NotImplementedError()


class AlphaCalculator(EnrichmentCalculator[S]):

Value 'EnrichmentCalculator' is unsubscriptable

Missing class docstring

    def for_pair(self, hits: Sequence[AbstractHit]) -> float:

Missing function or method docstring

        source_to_hits = Au.hit_multidict(hits, "data_source")
        terms = []
        for source, source_hits in source_to_hits.items():

The variable source seems to be unused.

            terms.append(self._calc_term(source_hits))
        return math.fsum(terms) / len(terms)

    def _calc_term(self, hits: Sequence[AbstractHit]) -> float:
        values = [
            Au.elle(hit.value) * (2 * float(self.scores[hit.origin_inchikey] - 1)) ** 2
            for hit in hits
        ]
        return math.fsum(values) / len(values)


class FoldUnweightedCalc(EnrichmentCalculator[bool]):

Value 'EnrichmentCalculator' is unsubscriptable

Missing class docstring

    def for_pair(self, hits: Sequence[AbstractHit]) -> float:

Missing function or method docstring

        numerator = len([hit for hit in hits if self.scores[hit.origin_inchikey]])
        denominator = len([hit for hit in hits if not self.scores[hit.origin_inchikey]])
        if denominator == 0:
            return float("inf")
        return numerator / denominator


class FoldWeightedCalc(EnrichmentCalculator[bool]):

Missing class docstring

Value 'EnrichmentCalculator' is unsubscriptable

    def for_pair(self, hits: Sequence[AbstractHit]) -> float:

Missing function or method docstring

        yes = [hit for hit in hits if self.scores[hit.origin_inchikey]]
        no = [hit for hit in hits if not self.scores[hit.origin_inchikey]]

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

        numerator = math.fsum((hit.value for hit in yes))

The variable hit does not seem to be defined.

        denominator = math.fsum((hit.value for hit in no))
        if denominator == 0:
            return float("inf")
        return numerator / denominator


class SumWeightedCalc(EnrichmentCalculator[S]):

Missing class docstring

Value 'EnrichmentCalculator' is unsubscriptable

    def for_pair(self, hits: Sequence[AbstractHit]) -> float:

Missing function or method docstring

        return math.fsum([self.scores[hit.origin_inchikey] * hit.value for hit in hits]) / len(hits)


class SumUnweightedCalc(EnrichmentCalculator[S]):

Missing class docstring

Value 'EnrichmentCalculator' is unsubscriptable

    def for_pair(self, hits: Sequence[AbstractHit]) -> float:

Missing function or method docstring

        return math.fsum([self.scores[hit.origin_inchikey] for hit in hits]) / len(hits)


class EnrichmentAlg(CleverEnum):

Empty class docstring

    """ """

    alpha = enum.auto()
    fold = enum.auto()
    fold_w = enum.auto()
    sum = enum.auto()
    sum_w = enum.auto()

    @property
    def description(self) -> str:

Missing function or method docstring

        s = self.symbol

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

        return {
            EnrichmentAlg.alpha: rf"[float] {s}(p) = Mean product of rescaled weights and scores; see the docs",

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

            EnrichmentAlg.fold: rf"[bool] {s}(p) = #(c has p and hit) / #(c has p and not hit)",
            EnrichmentAlg.fold_w: rf"[bool] {s}(p) = ∑(w(c, pair) s.t. c is hit) / ∑(w(c, pair) s.t. c not hit)",

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

            EnrichmentAlg.sum: rf"{s}(p) = ∑(score(c) s.t. c has p)",
            EnrichmentAlg.sum_w: rf"{s}(p) = ∑(score(c) × w(c, p) for all c)",
        }[self]

    @property
    def symbol(self) -> str:

Missing function or method docstring

        return {
            EnrichmentAlg.alpha: "α",
            EnrichmentAlg.fold: r"β",
            EnrichmentAlg.fold_w: "β*",
            EnrichmentAlg.sum: r"γ",
            EnrichmentAlg.sum_w: r"γ*",
        }[self]

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

Missing function or method docstring

        return {
            EnrichmentAlg.alpha: AlphaCalculator,
            EnrichmentAlg.fold: FoldUnweightedCalc,
            EnrichmentAlg.fold_w: FoldWeightedCalc,
            EnrichmentAlg.sum: SumUnweightedCalc,
            EnrichmentAlg.sum_w: SumWeightedCalc,
        }[self.name]


class EnrichmentCalculation:

Missing class docstring

    def calc(

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

Missing function or method docstring

        self, data: HitFrame, score_df: ScoreDf, algorithm: Union[str, EnrichmentAlg]

The argument data seems to be unused.

Wrong hanging indentation before block (add 4 spaces).

    ) -> EnrichmentDf:
        hits = ...
        alg_clazz = EnrichmentDf.of(algorithm).clazz
        alg_name = str(algorithm)
        samples_per_pair = {k: len(v) for k, v in Au.hit_multidict(hits, "pair").items()}
        results = {}
        for col, scores in score_df.all_scores():
            calculated = alg_clazz(scores).calc(hits)
            results[col] = calculated
        return EnrichmentDf(
            [
                pd.Series(
                    {
                        **dict(
                            predicate=pair.pred,
                            object=pair.obj,
                            samples=n_samples,
                        ),
                        **{f"{alg_name}({c})": vs[pair] for c, vs in results.items()},
                    }
                )
                for pair, n_samples in samples_per_pair.items()
            ]
        )


__all__ = [
    "AlphaCalculator",
    "EnrichmentCalculator",
    "FoldUnweightedCalc",
    "FoldWeightedCalc",
    "SumUnweightedCalc",
    "SumWeightedCalc",
    "EnrichmentCalculation",
    "EnrichmentAlg",
    "EnrichmentDf",
    "ScoreDf",
]