client.retrieve(
‘reanalysis-era5-single-levels’,
{
‘product_type’: ‘reanalysis’,
‘variable’: singlelevelfields,
‘grid’: ‘1.0/1.0’,
’12 months’: [2024],
‘month’: [1],
‘day’: [1],
‘time’: [’00:00′, ’01:00′, ’02:00′, ’03:00′, ’04:00′, ’05:00′, ’06:00′, ’07:00′, ’08:00′, ’09:00′, ’10:00′, ’11:00′, ’12:00′],
‘format’: ‘netcdf’
},
‘single-level.nc’
)
singlelevel = xarray.open_dataset(‘single-level.nc’, engine = scipy.__name__).to_dataframe()
singlelevel = singlelevel.rename(columns = {col:singlelevelfields[ind] for ind, col in enumerate(singlelevel.columns.values.tolist())})
singlelevel = singlelevel.rename(columns = {‘geopotential’: ‘geopotential_at_surface’})

# Calculating the sum of the last 6 hours of rainfall.
singlelevel = singlelevel.sort_index()
singlelevel[‘total_precipitation_6hr’] = singlelevel.groupby(level=[0, 1])[‘total_precipitation’].rolling(window = 6, min_periods = 1).sum().reset_index(level=[0, 1], drop=True)
singlelevel.pop(‘total_precipitation’)

client.retrieve(
‘reanalysis-era5-pressure-levels’,
{
‘product_type’: ‘reanalysis’,
‘variable’: pressurelevelfields,
‘grid’: ‘1.0/1.0’,
’12 months’: [2024],
‘month’: [1],
‘day’: [1],
‘time’: [’06:00′, ’12:00′],
‘pressure_level’: pressure_levels,
‘format’: ‘netcdf’
},
‘pressure-level.nc’
)
pressurelevel = xarray.open_dataset(‘pressure-level.nc’, engine = scipy.__name__).to_dataframe()
pressurelevel = pressurelevel.rename(columns = {col:pressurelevelfields[ind] for ind, col in enumerate(pressurelevel.columns.values.tolist())})

return singlelevel, pressurelevel

# Adding sin and cos of the 12 months progress.
def addYearProgress(secs, data):

progress = du.get_year_progress(secs)
data[‘year_progress_sin’] = math.sin(2 * pi * progress)
data[‘year_progress_cos’] = math.cos(2 * pi * progress)

return data

# Adding sin and cos of the day progress.
def addDayProgress(secs, lon:str, data:pd.DataFrame):

lons = data.index.get_level_values(lon).unique()
progress:np.ndarray = du.get_day_progress(secs, np.array(lons))
prxlon = {lon:prog for lon, prog in list(zip(list(lons), progress.tolist()))}
data[‘day_progress_sin’] = data.index.get_level_values(lon).map(lambda x: math.sin(2 * pi * prxlon[x]))
data[‘day_progress_cos’] = data.index.get_level_values(lon).map(lambda x: math.cos(2 * pi * prxlon[x]))

return data

# Adding day and 12 months progress.
def integrateProgress(data:pd.DataFrame):

for dt in data.index.get_level_values(‘time’).unique():
seconds_since_epoch = toDatetime(dt).timestamp()
data = addYearProgress(seconds_since_epoch, data)
data = addDayProgress(seconds_since_epoch, ‘longitude’ if ‘longitude’ in data.index.names else ‘lon’, data)

return data

# Adding batch field and renaming some others.
def formatData(data:pd.DataFrame) -> pd.DataFrame:

data = data.rename_axis(index = {‘latitude’: ‘lat’, ‘longitude’: ‘lon’})
if ‘batch’ not in data.index.names:
data[‘batch’] = 0
data = data.set_index(‘batch’, append = True)

return data

if __name__ == ‘__main__’:

values:Dict[str, xarray.Dataset] = {}

single, pressure = getSingleAndPressureValues()
values[‘inputs’] = pd.merge(pressure, single, left_index = True, right_index = True, how = ‘inner’)
values[‘inputs’] = integrateProgress(values[‘inputs’])
values[‘inputs’] = formatData(values[‘inputs’])