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