src/idsc/Profiling.py

import requests
import json
import pickle
import os


class Profiling():
    def __init__(self, apikey):
        self.apikey = apikey
        pass

    def profile(self, ts):
        '''
        ts: list of number of timeseries value, the y column

        return: 
        {
            'classification_res':{
                'time_series_class':{
                    'overall_characteristic':'continuous'
                    }
                },
            'change_point_res':{'inter_order_cpi':[]},
            '',
        }
        '''
        print('Start profiling, note, predictability been disabled')

        # ======= #
        # TESTING #
        # ======= #
        # return self.__load_res()

        self.change_point(list(ts))
        inter_order_cpi = self.change_point_res['inter_order_cpi']
        order_quantity_cpi = self.change_point_res['order_quantity_cpi']
        
        self.classification(list(ts), inter_order_cpi=inter_order_cpi)

        '''
        Predictability is disabled because not really required at this moment
        '''
        self.predictability(
            list(ts), 
            inter_order_cpi=inter_order_cpi, 
            order_quantity_cpi=order_quantity_cpi)

        profile_res = {
            'change_point_res': self.change_point_res,
            'classification_res': self.classification_res,
            'predictability_res': self.predictability_res
        }

        print('Predictability')
        print(self.predictability_res)

        # self.__save_res(profile_res)

        # print('Profiling Completed\n', profile_res)

        return profile_res

    def parse_res(self, res):
        if str(res.status_code)[0] != '2':
            raise ValueError('API Call Failed!\n', res.text)
        else:
            return res.json()

    def classification(self, ts, inter_order_cpi=None):
        endpoint = 'https://idsc.com.sg/foretell/profiling/classification'
        payloads = {'time_series': ts, 'inter_order_cpi': inter_order_cpi}
        headers = {'api-key': self.apikey}
        res = requests.post(endpoint, json=payloads, headers=headers)

        parsed_res = self.parse_res(res)

        # print('classification res',parsed_res)
        # parsed_res['time_series_class'] = json.loads(
        #     parsed_res['time_series_class'])

        self.classification_res = parsed_res
        return

    def change_point(
            self,
            ts,
            inter_order_penalty=None,
            order_qty_penalty=None):
        endpoint = 'https://idsc.com.sg/foretell/profiling/cpd'
        payloads = {'time_series': ts}

        if inter_order_penalty is not None:
            payloads['inter_order_penalty'] = inter_order_penalty
        if order_qty_penalty is not None:
            payloads['order_qty_penalty'] = order_qty_penalty

        headers = {'api-key': self.apikey}

        print('Change point detection')

        res = requests.post(endpoint, json=payloads, headers=headers)

        # print('Change point detection completed')

        self.change_point_res = self.parse_res(res)
        return

    def predictability(
            self,
            ts,
            inter_order_cpi=None,
            order_quantity_cpi=None,
            inter_discern_param=None,
            quantity_discern_param=None):
        endpoint = 'https://idsc.com.sg/foretell/profiling/predictability'

        payloads = {'time_series': ts}

        if inter_order_cpi is not None:
            payloads['inter_order_cpi'] = inter_order_cpi
        if order_quantity_cpi is not None:
            payloads['order_quantity_cpi'] = order_quantity_cpi
        if inter_discern_param is not None:
            payloads['inter_discern_param'] = inter_discern_param
        if quantity_discern_param is not None:
            payloads['quantity_discern_param'] = quantity_discern_param

        headers = {'api-key': self.apikey}
        res = requests.post(endpoint, json=payloads, headers=headers)

        self.predictability_res = self.parse_res(res)
        return

    # ------------------------------------------------ #
    # For testing purpose - save and load API response #
    # ------------------------------------------------ #

    def __save_res(self, res):
        # The successful res is based on "data/test.csv"
        script_dir = os.path.dirname(__file__)
        res_path = "./tests/profile_test.dict"
        abs_file_path = os.path.join(script_dir, res_path)

        print('Saving response to ' + res_path)

        with open(abs_file_path, 'wb') as f:
            pickle.dump(res, f)

    def __load_res(self):
        # The successful res is based on "data/test.csv"
        script_dir = os.path.dirname(__file__)
        res_path = "./tests/profile_test.dict"
        abs_file_path = os.path.join(script_dir, res_path)

        print('Reading response from ' + res_path)

        with open(abs_file_path, 'rb') as f:
            res = pickle.load(f)

        return res