Spaces:
Runtime error
Runtime error
File size: 4,972 Bytes
9ddee9f 8cf4695 9ddee9f 8cf4695 9ddee9f 8cf4695 9ddee9f 8cf4695 9ddee9f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 |
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
|