src/data_utils/loader.py
def load_data(file_path, indices=None, K=10, T=10, training=True, batch_size=1):
    """Load data for one test day, return as numpy array with normalized samples of each
        6 time steps in random order.

        Args.:
            file_path (str): file path of h5 file for one day
            indices (list): list with prediction times (as list indices in the interval [0, 288])
            K (int): the number of images look back
            T (int): the number of images look ahead

        Returns: numpy array of shape (5, 6, 3, 495, 436)
    """
    # load h5 file
    logging.info("load data: {}".format(file_path))
    fr = h5py.File(file_path, 'r')
    a_group_key = list(fr.keys())[0]
    data = fr[a_group_key]
    data = np.array(data)

    # data = [crop_image(image=i) for i in data]
    # resize image to target size
    # data = [resize_image(image=i, target=ORIGINAL_SHAPE) for i in data]

    data = [to_tiles(_salt_and_pepper_noise(crop_image(im)) / 255., TILE_SHAPE[0], TILE_SHAPE[1]) for im in data]

    # add gaussian noise
    if training:
        data = [_salt_and_pepper_noise(image=i) for i in data]
    # identify test cases and split in samples of each length 3 time bins
    data = [data[y - K: y + T] for y in indices] if indices else [data[int(y) - K: y + T] for y in
                                                                  range(K, len(data) - T - 1)]
    data = np.stack(data, axis=0)
    logging.debug("data shape: {}".format(np.shape(data)))
    data = np.transpose(data, (0, 2, 1, 3, 4, 5))  # remove channels

    print("data shape: {}".format(np.shape(data)))
    # rescale and return data
    data = data.astype(np.float32)
    # only shuffles the array along the first axis of a multi-dimensional array.
    if training:
        np.random.shuffle(data)

    minibatch = list(chunks(data, batch_size))
    print(f"length of minibatch: {len(minibatch)}")

    date = return_date(file_path)
    return minibatch, np.array([date.weekday()] * 12).reshape(1, -1)
Load data for one test day, return as numpy array with normalized samples of each
        6 time steps in random order.

        Args.:
            file_path (str): file path of h5 file for one day
            indices (list): list with prediction times (as list indices in the interval [0, 288])
            K (int): the number of images look back
            T (int): the number of images look ahead

        Returns: numpy array of shape (5, 6, 3, 495, 436)
    
Load data for one test day, return as numpy array with normalized samples of each
6 time steps in random order.




Load data for one test day return as numpy array with normalized samples of each 6 time steps in random order .
def load_data(file_path, indices=None, K=10, T=10, training=True, batch_size=1):
    logging.info("load data: {}".format(file_path))
    fr = h5py.File(file_path, 'r')
    a_group_key = list(fr.keys())[0]
    data = fr[a_group_key]
    data = np.array(data)
    data = [to_tiles(_salt_and_pepper_noise(crop_image(im)) / 255., TILE_SHAPE[0], TILE_SHAPE[1]) for im in data]
    if training:
        data = [_salt_and_pepper_noise(image=i) for i in data]
    data = [data[y - K: y + T] for y in indices] if indices else [data[int(y) - K: y + T] for y in
                                                                  range(K, len(data) - T - 1)]
    data = np.stack(data, axis=0)
    logging.debug("data shape: {}".format(np.shape(data)))
    data = np.transpose(data, (0, 2, 1, 3, 4, 5))  
    print("data shape: {}".format(np.shape(data)))
    data = data.astype(np.float32)
    if training:
        np.random.shuffle(data)
    minibatch = list(chunks(data, batch_size))
    print(f"length of minibatch: {len(minibatch)}")
    date = return_date(file_path)
    return minibatch, np.array([date.weekday()] * 12).reshape(1, -1)
def load_data ( file_path , indices = None , K = 10 , T = 10 , training = True , batch_size = 1 ) : logging . info ( "load data: {}" . format ( file_path ) ) fr = h5py . File ( file_path , 'r' ) a_group_key = list ( fr . keys ( ) ) [ 0 ] data = fr [ a_group_key ] data = np . array ( data ) data = [ to_tiles ( _salt_and_pepper_noise ( crop_image ( im ) ) / 255. , TILE_SHAPE [ 0 ] , TILE_SHAPE [ 1 ] ) for im in data ] if training : data = [ _salt_and_pepper_noise ( image = i ) for i in data ] data = [ data [ y - K : y + T ] for y in indices ] if indices else [ data [ int ( y ) - K : y + T ] for y in range ( K , len ( data ) - T - 1 ) ] data = np . stack ( data , axis = 0 ) logging . debug ( "data shape: {}" . format ( np . shape ( data ) ) ) data = np . transpose ( data , ( 0 , 2 , 1 , 3 , 4 , 5 ) ) print ( "data shape: {}" . format ( np . shape ( data ) ) ) data = data . astype ( np . float32 ) if training : np . random . shuffle ( data ) minibatch = list ( chunks ( data , batch_size ) ) print ( f"length of minibatch: {len(minibatch)}" ) date = return_date ( file_path ) return minibatch , np . array ( [ date . weekday ( ) ] * 12 ) . reshape ( 1 , - 1 )
Load data for one test day, return as numpy array with normalized samples of each
6 time steps in random order.
Load data for one test day return as numpy array with normalized samples of each 6 time steps in random order .
"""Load data for one test day, return as numpy array with normalized samples of each
        6 time steps in random order.

        Args.:
            file_path (str): file path of h5 file for one day
            indices (list): list with prediction times (as list indices in the interval [0, 288])
            K (int): the number of images look back
            T (int): the number of images look ahead

        Returns: numpy array of shape (5, 6, 3, 495, 436)
    """ # load h5 file # data = [crop_image(image=i) for i in data] # resize image to target size # data = [resize_image(image=i, target=ORIGINAL_SHAPE) for i in data] # add gaussian noise # identify test cases and split in samples of each length 3 time bins # remove channels # rescale and return data # only shuffles the array along the first axis of a multi-dimensional array.
file_path
indices
K
T
training
batch_size