so_magic.som.manager

Complexity

Total Complexity

8

Size/Duplication

Total Lines	66
Duplicated Lines	0 %

Importance

Changes

0

7 Methods

Rating	Name	Duplication	Size	Complexity
A	MapId.__iter__()	0	3	1
A	MapId.__dir__()	0	2	1
A	MapId.__str__()	0	2	1
A	MapManager.get_map()	0	5	1
A	MagicMapManager.train()	0	2	1
A	MapManager.train()	0	2	1
A	MapId.from_self_organizing_map()	0	4	1

import logging
import attr

from so_magic.utils import ObjectsPool
from .factory import SelfOrganizingMapFactory

logger = logging.getLogger(__name__)


def _build_hash(_self, *args, **kwargs):
    return str(MapId(*args, kwargs.get('initialization'), kwargs.get('map_type'), kwargs.get('grid_type')))


@attr.s
class MapManager:
    map_factory = attr.ib(init=True, default=SelfOrganizingMapFactory())
    pool = attr.ib(init=False, default=attr.Factory(
        lambda self: ObjectsPool.new_empty(self.map_factory.create, build_hash=_build_hash), takes_self=True))

    def get_map(self, *args, **kwargs):
        """
        'dataset', 'nb_cols', 'nb_rows', 'initialization', 'map_type', 'grid_type'
        """
        return self.pool.get_object(*args, **kwargs)

    def train(self, dataset, nb_cols, nb_rows, **kwargs):
        return self.map_factory.create(dataset, nb_cols, nb_rows, **kwargs)


@attr.s
class MagicMapManager:
    so_master = attr.ib(init=True)
    manager = attr.ib(init=False, default=MapManager())

    def train(self, nb_cols, nb_rows, **kwargs):
        return self.manager.train(self.so_master.dataset, nb_cols, nb_rows, **kwargs)


@attr.s
class MapId:
    dataset_name = attr.ib(init=True)
    _n_columns = attr.ib(init=True)
    _n_rows = attr.ib(init=True)
    initialization = attr.ib(init=True)
    map_type = attr.ib(init=True)
    grid_type = attr.ib(init=True)

    @staticmethod
    def from_self_organizing_map(somap, **kwargs):
        return MapId(kwargs.get('dataset_name', somap.dataset_name),
                     *[getattr(somap, attribute.name) for attribute in MapId.__attrs_attrs__[1:]])

    def __dir__(self):
        return sorted([attribute.name for attribute in self.__attrs_attrs__])

    def __iter__(self):
        """Default implementation of __iter__ to allow dict(self) in client code"""
        return iter([(k, getattr(self, k)) for k in self.__dir__()])

    def __str__(self):
        return '-'.join(str(getattr(self, _)) for _ in dir(self))


if __name__ == '__main__':
    map_manager = MapManager()