Source code for visions.types.type

from abc import ABCMeta, abstractmethod
from typing import Any, Callable, Dict, Optional, Sequence, Type, Union

import attr
import pandas as pd

from visions.relations import TypeRelation

_DEFAULT = object()


class RelationsIterManager:
    def __init__(self, relations: Sequence[TypeRelation]):
        self._keys: Dict["Type[VisionsBaseType]", int] = {
            item.related_type: i for i, item in enumerate(relations)
        }
        self.values = tuple(relations)

    def __getitem__(self, index: Union["Type[VisionsBaseType]", int]) -> TypeRelation:
        idx = index if isinstance(index, int) else self._keys[index]
        return self.values[idx]

    def get(
        self, index: Union["Type[VisionsBaseType]", int], default: Any = _DEFAULT
    ) -> Union[TypeRelation, Any]:
        try:
            return self[index]
        except (IndexError, KeyError) as err:
            if default is _DEFAULT:
                raise err
            else:
                return default


class VisionsBaseTypeMeta(ABCMeta):
    def __contains__(cls, series: pd.Series, state: dict = {}) -> bool:
        return cls.contains_op(series, state)  # type: ignore

    @property
    def relations(cls) -> RelationsIterManager:
        if cls._relations is None:  # type: ignore
            cls._relations = RelationsIterManager(cls.get_relations())  # type: ignore
        return cls._relations

    def __add__(cls, other):
        from visions.types import Generic
        from visions.typesets import VisionsTypeset

        if not any(issubclass(x, Generic) for x in [cls, other]):
            return VisionsTypeset([Generic, cls, other])
        return VisionsTypeset([cls, other])

    def __str__(cls) -> str:
        return str(cls.__name__)

    def __repr__(cls) -> str:
        return str(cls)


[docs]class VisionsBaseType(metaclass=VisionsBaseTypeMeta): """Abstract implementation of a vision type. Provides a common API for building custom visions data types. """ _relations: Optional[Sequence[TypeRelation]] = None
[docs] def __init__(self): raise ValueError("Types cannot be initialized")
@classmethod @abstractmethod def get_relations(cls) -> Sequence[TypeRelation]: raise NotImplementedError @classmethod @abstractmethod def contains_op(cls, series: pd.Series, state: dict) -> bool: raise NotImplementedError @classmethod def evolve_type( cls, type_name: str, relations_generator: Optional[ Callable[[Type[VisionsBaseTypeMeta]], Sequence[TypeRelation]] ] = None, replace: bool = False, ) -> "Type[VisionsBaseType]": """Make a copy of the type Args: type_name: the new type suffix, the type name will be `type[type_name]` relations_generator: a function returning all TypeRelations for the new type replace: if True, do not include the existing relations Returns: A new type """ def get_new_relations(cls) -> Sequence[TypeRelation]: return relations name = cls.__name__ new_type = type( f"{name}[{type_name}]", (cls,), { "get_relations": classmethod(get_new_relations), "contains_op": cls.contains_op, }, ) new_relations = ( list(relations_generator(new_type)) if relations_generator else [] ) if replace: assert ( relations_generator is not None ), "When calling evolve_type with `replace=True`, a `relations_generator` is required." relations = new_relations else: old_relations = [ attr.evolve(relation, type=new_type) for relation in cls.relations ] relations = old_relations + new_relations return new_type