Spaces:
Runtime error
Runtime error
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) | |