src/idsc/IDSC.py

import os
import yaml
from datetime import datetime, timedelta
import requests
import json

from .Profiling import Profiling
from .Prophet import Prophet
from .CEIF import CEIF
from .fft import fft
from .holt_winters import holt_winters
from .auto_arima import auto_arima


from dotenv import load_dotenv
# load_dotenv()


class IDSC():
    def __init__(self) -> None:
        __location__ = os.path.realpath(
            os.path.join(os.getcwd(), os.path.dirname(__file__)))

        self.config_path = os.path.join(__location__, 'config.yml')
        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()

    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)
        payloads = {
            '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=payloads)

        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

        self.expire = expire

        with open(self.config_path, 'w') as f:
            yaml.dump(self.config, f, default_flow_style=False)

    def profiling(self, ts):
        '''
        ts_list : list of integers for IDSC profiling APIs
        '''
        ts_obj = json.loads(ts)
        ts_list = ts_obj['target'].values()

        self.login()
        profiling = Profiling(apikey=self.apikey)

        return profiling.profile(ts_list)

        # return self.profile

    def prophet(self, ts: str, n_predict, profile):
        '''
        ts: time series object string
        profile: refer to self.profiling response
        '''
        self.login()
        prophet = Prophet(apikey=self.apikey)

        characteristic = profile['classification_res']['time_series_class']['overall_characteristic']
        inter_order_cpi = profile['change_point_res']['inter_order_cpi']

        if characteristic == 'continuous':
            res = prophet.continuous(
                ts, n_predict, inter_order_cpi=inter_order_cpi)
        else:
            res = prophet.intermittent(
                ts, n_predict, inter_order_cpi=inter_order_cpi)

        return res

    def ceif(
            self,
            ts: str,
            n_predict,
            quantity_score_weight=None,
            rate_score_weight=None,
            timing_score_weight=None):
        print('IDSC ceif')

        self.login()
        ceif = CEIF(apikey=self.apikey)

        ts_obj = json.loads(ts)
        ts_list = list(ts_obj['target'].values())

        res = ceif.forecast(
            ts_list,
            n_predict,
            quantity_score_weight=quantity_score_weight,
            rate_score_weight=rate_score_weight,
            timing_score_weight=timing_score_weight)
        print('IDSC ceif completed')

        return res

    def fft(
            self,
            ts: str,
            n_predict,
            num_harmonics=None):

        print('IDSC fft')

        self.login()
        model = fft(apikey=self.apikey)

        ts_obj = json.loads(ts)
        ts_list = list(ts_obj['target'].values())

        res = model.forecast(
            ts_list,
            n_predict,
            num_harmonics=num_harmonics)

        return res

    def holt_winters(self, ts, n_predict, seasonal_cycle=None):

        print('IDSC holt_winters')

        self.login()
        model = holt_winters(apikey=self.apikey)

        ts_obj = json.loads(ts)
        ts_list = list(ts_obj['target'].values())

        res = model.forecast(
            ts_list,
            n_predict,
            seasonal_cycle=seasonal_cycle)

        return res

    def auto_arima(self, ts, n_predict):
        print('IDSC auto_arima')

        self.login()
        model = auto_arima(apikey=self.apikey)
        ts_obj = json.loads(ts)
        ts_list = list(ts_obj['target'].values())

        res = model.forecast(
            ts_list,
            n_predict,
        )

        return res