mandos.analysis.filtration

Complexity

Total Complexity

24

Size/Duplication

Total Lines	116
Duplicated Lines	0 %

Importance

Changes

0

11 Methods

Rating	Name	Duplication	Size	Complexity
A	Filtration._matches()	0	4	3
A	Filtration.from_file()	0	4	1
A	Filtration.apply()	0	3	1
A	Operator.symbol()	0	5	1
A	Expression.parse()	0	7	2
A	Operator.rev_symbols()	0	3	1
A	Filtration.keep()	0	9	5
A	Expression.matches()	0	11	5
A	Operator.parse()	0	7	3
A	Filtration.__init__()	0	2	1
A	Filtration.from_toml()	0	4	1

"""
Tool to filter annotations.
"""
from __future__ import annotations

import re
from dataclasses import dataclass
import enum
from datetime import datetime
from pathlib import Path
from typing import Union, Sequence, Mapping, Set, Optional

from pocketutils.core.dot_dict import NestedDotDict

Unable to import 'pocketutils.core.dot_dict'

from mandos.model.hits import AbstractHit, HitFrame

_Type = Union[str, int, float, datetime]


class Operator(enum.Enum):

Empty class docstring

    """"""

    eq = enum.auto()
    ne = enum.auto()
    lt = enum.auto()
    gt = enum.auto()
    le = enum.auto()
    ge = enum.auto()
    like = enum.auto()
    is_in = enum.auto()
    not_in = enum.auto()

    @classmethod
    def parse(cls, s: str):

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

        if isinstance(s, Operator):
            return s
        if s in cls:
            return cls[s]
        return cls.rev_symbols().get(s)

    @property
    def symbol(self) -> str:

Missing function or method docstring

        return dict(
            eq="=", ne="!=", lt="<", gt=">", le="<=", ge=">=", like="$", is_in="<<", not_in="!<<"
        )[self.name]

    @classmethod
    def rev_symbols(cls) -> Mapping[str, Operator]:

Missing function or method docstring

        return {e.symbol: e for e in cls}


@dataclass(frozen=True, repr=True)

Empty class docstring

class Expression:
    """"""

    op: Operator

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

    val: _Type
    extra: Optional[Set[_Type]]

    def matches(self, value: _Type) -> bool:

Missing function or method docstring

        if type(value) != type(self.val):

Using type() instead of isinstance() for a typecheck.

            raise TypeError(f"{type(value)} != {type(self.val)}")
        if self.op is Operator.like:
            return re.compile(self.val).fullmatch(str(value)) is not None
        if self.op is Operator.is_in:
            return value in self.extra
        if self.op is Operator.not_in:
            return value not in self.extra
        call = f"__{self.op}__"
        return getattr(value, call) != getattr(self.val, call)

    @classmethod
    def parse(cls, s: str, context: NestedDotDict) -> Expression:

Missing function or method docstring

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

        op = Operator.parse(s)

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

        if op is None:
            op = Operator.eq

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

        val = s[len(op.symbol) :]
        return Expression(op, val, set(context.get(val, [])))


class Filtration:

Missing class docstring

    def __init__(self, key_to_statements: Mapping[str, Mapping[str, Sequence[Expression]]]):
        self._x = key_to_statements

    @classmethod
    def from_file(cls, path: Path) -> Filtration:

Missing function or method docstring

        dot = NestedDotDict.read_toml(path)
        return cls.from_toml(dot)

    @classmethod
    def from_toml(cls, dot: NestedDotDict) -> Filtration:

Missing function or method docstring

        data = dot.get("mandos.filter", [])
        return Filtration({d["key"]: {k: v for k, v in d.items() if k != "key"} for d in data})

    def apply(self, df: HitFrame) -> HitFrame:

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

        hits = [h for h in df.to_hits() if self.keep(h)]
        return HitFrame.from_hits(hits)

    def keep(self, hit: AbstractHit) -> bool:

Missing function or method docstring

        if hit.search_key not in self._x:
            return True
        for field, values in self._x[hit.search_key].items():

The variable values seems to be unused.

            if not hasattr(hit, field):
                raise ValueError(f"No field {field} in {hit.__class__.__name__}")
            if not self._matches(getattr(hit, field), field):
                return False
        return True

    def _matches(self, actual: _Type, allowed: Sequence[Expression]) -> bool:

Either all return statements in a function should return an expression, or none of them should.

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

        for e in allowed:

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

            if e.matches(actual):
                return True


__all__ = ["Filtration"]