time.py

Go to the documentation of this file.

# -*- coding: utf-8 -*-
"""
Created on Tue Apr 28 14:56:51 2020
 
@author: cd
"""
 
import abc
import numpy as np
from pandas import DatetimeIndex
import datetime as dt
from datetime import timedelta
from copy import copy
 
class TimeBase(metaclass=abc.ABCMeta):
    def __init__(self, name):
        self.namename = name
        self._observers_observers = []
        super().__init__()    
    # @abc.abstractmethod
    # def calc_qout(self, student):
    #     pass
    @classmethod
    def __subclasshook__(cls,C):
        if cls is TimeBase:
            attrs = set(dir(C))
            if set(cls.__abstractmethods__) <= attrs:
                return True
        return NotImplemented
    
    def bind_to(self, callback):
        self._observers_observers.append(callback)
    
        
 
# TODO: Is there a way you could allow the user to modify the start and end
# dates to allow changing sim period. There would have to be some sort of sampling
# method that is run for each input that resamples based on updated dates.
class TimeCont(TimeBase):
    def __init__(self, name, simBgn, simEnd, timeUnit='h', nHrs=1):
        # Call super class constructor
        super().__init__(name)
        # Set subclass properties
        self.simBgnsimBgn = simBgn
        self.simEndsimEnd = simEnd
        self.timeUnittimeUnit = timeUnit
        # TODO: Not sure the nHrs property is necessary
        if timeUnit == 'D':
            self.nHrsnHrs = 24
        else:
            self.nHrsnHrs = int(nHrs)
        # TODO: Look into using np.datetime64 for tDelta (needs to be compative with lookup classes)
        # self.tDelta = np.timedelta64(self.nHrs, 'h')
        self.tDeltatDelta = timedelta(hours=self.nHrsnHrs)
        # TODO: The format of this should be more readable
        # TODO: These properties could possibly be private
        self.dateTimedateTime = DatetimeIndex(
            np.arange(np.datetime64(self.simBgnsimBgn),
                      np.datetime64(self.simEndsimEnd)+np.timedelta64(self.nHrsnHrs, 'h'), 
            np.timedelta64(self.nHrsnHrs, 'h'), dtype='datetime64[h]').astype('datetime64[h]'))
        self.nStepsnSteps = self.dateTimedateTime.size
        self.endend = self.nStepsnSteps - 1
        self.stepssteps = np.arange(self.nStepsnSteps, dtype=int)
        self._step_step = self.stepssteps[0]
        self._curDT_curDT = self.dateTimedateTime[0]
 
    @property
    def step(self):
        return self._step_step
 
    @step.setter        
    def step(self, i):
        self._step_step = self.stepssteps[i].item()
        self._curDT_curDT = copy(self.dateTimedateTime[i])
        for callback in self._observers_observers:
            callback(self._step_step, self._curDT_curDT)
 
    @property
    def curDT(self):
        return self._curDT_curDT
 
    def get_datetime_offset(self, tsOffset):
        return self.dateTimedateTime[min(self.nStepsnSteps-1, self.stepstepstep + tsOffset)]
 
    def get_period(self, dateTimeBgn, dateTimeEnd):
        pass
        # Matlab code
        # iPer = this.nDate > nDateBgn - this.roundFactor ...
        #         & this.nDate < nDateEnd + this.roundFactor;
        
    def get_vdate(self):
        vDate = np.full((len(self.dateTime), 4), np.nan)
        # TODO: I think you can remove this for loop by just concatenating arrays vertically
        for i, curDt in enumerate(self.dateTime):
            vDate[i, :] = [curDt.year, curDt.month, curDt.day, curDt.hour]
        return vDate
    
    def get_ndate(self):
        return self.dateTimedateTime.astype(np.int64)
        
        
class TimeFcst(TimeBase):
    def __init__(self, name, Tcont, fcstFreq=24, fcstHoriz=360, tFcst=None, fcstDate=None):
        # Call super class constructor
        super().__init__(name)
        # TODO: Maybe build a new Tcont instead of having it passed
        self.TcontTcont = Tcont
        self.nHrsnHrs = Tcont.nHrs
        self.timeUnittimeUnit = Tcont.timeUnit
        self.tDeltatDelta = Tcont.tDelta
        # self.nSteps = self.Tcont.nSteps
        self.nStepsnSteps = int(fcstHoriz/self.TcontTcont.nHrs + 1)
        self.endend = self.nStepsnSteps - 1
        self.stepssteps = np.arange(self.nStepsnSteps)
        self.fcstFreqfcstFreq = int(fcstFreq)
        self.nFcstsnFcsts = int(np.ceil(self.TcontTcont.nSteps*self.nHrsnHrs/self.fcstFreqfcstFreq))
        self._rowCurFcst_rowCurFcst = 0
        self.TcontTcont.bind_to(self.update_current_tFcstupdate_current_tFcst)
        tDeltaFcst = np.timedelta64(self.fcstFreqfcstFreq, 'h')
        if fcstDate is None:
            # tDeltaStep = np.timedelta64(self.nHrs, 'h')
            # self.fcstDate = DatetimeIndex(
            #     np.arange(np.datetime64(self.Tcont.simBgn) + tDeltaStep,
            #               np.datetime64(self.Tcont.simEnd) + tDeltaStep + tDeltaFcst, 
            #     tDeltaFcst, dtype='datetime64[h]').astype('datetime64[h]'))
            self.fcstDatefcstDate = DatetimeIndex(
                np.arange(np.datetime64(self.TcontTcont.simBgn),
                          np.datetime64(self.TcontTcont.simEnd) + (tDeltaFcst-1), 
                tDeltaFcst, dtype='datetime64[h]').astype('datetime64[h]'))
            # self.fcstDate = DatetimeIndex(
            #     np.arange(np.datetime64(self.Tcont.simBgn),
            #               np.datetime64(self.Tcont.simEnd), 
            #     tDeltaFcst, dtype='datetime64[h]').astype('datetime64[h]'))
        if tFcst is None:
            tFcst = np.empty(self.nFcstsnFcsts, dtype=TimeCont)
            # TODO: This loop assumes that each step is a day. Should be user specified
            for i in range(0, self.nFcstsnFcsts):
                # tBgn = np.datetime64(Tcont.dateTime[i]+np.timedelta64(self.nHrs, 'h'))
                # tBgn = Tcont.dateTime[max(0, i-1)]
                idxTcontBgn = max(0, np.where(Tcont.dateTime==self.fcstDatefcstDate[i])[0][0] - 1)
                tBgn = Tcont.dateTime[idxTcontBgn]
                tEnd = np.datetime64(tBgn+np.timedelta64(fcstHoriz, 'h'))
                # Alternative using Pandas:
                # tEnd = Timestamp(Tcont.dateTime[i]+Timedelta(fcstHoriz, unit='h'))
                tFcst[i] = TimeCont('timeFcst'+str(i), tBgn, tEnd, Tcont.timeUnit)
            self.tFcsttFcst = tFcst
        self._tFcstCur_tFcstCur = self.tFcsttFcst[0]
 
    @property
    def step(self):
        return self._step_step
 
    @step.setter
    def step(self, i):
        self._step_step = self._tFcstCur_tFcstCur.steps[i].item()
        self._curDT_curDT = copy(self._tFcstCur_tFcstCur.dateTime[i])
        self.get_fcst_timeget_fcst_time().step = self.stepstepstep
 
    @property
    def curDT(self):
        return self._curDT_curDT
    
    def update_current_tFcst(self, step, dateTime):
        chkCurRow = np.where(self.fcstDatefcstDate<=dateTime)[0][-1]
        if chkCurRow != self._rowCurFcst_rowCurFcst:
            self._rowCurFcst_rowCurFcst = chkCurRow
            self._tFcstCur_tFcstCur = copy(self.tFcsttFcst[self._rowCurFcst_rowCurFcst])
            for callback in self._observers_observers:
                callback(self._rowCurFcst_rowCurFcst, dateTime)
                  
        
    def get_fcst_time(self):
        return self._tFcstCur_tFcstCur
                
    def get_fcst_step(self):
        return np.where(self._tFcstCur_tFcstCur.dateTime==self.TcontTcont.curDT)[0][0]
    
    def get_datetime_offset(self, tsOffset):
        return self._tFcstCur_tFcstCur.get_datetime_offset(tsOffset)
    
    # def check_fcst_date(self):
    #     return np.any(self.fcstDate == self.Tcont.curDT)