Source code for featuretools.feature_base.features_deserializer

import json

from featuretools.entityset.deserialize import (
    description_to_entityset as deserialize_es,
)
from featuretools.feature_base.feature_base import (
    AggregationFeature,
    DirectFeature,
    Feature,
    FeatureBase,
    FeatureOutputSlice,
    GroupByTransformFeature,
    IdentityFeature,
    TransformFeature,
)
from featuretools.primitives.utils import PrimitivesDeserializer
from featuretools.utils.gen_utils import check_schema_version
from featuretools.utils.s3_utils import get_transport_params, use_smartopen_features
from featuretools.utils.wrangle import _is_s3, _is_url


def load_features(features, profile_name=None):
    """Loads the features from a filepath, S3 path, URL, an open file, or a JSON formatted string.

    Args:
        features (str or :class:`.FileObject`): The location of where features has
        been saved which this must include the name of the file, or a JSON formatted
        string, or a readable file handle where the features have been saved.

        profile_name (str, bool): The AWS profile specified to write to S3. Will default to None and search for AWS credentials.
            Set to False to use an anonymous profile.

    Returns:
        features (list[:class:`.FeatureBase`]): Feature definitions list.

    Note:
        Features saved in one version of Featuretools or python are not guaranteed to work in another.
        After upgrading Featuretools or python, features may need to be generated again.

    Example:
        .. ipython:: python
            :suppress:

            import featuretools as ft
            import os

        .. code-block:: python

            filepath = os.path.join('/Home/features/', 'list.json')
            ft.load_features(filepath)

            f = open(filepath, 'r')
            ft.load_features(f)

            feature_str = f.read()
            ft.load_features(feature_str)

    .. seealso::
        :func:`.save_features`
    """
    return FeaturesDeserializer.load(features, profile_name).to_list()


class FeaturesDeserializer(object):
    FEATURE_CLASSES = {
        "AggregationFeature": AggregationFeature,
        "DirectFeature": DirectFeature,
        "Feature": Feature,
        "FeatureBase": FeatureBase,
        "GroupByTransformFeature": GroupByTransformFeature,
        "IdentityFeature": IdentityFeature,
        "TransformFeature": TransformFeature,
        "FeatureOutputSlice": FeatureOutputSlice,
    }

    def __init__(self, features_dict):
        self.features_dict = features_dict
        self._check_schema_version()
        self.entityset = deserialize_es(features_dict["entityset"])
        self._deserialized_features = {}  # name -> feature
        self._primitives_deserializer = PrimitivesDeserializer()

    @classmethod
    def load(cls, features, profile_name):
        if isinstance(features, str):
            try:
                features_dict = json.loads(features)
            except ValueError:
                if _is_url(features) or _is_s3(features):
                    transport_params = None
                    if _is_s3(features):
                        transport_params = get_transport_params(profile_name)
                    features_dict = use_smartopen_features(
                        features, transport_params=transport_params
                    )
                else:
                    with open(features, "r") as f:
                        features_dict = json.load(f)
            return cls(features_dict)
        return cls(json.load(features))

    def to_list(self):
        feature_names = self.features_dict["feature_list"]
        return [self._deserialize_feature(name) for name in feature_names]

    def _deserialize_feature(self, feature_name):
        if feature_name in self._deserialized_features:
            return self._deserialized_features[feature_name]

        feature_dict = self.features_dict["feature_definitions"][feature_name]
        dependencies_list = feature_dict["dependencies"]

        # Collect dependencies into a dictionary of name -> feature.
        dependencies = {
            dependency: self._deserialize_feature(dependency)
            for dependency in dependencies_list
        }

        type = feature_dict["type"]
        cls = self.FEATURE_CLASSES.get(type)
        if not cls:
            raise RuntimeError('Unrecognized feature type "%s"' % type)

        args = feature_dict["arguments"]
        feature = cls.from_dictionary(
            args, self.entityset, dependencies, self._primitives_deserializer
        )

        self._deserialized_features[feature_name] = feature
        return feature

    def _check_schema_version(self):
        check_schema_version(self, "features")