so_magic.data.datapoints.datapoints

Complexity

Total Complexity

23

Size/Duplication

Total Lines	131
Duplicated Lines	0 %

Importance

Changes

0

20 Methods

Rating	Name	Duplication	Size	Complexity
A	TabularData.get_numerical_attributes()	0	2	1
A	TabularData.row()	0	2	1
A	TabularData.__len__()	0	2	1
A	TabularData.attributes()	0	3	1
A	TabularData.iterrows()	0	2	1
A	TabularData.columns()	0	3	1
A	TabularData.rows()	0	3	1
A	TabularData.itercolumns()	0	2	1
A	StructuredData.attributes()	0	3	1
A	StructuredData.observations()	0	3	1
A	StructuredDataInterface.attributes()	0	4	1
A	TabularData.get_categorical_attributes()	0	2	1
A	AbstractTabularData.__iter__()	0	2	1
A	DatapointsInterface.observations()	0	4	1
A	TabularData.__iter__()	0	2	1
A	TabularData.nb_rows()	0	3	1
A	TabularData.column()	0	2	1
A	TabularData.nb_columns()	0	3	1
A	DatapointsFactory.create()	0	13	3
A	DatapointsFactory.register_constructor()	0	6	1

from abc import ABC, abstractmethod
from typing import Iterable
import attr

from so_magic.utils import Observer
from .tabular_data_interface import TabularDataInterface


class DatapointsInterface(ABC):
    """The Datapoints interface gives access to the 'observations' property."""
    @property
    @abstractmethod
    def observations(self):
        raise NotImplementedError

class StructuredDataInterface(ABC):
    @property
    @abstractmethod
    def attributes(self):
        raise NotImplementedError



class DatapointsFactory:
    constructors = {}

    @classmethod
    def register_constructor(cls, name):
        def wrapper(subclass):
            cls.constructors[name] = subclass
            return subclass
        return wrapper

    @classmethod
    def create(cls, name, *args, **kwargs) -> Iterable:
        if name not in cls.constructors:
            raise ValueError(
                f"Request Engine of type '{name}'; supported are [{', '.join(sorted(cls.constructors.keys()))}]")
        try:
            return cls.constructors[name](*args, **kwargs)
        except TypeError as e:
            print(f"Datapoints creation failed. Args: [{', '.join(f'{i}: {str(_)}' for i, _ in enumerate(args))}]")
            print(f"Kwargs: [{', '.join(f'{k}: {v}' for k, v in kwargs.items())}]")
            print(e)
            import sys
            sys.exit(1)


@attr.s
@DatapointsFactory.register_constructor('structured-data')
class StructuredData(DatapointsInterface, StructuredDataInterface):
    """Structured data. There are specific attributes/variables per observation.

    Args:
        observations (object): a reference to an object that encapsulates structured data
    """
    _observations = attr.ib(init=True)
    _attributes = attr.ib(init=True, converter=lambda input_value: [x for x in input_value])

    @property
    def attributes(self):
        return self._attributes

    @property
    def observations(self):
        return self._observations

    @observations.setter
    def observations(self, observations):
        self._observations = observations


class AbstractTabularData(StructuredData, TabularDataInterface, ABC):

    def __iter__(self):
        return self.iterrows()


@attr.s
@DatapointsFactory.register_constructor('tabular-data')
class TabularData(AbstractTabularData):
    """Table-like datapoints that are loaded in memory"""

    @property
    def columns(self) -> Iterable:
        pass

    @property
    def rows(self) -> Iterable:
        pass

    retriever = attr.ib(init=True)
    iterator = attr.ib(init=True)
    mutator = attr.ib(init=True)

    @property
    def attributes(self):
        return self.iterator.columnnames(self)

    def column(self, identifier):
        return self.retriever.column(identifier, self)

    def row(self, identifier):
        return self.retriever.row(identifier, self)

    def get_numerical_attributes(self):
        return self.retriever.get_numerical_attributes(self)

    def get_categorical_attributes(self):
        return iter(set(self.attributes) - set([_ for _ in self.retriever.get_numerical_attributes(self)]))

    @property
    def nb_columns(self):
        return self.retriever.nb_columns(self)

    @property
    def nb_rows(self):
        return self.retriever.nb_rows(self)

    def __len__(self):
        return self.retriever.nb_rows(self)

    def __iter__(self):
        return self.iterator.iterrows(self)

    def iterrows(self):
        return self.iterator.iterrows(self)

    def itercolumns(self):
        return self.iterator.itercolumns(self)