From 436a3d3eaebcca11d8da6d8b1bb4f6f543e135e4 Mon Sep 17 00:00:00 2001 From: Jonas Gunz Date: Wed, 18 Oct 2023 23:55:21 +0200 Subject: implemented aggreator --- aggregator/dwd_icon.py | 126 ++++++++++++++++++++++++++++++++++++++++++ config.yaml | 24 +++++++- download.py | 120 ---------------------------------------- plotter/horizontal.py | 4 +- plotter/vertical_from_grib.py | 4 +- run.py | 13 +++++ 6 files changed, 163 insertions(+), 128 deletions(-) create mode 100755 aggregator/dwd_icon.py delete mode 100755 download.py diff --git a/aggregator/dwd_icon.py b/aggregator/dwd_icon.py new file mode 100755 index 0000000..60c8433 --- /dev/null +++ b/aggregator/dwd_icon.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 + +import requests +import datetime +import pytz +import requests +import os + +from multiprocessing import cpu_count +from multiprocessing.pool import ThreadPool + +import subprocess + +import xarray as xr + +import misc + +BASE='https://opendata.dwd.de/weather/nwp' + +def get_current_run(): + # we allow up to 3h of slack for DWD to upload the latest run + tz = pytz.timezone('UTC') + now = datetime.datetime.now(datetime.timezone.utc) + corrected = now - datetime.timedelta(hours=3) + + run = int(corrected.hour / 6) * 6 + + return (f'{run:02d}', corrected.strftime('%Y%m%d')) + +def download_url(args): + url, dest = args + r = requests.get(url) + try: + with open(dest, 'wb') as f: + f.write(r.content) + print(f'Downloaded {dest}') + except Exception as e: + print(f'Failed to download {dest}:\n', e) + +def unpack_bz2(dest): + res = subprocess.run(['bzip2', '-df', dest]) + if res.returncode != 0: + print(f'There was an error unpacking {dest}:', res.stderr) + +def download_dwd_gribs(config, date, run, target): + model = config['model'] + model_long = config['model_long'] + + misc.create_output_dir(config['output']) + + to_download = [] + + for step in config['steps']: + step_str = f'{step:03d}' + + for parameter in config['pressure_level_parameters']: + parameter2 = parameter.upper() if config['parameter_caps_in_filename'] else parameter + + for level in config['pressure_levels']: + filename = f'{model_long}_regular-lat-lon_pressure-level_{date}{run}_{step_str}_{level}_{parameter2}.grib2.bz2' + URL = f'{BASE}/{model}/grib/{run}/{parameter}/{filename}' + + to_download.append((URL, os.path.join(config['output'], filename))) + + for parameter in config['single_level_parameters']: + parameter2 = parameter.upper() if config['parameter_caps_in_filename'] else parameter + filename = f'{model_long}_regular-lat-lon_single-level_{date}{run}_{step_str}_{parameter2}.grib2.bz2' + URL = f'{BASE}/{model}/grib/{run}/{parameter}/{filename}' + + to_download.append((URL, os.path.join(config['output'], filename))) + + + for _ in ThreadPool(cpu_count()).imap_unordered(download_url, to_download): + pass + + print('Done Downloading. Uncompressing...') + + for _ in ThreadPool(cpu_count()).imap_unordered(unpack_bz2, [dest for _, dest in to_download]): + pass + + downloaded_gribs = [dest.removesuffix('.bz2') for _, dest in to_download] + + res = subprocess.run(['grib_copy'] + downloaded_gribs + [target]) + if res.returncode != 0: + print('grib_copy failed with: ', res.stderr) + + res = subprocess.run(['rm', '-f'] + downloaded_gribs) + if res.returncode != 0: + print('rm failed with: ', res.stderr) + +def load_data(name, config): + run, date = get_current_run() + target = os.path.join(config['output'], f'{name}_{date}_{run}.grib2') + + if not os.path.exists(target): + download_dwd_gribs(config, date, run, target) + else: + print(f'{target} alreasy exists. Using the cached version.') + + return xr.load_dataset(target, engine='cfgrib') + +config = { + 'output':'dwd_icon-eu', + 'model':'icon-eu', + 'model_long':'icon-eu_europe', + 'parameter_caps_in_filename':True, + 'pressure_level_parameters': [ + 't', + 'relhum', + 'u', + 'v', + 'fi', + 'clc' + ], + 'single_level_parameters': [ + 'pmsl', + 't_2m', + 'relhum_2m' + ], + 'pressure_levels':[ 1000, 950, 925, 900, 875, 850, 825, 800, 775, 700, 600, 500, 400, 300, 250, 200, 150, 100 ], + 'steps':[0, 3, 6, 9, 12, 15, 18, 21, 24, 27, 30, 33, 36, 39, 42, 45, 48] +} + +if __name__ == '__main__': + load_data('test_icon_eu', config) + diff --git a/config.yaml b/config.yaml index 8a882ca..d849625 100644 --- a/config.yaml +++ b/config.yaml @@ -1,8 +1,28 @@ --- index: web/data/index.json +aggregator: + icon_eu: + module: aggregator.dwd_icon + output: dwd_icon_cache + model: icon-eu + model_long: icon-eu_europe + parameter_caps_in_filename: true + pressure_level_parameters: + - t + - relhum + - u + - v + - fi + - clc + single_level_parameters: + - pmsl + - t_2m + - relhum_2m + pressure_levels: [1000, 950, 925, 900, 875, 850, 825, 800, 775, 700, 600, 500, 400, 300, 250, 200, 150, 100] + steps: [0, 3, 6, 9, 12, 15, 18, 21, 24, 27, 30, 33, 36, 39, 42, 45, 48] plotter: - module: 'plotter.vertical_from_grib' - source: dwd_icon-eu/combined.grib2 + aggregator: icon_eu output: web/data plots: - lat: 47.96 @@ -10,7 +30,7 @@ plotter: name: Antersberg analysis: lcl - module: 'plotter.horizontal' - source: dwd_icon-eu/combined.grib2 + aggregator: icon_eu output: web/data plots: - name: t_fi_850 diff --git a/download.py b/download.py deleted file mode 100755 index fdaa3f4..0000000 --- a/download.py +++ /dev/null @@ -1,120 +0,0 @@ -#!/usr/bin/env python3 - -import requests -import datetime -import pytz -import requests -import os - -from multiprocessing import cpu_count -from multiprocessing.pool import ThreadPool - -import subprocess - -import misc - -BASE='https://opendata.dwd.de/weather/nwp' - -def get_current_run(): - # we allow up to 3h of slack for DWD to upload the latest run - tz = pytz.timezone('UTC') - now = datetime.datetime.now(datetime.timezone.utc) - corrected = now - datetime.timedelta(hours=3) - - run = int(corrected.hour / 6) * 6 - - return (f'{run:02d}', corrected.strftime('%Y%m%d')) - -def download_url(args): - url, dest = args - r = requests.get(url) - try: - with open(dest, 'wb') as f: - f.write(r.content) - print(f'Downloaded {dest}') - except Exception as e: - print(f'Failed to download {dest}:\n', e) - -def unpack_bz2(dest): - res = subprocess.run(['bzip2', '-df', dest]) - if res.returncode != 0: - print(f'There was an error unpacking {dest}:', res.stderr) - -def download_dwd_gribs(config, date, run, target): - model = config['model'] - model_long = config['model_long'] - - misc.create_output_dir(config['output']) - - to_download = [] - - for step in config['steps']: - step_str = f'{step:03d}' - - for parameter in config['pressure_level_parameters']: - parameter2 = parameter.upper() if config['parameter_caps_in_filename'] else parameter - - for level in config['pressure_levels']: - filename = f'{model_long}_regular-lat-lon_pressure-level_{date}{run}_{step_str}_{level}_{parameter2}.grib2.bz2' - URL = f'{BASE}/{model}/grib/{run}/{parameter}/{filename}' - - to_download.append((URL, os.path.join(config['output'], filename))) - - for parameter in config['single_level_parameters']: - parameter2 = parameter.upper() if config['parameter_caps_in_filename'] else parameter - filename = f'{model_long}_regular-lat-lon_single-level_{date}{run}_{step_str}_{parameter2}.grib2.bz2' - URL = f'{BASE}/{model}/grib/{run}/{parameter}/{filename}' - - to_download.append((URL, os.path.join(config['output'], filename))) - - - for _ in ThreadPool(cpu_count()).imap_unordered(download_url, to_download): - pass - - print('Done Downloading. Uncompressing...') - - for _ in ThreadPool(cpu_count()).imap_unordered(unpack_bz2, [dest for _, dest in to_download]): - pass - - downloaded_gribs = [dest.removesuffix('.bz2') for _, dest in to_download] - - res = subprocess.run(['grib_copy'] + downloaded_gribs + [target]) - if res.returncode != 0: - print('grib_copy failed with: ', res.stderr) - - res = subprocess.run(['rm', '-f'] + downloaded_gribs) - if res.returncode != 0: - print('rm failed with: ', res.stderr) - -def load_data(config): - run, date = get_current_run() - target = os.path.join(config['output'], f'combined_{date}_{run}.grib2') - - if not os.path.exists(target): - download_dwd_gribs(config, date, run, target) - -config = { - 'output':'dwd_icon-eu', - 'model':'icon-eu', - 'model_long':'icon-eu_europe', - 'parameter_caps_in_filename':True, - 'pressure_level_parameters': [ - 't', - 'relhum', - 'u', - 'v', - 'fi', - 'clc' - ], - 'single_level_parameters': [ - 'pmsl', - 't_2m', - 'relhum_2m' - ], - 'pressure_levels':[ 1000, 950, 925, 900, 875, 850, 825, 800, 775, 700, 600, 500, 400, 300, 250, 200, 150, 100 ], - 'steps':[0, 3, 6, 9, 12, 15, 18, 21, 24, 27, 30, 33, 36, 39, 42, 45, 48] -} - -if __name__ == '__main__': - load_data(config) - diff --git a/plotter/horizontal.py b/plotter/horizontal.py index e83b683..fa5f9df 100755 --- a/plotter/horizontal.py +++ b/plotter/horizontal.py @@ -33,10 +33,8 @@ config = { ] } -def run(source, plots, output='.'): +def run(data, plots, output='.'): misc.create_output_dir(output) - data = xr.load_dataset(source, engine='cfgrib') - index = [] for plot in plots: diff --git a/plotter/vertical_from_grib.py b/plotter/vertical_from_grib.py index 929782e..9f927a7 100755 --- a/plotter/vertical_from_grib.py +++ b/plotter/vertical_from_grib.py @@ -25,10 +25,8 @@ config = { ] } -def run(source, plots, output='.'): +def run(data, plots, output='.'): misc.create_output_dir(output) - data = xr.load_dataset(source, engine='cfgrib') - index = [] for plot in plots: diff --git a/run.py b/run.py index de6a078..fa0f598 100755 --- a/run.py +++ b/run.py @@ -26,6 +26,19 @@ for plotter in conf['plotter']: modname = plotter['module'] del plotter['module'] + if 'aggregator' in plotter: + aggname = plotter['aggregator'] + del plotter['aggregator'] + aggconf = conf['aggregator'][aggname] + classname = aggconf['module'] + # We should prbly not delete it like in the plotter, since it is not a deepcopy + # and may be used again. + + agg = __import__(classname, fromlist=[None]) + + # TODO: figure out a way to use **aggconf instead. + plotter['data'] = agg.load_data(aggname, aggconf) + mod = __import__(modname, fromlist=[None]) index.extend(mod.run(**plotter)) -- cgit v1.2.3