src/idsc/CEIF.py

import requests
import json
import pickle
import os


class CEIF():
    def __init__(self, apikey) -> None:
        self.apikey = apikey

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

    def forecast(
            self,
            ts_list,
            n_predict,
            quantity_score_weight=None,
            rate_score_weight=None,
            timing_score_weight=None):

        endpoint = 'https://idsc.com.sg/foretell/prediction/time-series/intermittent/classic-explicit-intermittent'

        payloads = {
            'time_series_data': ts_list,
            'num_predict': n_predict}

        if quantity_score_weight is not None:
            payloads['quantity_score_weight'] = quantity_score_weight
        if rate_score_weight is not None:
            payloads['rate_score_weight'] = rate_score_weight
        if timing_score_weight is not None:
            payloads['timing_score_weight'] = timing_score_weight

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

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

        return self.parse_res(res)

    def __success_api_res_test(self):
        print('Warning - Using idsc ceif testing api response')

        script_dir = os.path.dirname(__file__)
        rel_path = "./tests/CEIF_success_res.dict"
        abs_file_path = os.path.join(script_dir, rel_path)

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

        return res

    def __save_res(self, res):
        '''
        Save the API response as local file for test
        '''
        with open('test/CEIF_success_res.dict', 'wb') as f:
            pickle.dump(res, f)