<filename>previsionio/model.py<gh_stars>1-10
# -*- coding: utf-8 -*-
from typing import Dict, Union

from pandas.core.frame import DataFrame
from previsionio.experiment_config import TypeProblem
import json
import uuid
import requests
import pandas as pd
from functools import lru_cache

from .logger import logger
from .dataset import Dataset
# from .deployed_model import DeployedModel
from .prevision_client import client
from .api_resource import ApiResource
from .utils import parse_json, PrevisionException, zip_to_pandas
from .prediction import ValidationPrediction


class Model(ApiResource):
    """ A Model object is generated by Prevision AutoML plateform when you launch an experiment version.
    All models generated by Prevision.io are deployable in our Store

    With this Model class, you can select the model with the optimal hyperparameters
    that responds to your buisiness requirements, then you can deploy it
    as a real-time/batch endpoint that can be used for a web Service.

    Args:
        _id (str): Unique id of the model
        experiment_version_id (str): Unique id of the experiment version of the model
        name (str, optional): Name of the model (default: ``None``)
    """

    def __init__(self, _id: str, experiment_version_id: str, project_id: str, model_name: str = None,
                 deployable: bool = False, **other_params):
        """ Instantiate a new :class:`.Model` object to manipulate a model resource on the platform. """
        super().__init__(_id=_id)
        self._id = _id
        self.experiment_version_id = experiment_version_id
        self.project_id = project_id
        self.name = model_name
        self.tags = {}
        self.deployable = deployable

        for k, v in other_params.items():
            self.__setattr__(k, v)

    def __repr__(self):
        return str(self._id)

    def __str__(self):
        """ Show up the Model object attributes.

        Returns:
            str: JSON-formatted info
        """
        args_to_show = {k: self.__dict__[k]
                        for k in self.__dict__
                        if all(map(lambda x: x not in k.lower(),
                                   ["event", "compositiondetails"])
                               )
                        }

        return json.dumps(args_to_show, sort_keys=True, indent=4, separators=(',', ': '))

    @classmethod
    def from_id(cls, _id: str) -> Union[
            'RegressionModel',
            'ClassificationModel',
            'MultiClassificationModel',
            'ExternalRegressionModel',
            'ExternalClassificationModel',
            'ExternalMultiClassificationModel',
            'TextSimilarityModel']:
        """Get an experiment from the platform by its unique id.

        Args:
            _id (str): Unique id of the experiment to retrieve
            version (int, optional): Specific version of the experiment to retrieve
                (default: 1)

        Returns:
            :class:`.BaseExperimentVersion`: Fetched experiment

        Raises:
            PrevisionException: Any error while fetching data from the platform
                or parsing result
        """
        end_point = '/models/{}'.format(_id)
        response = client.request(endpoint=end_point,
                                  method=requests.get,
                                  message_prefix='Model from id')
        model = json.loads(response.content.decode('utf-8'))
        training_type = TypeProblem(model.get('training_type', model.get('type_problem')))
        # NOTE: we should always set the provider in model collection
        provider = model.get('provider', 'prevision-auto-ml')
        if provider == 'prevision-auto-ml':
            if training_type == TypeProblem.Regression:
                return RegressionModel(**model)
            elif training_type == TypeProblem.Classification:
                return ClassificationModel(**model)
            elif training_type == TypeProblem.MultiClassification:
                return MultiClassificationModel(**model)
            elif training_type == TypeProblem.TextSimilarity:
                return TextSimilarityModel(**model)
            else:
                raise PrevisionException('Training type {} not supported'.format(training_type))
        elif provider == 'external':
            if training_type == TypeProblem.Regression:
                return ExternalRegressionModel(**model)
            elif training_type == TypeProblem.Classification:
                return ExternalClassificationModel(**model)
            elif training_type == TypeProblem.MultiClassification:
                return ExternalMultiClassificationModel(**model)
            else:
                raise PrevisionException('Training type {} not supported'.format(training_type))
        else:
            raise PrevisionException('Provider {} not supported'.format(provider))

    @property
    @lru_cache()
    def hyperparameters(self):
        """ Return the hyperparameters of a model.

        Returns:
            dict: Hyperparameters of the model
        """
        url = '/models/{}/hyperparameters/download'.format(self._id)
        response = client.request(
            endpoint=url,
            method=requests.get,
            message_prefix='Hyperparameters')
        return (json.loads(response.content.decode('utf-8')))

    def _predict_bulk(self,
                      dataset_id: str,
                      confidence: bool = False,
                      dataset_folder_id: str = None):
        """ (Util method) Private method used to handle bulk predict.

        .. note::

            This function should not be used directly. Use predict_from_* methods instead.

        Args:
            dataset_id (str): Unique id of the dataset to predict with
            confidence (bool, optional): Whether to predict with confidence estimator (default: ``False``)
            dataset_folder_id (str, optional): Unique id of the associated folder dataset to predict with,
                if need be (default: ``None``)

        Returns:
            :class:`.ValidationPrediction`: The registered prediction object in the current workspace

        Raises:
            PrevisionException: Any error while starting the prediction on the platform or parsing the result
        """
        data = {
            'dataset_id': dataset_id,
            'model_id': self._id,
            'confidence': str(confidence).lower(),
        }

        if dataset_folder_id is not None:
            data['folder_dataset_id'] = dataset_folder_id

        predict_start = client.request(
            '/experiment-versions/{}/validation-predictions'.format(self.experiment_version_id),
            method=requests.post,
            data=data,
            message_prefix='Bulk predict',
        )
        predict_start_parsed = parse_json(predict_start)

        if '_id' not in predict_start_parsed:
            err = 'Error starting prediction: {}'.format(predict_start_parsed)
            logger.error(err)
            raise PrevisionException(err)
        return ValidationPrediction(**predict_start_parsed)

    def predict_from_dataset(self, dataset: Dataset,
                             confidence: bool = False,
                             dataset_folder: Dataset = None) -> ValidationPrediction:
        """ Make a prediction for a dataset stored in the current active [client]
        workspace (using the current SDK dataset object).

        Args:
            dataset (:class:`.Dataset`): Dataset resource to make a prediction for
            confidence (bool, optional): Whether to predict with confidence values (default: ``False``)
            dataset_folder (:class:`.Dataset`, None): Matching folder dataset resource for the prediction,
                if necessary

        Returns:
            :class:`.ValidationPrediction`: The registered prediction object in the current workspace
        """

        prediction = self._predict_bulk(dataset.id,
                                        confidence=confidence,
                                        dataset_folder_id=dataset_folder.id if dataset_folder else None)

        return prediction

    def predict(self, df: DataFrame, confidence: bool = False, prediction_dataset_name: str = None) -> pd.DataFrame:
        """ Make a prediction in a Scikit-learn blocking style.

        .. warning::

            For large dataframes and complex (blend) models, this can be slow (up to 1-2 hours). Prefer using
            this for simple models and small dataframes or use option ``use_best_single = True``.

        Args:
            df (``pd.DataFrame``): A ``pandas`` dataframe containing the testing data
            confidence (bool, optional): Whether to predict with confidence estimator (default: ``False``)

        Returns:
            ``pd.DataFrame``: Prediction results dataframe
        """
        if prediction_dataset_name is None:
            prediction_dataset_name = 'test_{}_{}'.format(self.name, str(uuid.uuid4())[-6:])

        dataset = Dataset._new(self.project_id, prediction_dataset_name, dataframe=df)

        prediction = self._predict_bulk(dataset.id,
                                        confidence=confidence)

        return prediction.get_result()


class ClassicModel(Model):

    @property
    @lru_cache()
    def feature_importance(self) -> pd.DataFrame:
        """ Return a dataframe of feature importances for the given model features, with their corresponding
        scores (sorted by descending feature importance scores).

        Returns:
            ``pd.DataFrame``: Dataframe of feature importances

        Raises:
            PrevisionException: Any error while fetching data from the platform or parsing the result
        """
        endpoint = '/models/{}/features-importances/download'.format(self._id)
        response = client.request(endpoint=endpoint,
                                  method=requests.get,
                                  message_prefix='Fetch feature importance')

        df_feat_importance = zip_to_pandas(response)

        return df_feat_importance.sort_values(by="importance", ascending=False)

    @property
    def cross_validation(self) -> pd.DataFrame:
        """ Get model's cross validation dataframe.

        Returns:
            ``pd.Dataframe``: Cross-validation dataframe
        """
        logger.debug('getting cv, model_id: {}'.format(self.id))
        url = '/models/{}/cross-validation/download'.format(self._id)
        cv_response = client.request(url,
                                     requests.get,
                                     message_prefix="Fetch cross validation")
        df_cv = zip_to_pandas(cv_response)

        return df_cv

    def chart(self):
        """ Return chart analysis information for a model.

        Returns:
            dict: Chart analysis results

        Raises:
            PrevisionException: Any error while fetching data from the platform or parsing the result
        """
        endpoint = '/models/{}/analysis'.format(self._id)
        response = client.request(
            endpoint=endpoint,
            method=requests.get,
            message_prefix='Model analysis')

        result = (json.loads(response.content.decode('utf-8')))

        # drop chart-related information
        return result

    def predict_single(self, data: Dict, confidence: bool = False, explain: bool = False):
        """ Make a prediction for a single instance. Use :py:func:`predict_from_dataset_name` or
        :py:func:`predict` methods to predict multiple instances at the same time (it's faster).

        Args:
            data (dict): Features names and values (without target feature) - missing feature keys
                will be replaced by nans
            confidence (bool, optional): Whether to predict with confidence values (default: ``False``)
            explain (bool, optional): Whether to explain prediction (default: ``False``)

        Returns:
            dict: Dictionary containing the prediction result

            .. note::

                The prediction format depends on the problem type (regression, classification, etc...)
        """
        payload = {
            'features': {
                str(k): v for k, v in data.items() if str(v) != 'nan'
            },
            'explain': explain,
            'confidence': confidence,
            'model_id': self._id
        }

        logger.debug('[Predict Unit] sending payload ' + str(payload))
        url = '/experiment-versions/{}/unit-prediction'.format(self.experiment_version_id)
        response = client.request(url,
                                  requests.post,
                                  data=payload,
                                  content_type='application/json',
                                  message_prefix="Unit predict")
        response_json = parse_json(response)
        return response_json


class ClassificationModel(ClassicModel):
    """ A model object for a (binary) classification experiment, i.e. an experiment where the target
    is categorical with exactly 2 modalities.

    Args:
        _id (str): Unique id of the model
        experiment_version_id (str): Unique id of the experiment version of the model
        experiment_version (str, int): Version of the experiment of the model (either an integer for a specific
            version, or "last")
        name (str, optional): Name of the model (default: ``None``)
    """

    def __init__(self, _id, experiment_version_id, **other_params):
        """ Instantiate a new :class:`.ClassificationModel` object to manipulate a classification model
        resource on the platform. """
        super().__init__(_id, experiment_version_id, **other_params)
        self._predict_threshold = 0.5

    @property
    @lru_cache()
    def optimal_threshold(self):
        """ Get the value of threshold probability that optimizes the F1 Score.

        Returns:
            float: Optimal value of the threshold (if it not a classification problem it returns ``None``)

        Raises:
            PrevisionException: Any error while fetching data from the platform or parsing the result
        """
        return self.chart()["optimal_proba"]

    def get_dynamic_performances(self, threshold: float = 0.5):
        """ Get model performance for the given threshold.

        Args:
            threshold (float, optional): Threshold to check the model's performance for (default: 0.5)

        Returns:
            dict: Model classification performance dict with the following keys:

                - ``confusion_matrix``
                - ``accuracy``
                - ``precision``
                - ``recall``
                - ``f1_score``

        Raises:
            PrevisionException: Any error while fetching data from the platform or parsing the result
        """
        threshold = float(threshold)
        if any((threshold > 1, threshold < 0)):
            err = 'threshold value has to be between 0 and 1'
            logger.error(err)
            raise ValueError(err)

        result = dict()
        query = '?threshold={}'.format(str(threshold))
        endpoint = '/models/{}/dynamic-analysis{}'.format(self._id, query)

        response = client.request(endpoint=endpoint,
                                  method=requests.get,
                                  message_prefix='Dynamic analysis')
        resp = json.loads(response.content.decode('utf-8'))

        result['confusion_matrix'] = resp["confusion_matrix"]
        for metric in ['accuracy', 'precision', 'recall', 'f1Score']:
            result[metric] = resp["score"][metric]

        return result


class RegressionModel(ClassicModel):
    """ A model object for a regression experiment, i.e. an experiment where the target is numerical.

    Args:
        _id (str): Unique id of the model
        experiment_version_id (str): Unique id of the experiment version of the model
        experiment_version (str, int): Version of the experiment of the model (either an integer for a specific
            version, or "last")
        name (str, optional): Name of the model (default: ``None``)
    """


class MultiClassificationModel(ClassicModel):
    """ A model object for a multi-classification experiment, i.e. an experiment where the target
    is categorical with strictly more than 2 modalities.

    Args:
        _id (str): Unique id of the model
        experiment_version_id (str): Unique id of the experiment version of the model
        experiment_version (str, int): Version of the experiment of the model (either an integer for a specific
            version, or "last")
        name (str, optional): Name of the model (default: ``None``)
    """

    def predict_single(self, data: Dict, confidence: bool = False, explain: bool = False):
        raise NotImplementedError


# NOTE: we inherit the external models classes from classic model classes but with 2 method not implemented

class ExternalClassificationModel(ClassificationModel):
    """ A model object for an external (binary) classification experiment, i.e. an experiment where the target
    is categorical with exactly 2 modalities.

    Args:
        _id (str): Unique id of the model
        experiment_version_id (str): Unique id of the experiment version of the model
        experiment_version (str, int): Version of the experiment of the model (either an integer for a specific
            version, or "last")
        name (str, optional): Name of the model (default: ``None``)
    """

    @property
    def feature_importance(self) -> pd.DataFrame:
        raise NotImplementedError

    @property
    def cross_validation(self) -> pd.DataFrame:
        raise NotImplementedError

    def predict_single(self, data: Dict, confidence: bool = False, explain: bool = False):
        raise NotImplementedError


class ExternalRegressionModel(RegressionModel):
    """ A model object for an external regression experiment, i.e. an experiment where the target is numerical.

    Args:
        _id (str): Unique id of the model
        experiment_version_id (str): Unique id of the experiment version of the model
        experiment_version (str, int): Version of the experiment of the model (either an integer for a specific
            version, or "last")
        name (str, optional): Name of the model (default: ``None``)
    """

    @property
    def feature_importance(self) -> pd.DataFrame:
        raise NotImplementedError

    @property
    def cross_validation(self) -> pd.DataFrame:
        raise NotImplementedError

    def predict_single(self, data: Dict, confidence: bool = False, explain: bool = False):
        raise NotImplementedError


class ExternalMultiClassificationModel(MultiClassificationModel):
    """ A model object for an external multi-classification experiment, i.e. an experiment where the target
    is categorical with strictly more than 2 modalities.

    Args:
        _id (str): Unique id of the model
        experiment_version_id (str): Unique id of the experiment version of the model
        experiment_version (str, int): Version of the experiment of the model (either an integer for a specific
            version, or "last")
        name (str, optional): Name of the model (default: ``None``)
    """

    @property
    def feature_importance(self) -> pd.DataFrame:
        raise NotImplementedError

    @property
    def cross_validation(self) -> pd.DataFrame:
        raise NotImplementedError

    def predict_single(self, data: Dict, confidence: bool = False, explain: bool = False):
        raise NotImplementedError


class TextSimilarityModel(Model):

    def _predict_bulk(self,
                      queries_dataset_id: str,
                      queries_dataset_content_column: str,
                      top_k: int,
                      matching_id_description_column: str = None):
        """ (Util method) Private method used to handle bulk predict.

        .. note::

            This function should not be used directly. Use predict_from_* methods instead.

        Args:
            queries_dataset_id (str): Unique id of the quries dataset to predict with
            queries_dataset_content_column (str): Content queries column name
            queries_dataset_matching_id_description_column (str): Matching id description column name
            top_k (integer): Number of the nearest description to predict
        Returns:
            :class:`.ValidationPrediction`: The registered prediction object in the current workspace

        Raises:
            PrevisionException: Any error while starting the prediction on the platform or parsing the result
        """
        data = {
            'model_id': self._id,
            'queries_dataset_id': queries_dataset_id,
            'queries_dataset_content_column': queries_dataset_content_column,
            'top_k': top_k
        }
        if matching_id_description_column:
            data['queries_dataset_matching_id_description_column'] = matching_id_description_column
        endpoint = '/experiment-versions/{}/validation-predictions'.format(self.experiment_version_id)
        predict_start = client.request(endpoint,
                                       method=requests.post,
                                       data=data,
                                       message_prefix='Text-similarity bulk predict')
        predict_start_parsed = parse_json(predict_start)

        if '_id' not in predict_start_parsed:
            err = 'Error starting prediction: {}'.format(predict_start_parsed)
            logger.error(err)
            raise PrevisionException(err)

        return ValidationPrediction(**predict_start_parsed)

    def predict_from_dataset(self, queries_dataset: Dataset, queries_dataset_content_column: str, top_k: int = 10,
                             queries_dataset_matching_id_description_column: str = None) -> ValidationPrediction:
        """ Make a prediction for a dataset stored in the current active [client]
        workspace (using the current SDK dataset object).

        Args:
            dataset (:class:`.Dataset`): Dataset resource to make a prediction for
            queries_dataset_content_column (str): Content queries column name
            top_k (integer): Number of the nearest description to predict
            queries_dataset_matching_id_description_column (str): Matching id description column name

        Returns:
            :class:`.ValidationPrediction`: The registered prediction object in the current workspace
        """
        prediction = self._predict_bulk(queries_dataset.id,
                                        queries_dataset_content_column,
                                        top_k=top_k,
                                        matching_id_description_column=queries_dataset_matching_id_description_column)
        return prediction

    def predict(self, df: DataFrame, queries_dataset_content_column: str, top_k: int = 10,
                queries_dataset_matching_id_description_column: str = None,
                prediction_dataset_name: str = None) -> Union[pd.DataFrame, None]:
        """ Make a prediction for a dataset stored in the current active [client]
        workspace (using the current SDK dataset object).

        Args:
            dataset (:class:`.Dataset`): Dataset resource to make a prediction for
            queries_dataset_content_column (str): Content queries column name
            top_k (integer): Number of the nearest description to predict
            queries_dataset_matching_id_description_column (str): Matching id description column name

        Returns:
            ``pd.DataFrame``: Prediction results dataframe
        """
        if prediction_dataset_name is None:
            prediction_dataset_name = 'test_{}_{}'.format(self.name, str(uuid.uuid4())[-6:])

        dataset = Dataset._new(self.project_id, prediction_dataset_name, dataframe=df)
        prediction = self._predict_bulk(dataset.id,
                                        queries_dataset_content_column,
                                        top_k=top_k,
                                        matching_id_description_column=queries_dataset_matching_id_description_column)

        return prediction.get_result()
