lookup.py

Go to the documentation of this file.

# -*- coding: utf-8 -*-
"""
Created on Tue May  5 09:20:29 2020
 
@author: cd
"""
 
 
import abc
import numpy as np
# import time as tm
import datetime as dt
import calendar as cl
from datetime import timedelta
from scipy.interpolate import interp1d
# import warnings
# warnings.filterwarnings('error', category=RuntimeWarning)
 
 
class LkupTblBase(metaclass=abc.ABCMeta):
    def __init__(self, name, idxVals, lkupVals):
        self.namename = name
        # TODO: Error check that number of idxVals == lkupVals
        self.idxValsidxVals = idxVals
        self.lkupValslkupVals = lkupVals
        self._minIdx_minIdx = np.amin(self.idxValsidxVals)
        self._maxIdx_maxIdx = np.amax(self.idxValsidxVals)
        super().__init__()
 
    @abc.abstractmethod
    def get_val(self, idxVal):
        pass
 
    @classmethod
    def __subclasshook__(cls, C):
        if cls is LkupTblBase:
            attrs = set(dir(C))
            if set(cls.__abstractmethods__) <= attrs:
                return True
        return NotImplemented
 
    @property
    def maxIdx(self):
        return self._maxIdx_maxIdx
 
    @property
    def minIdx(self):
        return self._minIdx_minIdx
 
 
class LkupTbl(LkupTblBase):
    def __init__(self, name, idxVals, lkupVals):
        # Call super constructor
        super().__init__(name, idxVals, lkupVals)
 
    def get_val(self, valLkup):
        # iLkup = np.where(self.idxVals <= idxLkup)[0]
        # if iLkup.size == 0: iLkup = np.full(1, int(0))
        # return self.lkupVals[iLkup[-1]]
        try:
            iLkup = np.where(self.idxValsidxVals <= valLkup)[0]
            if iLkup.size == 0: iLkup = np.array([0])
            return self.lkupValslkupVals[iLkup[-1]].item()
        except Warning as e:
            print(e)
            print(valLkup)
        
 
class LkupTblElev(LkupTbl):
    def __init__(self, name, hypso, xElev, lkupVals):
        # Convert rating table to storage
        # xStor = np.interp(xElev, hypso.elev, hypso.stor)
        xStor = hypso.elev2stor(xElev)
        # Call super constructor
        super().__init__(name, xStor, lkupVals)
 
 
# TODO: Use SciPy Interp function here, see example in the YubaFeather project folder
# TODO: The interp function would be a property of the class (might increase speed)
class LkupTblInterp(LkupTblBase):
    def __init__(self, name, idxVals, lkupVals):
        # Call super constructor
        super().__init__(name, idxVals, lkupVals)
        self.interpFncinterpFnc = interp1d(self.idxValsidxVals, self.lkupValslkupVals, kind='linear', 
                                  fill_value=(min(lkupVals),max(lkupVals)), bounds_error=False)
        # self.interpFnc = interp1d(self.idxVals, self.lkupVals, kind='cubic')
 
    def get_val(self, xVals):
        # TODO: Scipy alternative the should work better because you could save the function as a property (faster)
        # if isinstance(xVals, np.ndarray):
        #     valsReturn = np.empty(xVals.size)
        #     for i, curVal in enumerate(xVals):
        #         valsReturn[i] = np.interp(curVal, self.idxVals, self.lkupVals)
        #     return valsReturn
        # else:
        #     return np.interp(xVals, self.idxVals, self.lkupVals)
        return self.interpFncinterpFnc(xVals)
        
 
 
# # TODO: This should be modified to return multiple values based on a date range
# # TODO: Maybe this should subclass LkupTblInterp to provide the most flexibility
# # TODO: Need error checking to ensure the proper values are passed
# class LkupTblAnn(LkupTblBase):
#     def __init__(self, name, lkupVals, timeUnit):
#         # TODO: On init if the timestep of the values passed does not match the
#         # timestep of the model then use interp to generate the values
#         self.timeUnit = timeUnit
#         # Create a datetime64 array for leap year 2020
#         dtLeapYr = DatetimeIndex(np.arange("2020-01-01", "2021-01-01", np.timedelta64(1, timeUnit),
#             dtype="datetime64[{}]".format(timeUnit)).astype("datetime64[{}]".format(timeUnit)))
#         month = [curDt.month for curDt in dtLeapYr]
#         day = [curDt.day for curDt in dtLeapYr]
#         hour = [curDt.hour for curDt in dtLeapYr]
#         # TODO: Could create a pandas table with different month, day, hour columns
#         # TODO: This could also be faster if it was just based on a single time step
#         # of year key and use the datetime day_of_year etc... functions; this might
#         # be better for retrieving multiple values for a date range. Also then this
#         # could subclass LkupTbl to use less code.
#         self.vDate = np.array(list(zip(*[month, day, hour])))
#         # TODO: Add error checking to make sure number of passed vals = number of dates
#         super().__init__(name, -1, lkupVals)
 
#     def get_val(self, dateTime, col=0):
#         iTime = self.vDate[:, 0] == dateTime.month
#         if self.timeUnit == 'D':
#             iTime = iTime & (self.vDate[:, 1] == dateTime.day)
#         elif self.timeUnit == 'h':
#             iTime = iTime & (self.vDate[:, 2] == dateTime.hour)
#         # TODO: There's got to be a cleaner way to do this:
#         rowLkup = np.where(iTime)[0][0]
#         if self.lkupVals.ndim > 1:
#             return self.lkupVals[rowLkup, col]
#         else:
#             return self.lkupVals[rowLkup]
        
        
# TODO: This should default to timeUnit='h'
class LkupTblAnn(LkupTblBase):
    def __init__(self, name, time, monthDayHr, tblVals, *, typ='interp', timeUnit=None):
        # TODO: This should take a keyword argument to override the timestep of the model
        # for instance if the table timestep is monthly then no sense in creating hourly values
        # TODO: Evaluate using numpy Datetime64 for all the date operations. Faster?
        self.typtyp = typ
        if timeUnit is not None:
            # TODO: This should be set to a default of 'h' and not None
            self.timeUnittimeUnit = timeUnit
            if timeUnit == 'M':
                self.tDeltatDelta = np.nan
            elif timeUnit == 'D':
                self.tDeltatDelta = timedelta(days=1)
            else:
                self.tDeltatDelta = timedelta(hours=1)
        else:
            self.timeUnittimeUnit = time.timeUnit
            self.tDeltatDelta = time.tDelta
        # TODO: This should be able to handle single value scalars as well as arrays
        if np.isscalar(tblVals): tblVals = np.array([tblVals], dtype=float)
        tblVals = tblVals.reshape(len(tblVals), tblVals.ndim) if tblVals.ndim == 1 else tblVals
        monthDayHr = monthDayHr.reshape(
            monthDayHr.ndim, monthDayHr.shape[0]) if monthDayHr.ndim == 1 else monthDayHr
        xDateTime = np.empty(monthDayHr.shape[0] + 1)
        yVals = np.empty([monthDayHr.shape[0] + 1, tblVals.shape[1]])
        # TODO: If jan 1 is not the first date then set it as first date?
        for i in range(0, monthDayHr.shape[0]):
            curDateTime = dt.datetime(2020,
                                      monthDayHr[i, 0],
                                      monthDayHr[i, 1],
                                      monthDayHr[i, 2], tzinfo=dt.timezone.utc)
            xDateTime[i] = cl.timegm(curDateTime.timetuple())
            yVals[i, :] = tblVals[i, :]
        # Add date for beginning of 2021 to complete lookup table
        xDateTime[-1] = cl.timegm(dt.datetime(2021, 1, 1, 0, tzinfo=dt.timezone.utc).timetuple())
        yVals[-1, :] = tblVals[-1, :]
        # Build array of interpolation values
        if self.timeUnittimeUnit == 'M':
            xInterpDates = np.empty(12)
            for curMonth in range(1, 13):
                xInterpDates[curMonth-1] = cl.timegm(
                    dt.datetime(2020, curMonth, 1, 0).timetuple())
        else:
            dtBgn = dt.datetime(2020, 1, 1, 0)
            dtEnd = dt.datetime(2021, 1, 1, 0) - time.tDelta
            xInterpDates = np.linspace(
                cl.timegm(dtBgn.timetuple()),
                cl.timegm(dtEnd.timetuple()),
                int(366*24/(self.tDeltatDelta.total_seconds()/3600)))
        lkupVals = np.empty([len(xInterpDates), len(yVals[0, :])])
        if self.typtyp == 'interp':
            for i in range(len(yVals[0, :])):
                lkupVals[:, i] = np.interp(xInterpDates, xDateTime, yVals[:, i])
        else:
            for i in range(1, len(xDateTime)):
                iCurPer = (xInterpDates >= xDateTime[i-1]) & (xInterpDates < xDateTime[i])
                lkupVals[iCurPer, :] = yVals[i-1]
        self.vDatevDate = np.empty([len(xInterpDates), 3])
        for i, curDt in enumerate(xInterpDates):
            # TODO: Should avoid using datetime over pandas
            # TODO: See if there is a way to do this in pandas
            self.vDatevDate[i, 0] = dt.datetime.fromtimestamp(curDt, dt.timezone.utc).month
            self.vDatevDate[i, 1] = dt.datetime.fromtimestamp(curDt, dt.timezone.utc).day
            self.vDatevDate[i, 2] = dt.datetime.fromtimestamp(curDt, dt.timezone.utc).hour
        super().__init__(name, -1, lkupVals)
 
    def get_val(self, dateTime, col=0):
        # TODO: This should have an options for begin and end date to return multiple vals
        # TODO: They would be optional variables called dateRangeBgn and dataRangeEnd
        # TODO: Use dateTime functions here instead of vDate
        iTime = self.vDatevDate[:, 0] == dateTime.month
        if self.timeUnittimeUnit == 'D':
            iTime = iTime & (self.vDatevDate[:, 1] == dateTime.day)
        elif self.timeUnittimeUnit == 'h':
            iTime = iTime & (self.vDatevDate[:, 2] == dateTime.hour)
        rowLkup = np.where(iTime)[0][0]
        return self.lkupValslkupVals[rowLkup, col].item()