Spaces:
Runtime error
Runtime error
File size: 3,824 Bytes
49ba314 |
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 |
import yaml
import requests
import os
import pandas as pd
from datetime import datetime, timedelta
from dotenv import load_dotenv
load_dotenv()
class IDSC_API():
'''
Wrapper class for all IDSC apis
'''
def __init__(self):
__location__ = os.path.realpath(
os.path.join(os.getcwd(), os.path.dirname(__file__)))
self.config_path = os.path.join(__location__, 'idsc_config.yaml')
self.logged_in = False
self.timeformat = '%m/%d/%Y, %H:%M:%S'
with open(self.config_path, 'r') as file:
self.config = yaml.safe_load(file)
self.expire = self.config['apikey_expire']
self.apikey = self.config['apikey']
self.login()
self.product_mix__res = None # Save product_mix api response
def login(self):
now = datetime.now()
expire_date = datetime.strptime(self.expire, self.timeformat)
if now >= expire_date:
print('apikey expired, requesting new one.')
self.apikey = self.fetch_apikey()
self.update_config()
else:
print('apikey still available, logged in')
self.logged_in = True
def fetch_apikey(self):
# print(os.environ)
json = {
# "email": self.config['email'],
# "pwd": self.config['password']
'email': os.getenv('IDSC_ACC'),
'pwd': os.getenv('IDSC_PASS')
}
print('IDSC Logging in ...')
login_resp = requests.post(
'https://idsc.com.sg/user/login',
json=json)
if login_resp.status_code == 200:
apikey = login_resp.json()["API_key"]
return apikey
else:
raise Exception(login_resp.json())
def update_config(self):
now = datetime.now()
expire = (now + timedelta(days=3)).strftime(self.timeformat)
self.config['apikey'] = self.apikey
self.config['apikey_expire'] = expire
with open(self.config_path, 'w') as f:
yaml.dump(self.config, f, default_flow_style=False)
# =========== #
# Product mix #
# =========== #
def product_mix(
self,
product_data,
storage_capacity,
budget_constraint):
endpint = 'https://idsc.com.sg/optimax/product-mix/product-mix'
json = {'product_data': product_data,
'storage_capacity': storage_capacity,
'budget_constraint': budget_constraint
}
print('storage_capacity', storage_capacity)
print('budget_constraint', budget_constraint)
self.login()
headers = {'api-key': self.apikey}
res = requests.post(endpint, json=json, headers=headers)
parsed_res = self._parse_response(res)
self.product_mix__res = parsed_res
return self.product_mix__res
def _parse_response(self, res):
if res.status_code != 200:
print(f'API call failed. {res.text}')
return False
return res.json()
def get_product_mix_recommendations_df(self):
try:
recommendations = self.product_mix__res['product_mix_recommendations']
except Exception:
return None
df_arr = []
for k, v in recommendations.items():
df = pd.DataFrame(v, index=[k])
df['SKU'] = k
# Make sure expected sales is always integer
df['expected_sales'] = int(df['expected_sales'])
# Rearrange the order, so 'SKU' columns is the first
cols = df.columns.tolist()
cols = cols[-1:] + cols[:-1]
df = df[cols]
df_arr.append(df)
return pd.concat(df_arr, ignore_index=True)
|