diff --git a/CHANGELOG.md b/CHANGELOG.md
index 82163eced04bc50d1ebb352bff20ca23dda55711..b11a169c854465c5ea932f00f5da5a1688df7c18 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,40 @@
# Changelog
All notable changes to this project will be documented in this file.
+## v1.4.0 - 2021-07-27 - <release description>
+
+### general:
+* many technical adjustments to improve usability and transparency of MLAir
+* new FCN and CNN classes for easy NN model creation
+* new plots
+
+### new features:
+* new FCN class that can be customized in many ways (#284)
+* also new CNN class (#289)
+* added new bootstrap analysis method: mean bootstrapping (#300)
+* new data handler using FIR filters (#306)
+* performance measures are now stored in local files (#286)
+* histogram plots for inputs and targets (#299)
+* periodogram plots for filtered data (#298)
+
+### technical:
+* a calling run script can be stored inside experiment folder if reference to this script is parsed as argument (#99)
+* new callback to track epoch-runtime (#312)
+* added switch to use multiprocessing (#297)
+* customize maximum number of parallel processes (#308)
+* support non-monotonic window lead times (#313)
+* resolved bug with FileExistsError (#311)
+* resolved bug if no chemical is used at all (#307)
+* min/max scaler now scales between -1 and 1 (#302)
+* added missing offset parameter to some data handlers (#305)
+* improved data store logging (#304)
+* improved logging message on station removal in preprocessing (#294)
+* limited number of retries in JOIN module (#296)
+* adjusted competing skill score plot (#301)
+* transformation parameter check (#295)
+* implemented lazy data preprocessing for selected data handlers (#292)
+* fix bug in separation of scales data handler (#290)
+
## v1.3.0 - 2021-02-24 - competitors and improved transformation
### general:
diff --git a/README.md b/README.md
index cbaa61d2632e46ceb44d39bfeceef7423d1b9784..0e1df0561d15b743a85b0981b552a1444b6cc38c 100644
--- a/README.md
+++ b/README.md
@@ -31,7 +31,7 @@ HPC systems, see [here](#special-instructions-for-installation-on-jülich-hpc-sy
* Installation of **MLAir**:
* Either clone MLAir from the [gitlab repository](https://gitlab.version.fz-juelich.de/toar/mlair.git)
and use it without installation (beside the requirements)
- * or download the distribution file ([current version](https://gitlab.version.fz-juelich.de/toar/mlair/-/blob/master/dist/mlair-1.3.0-py3-none-any.whl))
+ * or download the distribution file ([current version](https://gitlab.version.fz-juelich.de/toar/mlair/-/blob/master/dist/mlair-1.4.0-py3-none-any.whl))
and install it via `pip install <dist_file>.whl`. In this case, you can simply import MLAir in any python script
inside your virtual environment using `import mlair`.
* (tf) Currently, TensorFlow-1.13 is mentioned in the requirements. We already tested the TensorFlow-1.15 version and couldn't
@@ -518,3 +518,22 @@ add it to `src/join_settings.py` in the hourly data section. Replace the `TOAR_S
value. To make sure, that this **sensitive** data is not uploaded to the remote server, use the following command to
prevent git from tracking this file: `git update-index --assume-unchanged src/join_settings.py`
+
+## Known Issues
+
+### Problem with multiprocessing
+
+* cpython and python's native multiprocessing can crash when using the multiprocessing approach for preprocessing. This
+is caused by an internal limitation in order of 2GB. When using long periods and therefore very big data,
+multiprocessing is not able to handle these data correctly:
+```shell
+File "mlair/mlair/run_modules/pre_processing.py", line X, in validate_station
+ dh, s = p.get()
+File "multiprocessing/pool.py", line 644, in get
+ raise self._value
+multiprocessing.pool.MaybeEncodingError: Error sending result: '(DEMV012, 'DEMV012')'. Reason: 'error("'i' format requires -2147483648 <= number <= 2147483647",)'
+```
+* to solve this issue, either update your python version to >=3.8 (warning, this version is not tested with MLAir) or
+apply the patch that is applied in this commit
+https://github.com/python/cpython/commit/bccacd19fa7b56dcf2fbfab15992b6b94ab6666b or as proposed in this comment
+https://stackoverflow.com/questions/47776486/python-struct-error-i-format-requires-2147483648-number-2147483647
diff --git a/dist/mlair-1.4.0-py3-none-any.whl b/dist/mlair-1.4.0-py3-none-any.whl
new file mode 100644
index 0000000000000000000000000000000000000000..2b546f8a20d60c1de6128234ad6837314a391177
Binary files /dev/null and b/dist/mlair-1.4.0-py3-none-any.whl differ
diff --git a/docs/_source/installation.rst b/docs/_source/installation.rst
index 20db920216b2f0cda7568e7a153a6176d441e995..27543ac109439e487756cc211ecc47be946c586c 100644
--- a/docs/_source/installation.rst
+++ b/docs/_source/installation.rst
@@ -26,7 +26,7 @@ Installation of MLAir
* Install all requirements from `requirements.txt <https://gitlab.version.fz-juelich.de/toar/machinelearningtools/-/blob/master/requirements.txt>`_
preferably in a virtual environment
* Either clone MLAir from the `gitlab repository <https://gitlab.version.fz-juelich.de/toar/machinelearningtools.git>`_
-* or download the distribution file (`current version <https://gitlab.version.fz-juelich.de/toar/mlair/-/blob/master/dist/mlair-1.3.0-py3-none-any.whl>`_)
+* or download the distribution file (`current version <https://gitlab.version.fz-juelich.de/toar/mlair/-/blob/master/dist/mlair-1.4.0-py3-none-any.whl>`_)
and install it via :py:`pip install <dist_file>.whl`. In this case, you can simply
import MLAir in any python script inside your virtual environment using :py:`import mlair`.
* (tf) Currently, TensorFlow-1.13 is mentioned in the requirements. We already tested the TensorFlow-1.15 version and couldn't
diff --git a/mlair/__init__.py b/mlair/__init__.py
index 05e8d504fbf171b8889343161252bbd439e52473..f760f9b0fa4b87bde1f6ee409626f4428083d895 100644
--- a/mlair/__init__.py
+++ b/mlair/__init__.py
@@ -1,6 +1,6 @@
__version_info__ = {
'major': 1,
- 'minor': 3,
+ 'minor': 4,
'micro': 0,
}
diff --git a/mlair/configuration/defaults.py b/mlair/configuration/defaults.py
index f751ac24b4162eff5e34218d2fe983e7623df251..7c606ebf96c5bf00aacad04efa533504576c9bc1 100644
--- a/mlair/configuration/defaults.py
+++ b/mlair/configuration/defaults.py
@@ -13,6 +13,7 @@ DEFAULT_START = "1997-01-01"
DEFAULT_END = "2017-12-31"
DEFAULT_WINDOW_HISTORY_SIZE = 13
DEFAULT_OVERWRITE_LOCAL_DATA = False
+DEFAULT_OVERWRITE_LAZY_DATA = False
DEFAULT_HPC_LOGIN_LIST = ["ju", "hdfmll"] # ju[wels} #hdfmll(ogin)
DEFAULT_HPC_HOST_LIST = ["jw", "hdfmlc"] # first part of node names for Juwels (jw[comp], hdfmlc(ompute).
DEFAULT_CREATE_NEW_MODEL = True
diff --git a/mlair/configuration/path_config.py b/mlair/configuration/path_config.py
index e7418b984dab74b0527b8dca05a9f6c3636ac18f..6b9c799ceb190b9150be3a4cfcd336eaf45aa768 100644
--- a/mlair/configuration/path_config.py
+++ b/mlair/configuration/path_config.py
@@ -3,6 +3,7 @@ import getpass
import logging
import os
import re
+import shutil
import socket
from typing import Union
@@ -112,17 +113,23 @@ def set_bootstrap_path(bootstrap_path: str, data_path: str) -> str:
return os.path.abspath(bootstrap_path)
-def check_path_and_create(path: str) -> None:
+def check_path_and_create(path: str, remove_existing: bool = False) -> None:
"""
Check a given path and create if not existing.
:param path: path to check and create
+ :param remove_existing: if set to true an existing folder is removed and replaced by a new one (default False).
"""
try:
os.makedirs(path)
logging.debug(f"Created path: {path}")
except FileExistsError:
- logging.debug(f"Path already exists: {path}")
+ if remove_existing is True:
+ logging.debug(f"Remove / clean path: {path}")
+ shutil.rmtree(path)
+ check_path_and_create(path, remove_existing=False)
+ else:
+ logging.debug(f"Path already exists: {path}")
def get_host():
diff --git a/mlair/data_handler/data_handler_mixed_sampling.py b/mlair/data_handler/data_handler_mixed_sampling.py
index 8205ae6c28f3683b1052c292e5d063d8bca555dc..5aefb0368ec1cf544443bb5e0412dd16a97f2a7f 100644
--- a/mlair/data_handler/data_handler_mixed_sampling.py
+++ b/mlair/data_handler/data_handler_mixed_sampling.py
@@ -12,6 +12,7 @@ from mlair.helpers import remove_items
from mlair.configuration.defaults import DEFAULT_SAMPLING, DEFAULT_INTERPOLATION_LIMIT, DEFAULT_INTERPOLATION_METHOD
from mlair.helpers.filter import filter_width_kzf
+import copy
import inspect
from typing import Callable
import datetime as dt
@@ -140,13 +141,6 @@ class DataHandlerMixedSamplingWithFilterSingleStation(DataHandlerMixedSamplingSi
def load_and_interpolate(self, ind) -> [xr.DataArray, pd.DataFrame]:
start, end = self.update_start_end(ind)
- # if ind == 0: # for inputs
- # estimated_filter_width = self.estimate_filter_width()
- # start = self._add_time_delta(self.start, -estimated_filter_width)
- # end = self._add_time_delta(self.end, estimated_filter_width)
- # else: # target
- # start, end = self.start, self.end
-
vars = [self.variables, self.target_var]
stats_per_var = helpers.select_from_dict(self.statistics_per_var, vars[ind])
@@ -264,7 +258,83 @@ class DataHandlerMixedSamplingWithClimateFirFilter(DataHandlerClimateFirFilter):
data_handler = DataHandlerMixedSamplingWithClimateFirFilterSingleStation
data_handler_transformation = DataHandlerMixedSamplingWithClimateFirFilterSingleStation
- _requirements = data_handler.requirements()
+ data_handler_unfiltered = DataHandlerMixedSamplingSingleStation
+ _requirements = list(set(data_handler.requirements() + data_handler_unfiltered.requirements()))
+ DEFAULT_FILTER_ADD_UNFILTERED = False
+
+ def __init__(self, *args, data_handler_class_unfiltered: data_handler_unfiltered = None,
+ filter_add_unfiltered: bool = DEFAULT_FILTER_ADD_UNFILTERED, **kwargs):
+ self.dh_unfiltered = data_handler_class_unfiltered
+ self.filter_add_unfiltered = filter_add_unfiltered
+ super().__init__(*args, **kwargs)
+
+ @classmethod
+ def own_args(cls, *args):
+ """Return all arguments (including kwonlyargs)."""
+ super_own_args = DataHandlerClimateFirFilter.own_args(*args)
+ arg_spec = inspect.getfullargspec(cls)
+ list_of_args = arg_spec.args + arg_spec.kwonlyargs + super_own_args
+ return remove_items(list_of_args, ["self"] + list(args))
+
+ def _create_collection(self):
+ if self.filter_add_unfiltered is True and self.dh_unfiltered is not None:
+ return [self.id_class, self.dh_unfiltered]
+ else:
+ return super()._create_collection()
+
+ @classmethod
+ def build(cls, station: str, **kwargs):
+ sp_keys = {k: copy.deepcopy(kwargs[k]) for k in cls.data_handler.requirements() if k in kwargs}
+ filter_add_unfiltered = kwargs.get("filter_add_unfiltered", False)
+ sp_keys = cls.build_update_kwargs(sp_keys, dh_type="filtered")
+ sp = cls.data_handler(station, **sp_keys)
+ if filter_add_unfiltered is True:
+ sp_keys = {k: copy.deepcopy(kwargs[k]) for k in cls.data_handler_unfiltered.requirements() if k in kwargs}
+ sp_keys = cls.build_update_kwargs(sp_keys, dh_type="unfiltered")
+ sp_unfiltered = cls.data_handler_unfiltered(station, **sp_keys)
+ else:
+ sp_unfiltered = None
+ dp_args = {k: copy.deepcopy(kwargs[k]) for k in cls.own_args("id_class") if k in kwargs}
+ return cls(sp, data_handler_class_unfiltered=sp_unfiltered, **dp_args)
+
+ @classmethod
+ def build_update_kwargs(cls, kwargs_dict, dh_type="filtered"):
+ if "transformation" in kwargs_dict:
+ trafo_opts = kwargs_dict.get("transformation")
+ if isinstance(trafo_opts, dict):
+ kwargs_dict["transformation"] = trafo_opts.get(dh_type)
+ return kwargs_dict
+
+ @classmethod
+ def transformation(cls, set_stations, tmp_path=None, **kwargs):
+
+ sp_keys = {k: copy.deepcopy(kwargs[k]) for k in cls._requirements if k in kwargs}
+ if "transformation" not in sp_keys.keys():
+ return
+
+ transformation_filtered = super().transformation(set_stations, tmp_path=tmp_path,
+ dh_transformation=cls.data_handler_transformation, **kwargs)
+ if kwargs.get("filter_add_unfiltered", False) is False:
+ return transformation_filtered
+ else:
+ transformation_unfiltered = super().transformation(set_stations, tmp_path=tmp_path,
+ dh_transformation=cls.data_handler_unfiltered, **kwargs)
+ return {"filtered": transformation_filtered, "unfiltered": transformation_unfiltered}
+
+ def get_X_original(self):
+ if self.use_filter_branches is True:
+ X = []
+ for data in self._collection:
+ if hasattr(data, "filter_dim"):
+ X_total = data.get_X()
+ filter_dim = data.filter_dim
+ for filter_name in data.filter_dim_order:
+ X.append(X_total.sel({filter_dim: filter_name}, drop=True))
+ else:
+ X.append(data.get_X())
+ return X
+ else:
+ return super().get_X_original()
class DataHandlerSeparationOfScalesSingleStation(DataHandlerMixedSamplingWithKzFilterSingleStation):
diff --git a/mlair/data_handler/data_handler_neighbors.py b/mlair/data_handler/data_handler_neighbors.py
index 6c87946eaad5568e1ff59c3988bf8fe469442641..26fe6f8549639019169bf0b894186d354eaddc83 100644
--- a/mlair/data_handler/data_handler_neighbors.py
+++ b/mlair/data_handler/data_handler_neighbors.py
@@ -1,6 +1,10 @@
__author__ = 'Lukas Leufen'
__date__ = '2020-07-17'
+"""
+WARNING: This data handler is just a prototype and has not been validated to work properly! Use it with caution!
+"""
+
import datetime as dt
import numpy as np
@@ -19,7 +23,7 @@ number = Union[float, int]
num_or_list = Union[number, List[number]]
-class DataHandlerNeighbors(DefaultDataHandler):
+class DataHandlerNeighbors(DefaultDataHandler): # pragma: no cover
"""Data handler including neighboring stations."""
def __init__(self, id_class, data_path, neighbors=None, min_length=0,
@@ -48,7 +52,7 @@ class DataHandlerNeighbors(DefaultDataHandler):
return [super(DataHandlerNeighbors, self).get_coordinates()].append(neighbors)
-def run_data_prep():
+def run_data_prep(): # pragma: no cover
"""Comment: methods just to start write meaningful test routines."""
data = DummyDataHandler("main_class")
data.get_X()
@@ -63,7 +67,7 @@ def run_data_prep():
data_prep.get_data(upsampling=False)
-def create_data_prep():
+def create_data_prep(): # pragma: no cover
"""Comment: methods just to start write meaningful test routines."""
path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "testdata")
station_type = None
@@ -91,7 +95,7 @@ def create_data_prep():
return data_prep
-class DummyDataHandler(AbstractDataHandler):
+class DummyDataHandler(AbstractDataHandler): # pragma: no cover
def __init__(self, name, number_of_samples=None):
"""This data handler takes a name argument and the number of samples to generate. If not provided, a random
diff --git a/mlair/data_handler/data_handler_single_station.py b/mlair/data_handler/data_handler_single_station.py
index 3b9f36553fd16557326ffa350f729c901584682c..628e6b44aef975c67c468d1f1c27c6588bc701b6 100644
--- a/mlair/data_handler/data_handler_single_station.py
+++ b/mlair/data_handler/data_handler_single_station.py
@@ -5,6 +5,8 @@ __date__ = '2020-07-20'
import copy
import datetime as dt
+import gc
+
import dill
import hashlib
import logging
@@ -61,7 +63,8 @@ class DataHandlerSingleStation(AbstractDataHandler):
interpolation_method: Union[str, Tuple[str]] = DEFAULT_INTERPOLATION_METHOD,
overwrite_local_data: bool = False, transformation=None, store_data_locally: bool = True,
min_length: int = 0, start=None, end=None, variables=None, data_origin: Dict = None,
- lazy_preprocessing: bool = False, aggregation_dim=None, **kwargs):
+ lazy_preprocessing: bool = False, overwrite_lazy_data=False, aggregation_dim=None, **kwargs):
+
super().__init__()
self.station = helpers.to_list(station)
self.path = self.setup_data_path(data_path, sampling)
@@ -93,6 +96,7 @@ class DataHandlerSingleStation(AbstractDataHandler):
self.interpolation_method = interpolation_method
self.overwrite_local_data = overwrite_local_data
+ self.overwrite_lazy_data = True if self.overwrite_local_data is True else overwrite_lazy_data
self.store_data_locally = store_data_locally
self.min_length = min_length
self.start = start
@@ -108,6 +112,13 @@ class DataHandlerSingleStation(AbstractDataHandler):
# create samples
self.setup_samples()
+ self.clean_up()
+
+ def clean_up(self):
+ self._data = None
+ self.input_data = None
+ self.target_data = None
+ gc.collect()
def __str__(self):
return self.station[0]
@@ -214,7 +225,8 @@ class DataHandlerSingleStation(AbstractDataHandler):
elif method == "centre":
return statistics.centre_apply(data, mean), {"mean": mean, "method": method}
elif method == "min_max":
- return statistics.min_max_apply(data, min, max), {"min": min, "max": max, "method": method,
+ kws = {"feature_range": feature_range} if feature_range is not None else {}
+ return statistics.min_max_apply(data, min, max, **kws), {"min": min, "max": max, "method": method,
"feature_range": feature_range}
elif method == "log":
return statistics.log_apply(data, mean, std), {"mean": mean, "std": std, "method": method}
@@ -254,7 +266,7 @@ class DataHandlerSingleStation(AbstractDataHandler):
hash = self._get_hash()
filename = os.path.join(self.lazy_path, hash + ".pickle")
if not os.path.exists(filename):
- dill.dump(self._create_lazy_data(), file=open(filename, "wb"))
+ dill.dump(self._create_lazy_data(), file=open(filename, "wb"), protocol=4)
def _create_lazy_data(self):
return [self._data, self.meta, self.input_data, self.target_data]
@@ -263,6 +275,8 @@ class DataHandlerSingleStation(AbstractDataHandler):
hash = self._get_hash()
filename = os.path.join(self.lazy_path, hash + ".pickle")
try:
+ if self.overwrite_lazy_data is True:
+ raise FileNotFoundError
with open(filename, "rb") as pickle_file:
lazy_data = dill.load(pickle_file)
self._extract_lazy(lazy_data)
@@ -405,8 +419,7 @@ class DataHandlerSingleStation(AbstractDataHandler):
"""
chem_vars = ["benzene", "ch4", "co", "ethane", "no", "no2", "nox", "o3", "ox", "pm1", "pm10", "pm2p5",
"propane", "so2", "toluene"]
- # used_chem_vars = list(set(chem_vars) & set(self.statistics_per_var.keys()))
- used_chem_vars = list(set(chem_vars) & set(data.variables.values))
+ used_chem_vars = list(set(chem_vars) & set(data.coords[self.target_dim].values))
if len(used_chem_vars) > 0:
data.loc[..., used_chem_vars] = data.loc[..., used_chem_vars].clip(min=minimum)
return data
@@ -452,11 +465,8 @@ class DataHandlerSingleStation(AbstractDataHandler):
:return: this array
"""
ind = pd.DataFrame({'val': index_value}, index=index_value)
- # res = xr.Dataset.from_dataframe(ind).to_array().rename({'index': index_name}).squeeze(dim=squeez/e_dim, drop=True)
res = xr.Dataset.from_dataframe(ind).to_array(squeeze_dim).rename({'index': index_name}).squeeze(
- dim=squeeze_dim,
- drop=True
- )
+ dim=squeeze_dim, drop=True)
res.name = index_name
return res
@@ -764,8 +774,6 @@ class DataHandlerSingleStation(AbstractDataHandler):
if __name__ == "__main__":
- # dp = AbstractDataPrep('data/', 'dummy', 'DEBW107', ['o3', 'temp'], statistics_per_var={'o3': 'dma8eu', 'temp': 'maximum'})
- # print(dp)
statistics_per_var = {'o3': 'dma8eu', 'temp-rea-miub': 'maximum'}
sp = DataHandlerSingleStation(data_path='/home/felix/PycharmProjects/mlt_new/data/', station='DEBY122',
statistics_per_var=statistics_per_var, station_type='background',
diff --git a/mlair/data_handler/data_handler_with_filter.py b/mlair/data_handler/data_handler_with_filter.py
index e76f396aea80b2db76e01ea5baacf71d024b0d23..4707fd580562a68fd6b2dc0843551905e70d7e50 100644
--- a/mlair/data_handler/data_handler_with_filter.py
+++ b/mlair/data_handler/data_handler_with_filter.py
@@ -4,6 +4,7 @@ __author__ = 'Lukas Leufen'
__date__ = '2020-08-26'
import inspect
+import copy
import numpy as np
import pandas as pd
import xarray as xr
@@ -12,7 +13,7 @@ from functools import partial
import logging
from mlair.data_handler.data_handler_single_station import DataHandlerSingleStation
from mlair.data_handler import DefaultDataHandler
-from mlair.helpers import remove_items, to_list, TimeTrackingWrapper
+from mlair.helpers import remove_items, to_list, TimeTrackingWrapper, statistics
from mlair.helpers.filter import KolmogorovZurbenkoFilterMovingWindow as KZFilter
from mlair.helpers.filter import FIRFilter, ClimateFIRFilter, omega_null_kzf
@@ -126,32 +127,16 @@ class DataHandlerFilter(DefaultDataHandler):
list_of_args = arg_spec.args + arg_spec.kwonlyargs + super_own_args
return remove_items(list_of_args, ["self"] + list(args))
- def get_X_original(self):
- if self.use_filter_branches is True:
- X = []
- for data in self._collection:
- X_total = data.get_X()
- filter_dim = data.filter_dim
- for filter_name in data.filter_dim_order:
- X.append(X_total.sel({filter_dim: filter_name}, drop=True))
- return X
- else:
- return super().get_X_original()
-
class DataHandlerFirFilterSingleStation(DataHandlerFilterSingleStation):
"""Data handler for a single station to be used by a superior data handler. Inputs are FIR filtered."""
_requirements = remove_items(DataHandlerFilterSingleStation.requirements(), "station")
- _hash = DataHandlerFilterSingleStation._hash + ["filter_cutoff_period", "filter_order", "filter_window_type",
- "_add_unfiltered"]
+ _hash = DataHandlerFilterSingleStation._hash + ["filter_cutoff_period", "filter_order", "filter_window_type"]
DEFAULT_WINDOW_TYPE = ("kaiser", 5)
- DEFAULT_ADD_UNFILTERED = False
- def __init__(self, *args, filter_cutoff_period, filter_order, filter_window_type=DEFAULT_WINDOW_TYPE,
- filter_add_unfiltered=DEFAULT_ADD_UNFILTERED, **kwargs):
- # self._check_sampling(**kwargs)
+ def __init__(self, *args, filter_cutoff_period, filter_order, filter_window_type=DEFAULT_WINDOW_TYPE, **kwargs):
# self.original_data = None # ToDo: implement here something to store unfiltered data
self.fs = self._get_fs(**kwargs)
if filter_window_type == "kzf":
@@ -161,7 +146,7 @@ class DataHandlerFirFilterSingleStation(DataHandlerFilterSingleStation):
assert len(self.filter_cutoff_period) == (len(filter_order) - len(removed_index))
self.filter_order = self._prepare_filter_order(filter_order, removed_index, self.fs)
self.filter_window_type = filter_window_type
- self._add_unfiltered = filter_add_unfiltered
+ self.unfiltered_name = "unfiltered"
super().__init__(*args, **kwargs)
@staticmethod
@@ -223,8 +208,6 @@ class DataHandlerFirFilterSingleStation(DataHandlerFilterSingleStation):
self.filter_window_type, self.target_dim)
self.fir_coeff = fir.filter_coefficients()
fir_data = fir.filtered_data()
- if self._add_unfiltered is True:
- fir_data.append(self.input_data)
self.input_data = xr.concat(fir_data, pd.Index(self.create_filter_index(), name=self.filter_dim))
# this is just a code snippet to check the results of the kz filter
# import matplotlib
@@ -249,8 +232,6 @@ class DataHandlerFirFilterSingleStation(DataHandlerFilterSingleStation):
else:
index.append(f"band{band_num}")
band_num += 1
- if self._add_unfiltered:
- index.append("unfiltered")
self.filter_dim_order = index
return pd.Index(index, name=self.filter_dim)
@@ -261,13 +242,89 @@ class DataHandlerFirFilterSingleStation(DataHandlerFilterSingleStation):
_data, _meta, _input_data, _target_data, self.fir_coeff, self.filter_dim_order = lazy_data
super(__class__, self)._extract_lazy((_data, _meta, _input_data, _target_data))
+ def transform(self, data_in, dim: Union[str, int] = 0, inverse: bool = False, opts=None,
+ transformation_dim=None):
+ """
+ Transform data according to given transformation settings.
+
+ This function transforms a xarray.dataarray (along dim) or pandas.DataFrame (along axis) either with mean=0
+ and std=1 (`method=standardise`) or centers the data with mean=0 and no change in data scale
+ (`method=centre`). Furthermore, this sets an internal instance attribute for later inverse transformation. This
+ method will raise an AssertionError if an internal transform method was already set ('inverse=False') or if the
+ internal transform method, internal mean and internal standard deviation weren't set ('inverse=True').
+
+ :param string/int dim: This param is not used for inverse transformation.
+ | for xarray.DataArray as string: name of dimension which should be standardised
+ | for pandas.DataFrame as int: axis of dimension which should be standardised
+ :param inverse: Switch between transformation and inverse transformation.
+
+ :return: xarray.DataArrays or pandas.DataFrames:
+ #. mean: Mean of data
+ #. std: Standard deviation of data
+ #. data: Standardised data
+ """
+
+ if transformation_dim is None:
+ transformation_dim = self.DEFAULT_TARGET_DIM
+
+ def f(data, method="standardise", feature_range=None):
+ if method == "standardise":
+ return statistics.standardise(data, dim)
+ elif method == "centre":
+ return statistics.centre(data, dim)
+ elif method == "min_max":
+ kwargs = {"feature_range": feature_range} if feature_range is not None else {}
+ return statistics.min_max(data, dim, **kwargs)
+ elif method == "log":
+ return statistics.log(data, dim)
+ else:
+ raise NotImplementedError
+
+ def f_apply(data, method, **kwargs):
+ for k, v in kwargs.items():
+ if not (isinstance(v, xr.DataArray) or v is None):
+ _, opts = statistics.min_max(data, dim)
+ helper = xr.ones_like(opts['min'])
+ kwargs[k] = helper * v
+ mean = kwargs.pop('mean', None)
+ std = kwargs.pop('std', None)
+ min = kwargs.pop('min', None)
+ max = kwargs.pop('max', None)
+ feature_range = kwargs.pop('feature_range', None)
+
+ if method == "standardise":
+ return statistics.standardise_apply(data, mean, std), {"mean": mean, "std": std, "method": method}
+ elif method == "centre":
+ return statistics.centre_apply(data, mean), {"mean": mean, "method": method}
+ elif method == "min_max":
+ return statistics.min_max_apply(data, min, max), {"min": min, "max": max, "method": method,
+ "feature_range": feature_range}
+ elif method == "log":
+ return statistics.log_apply(data, mean, std), {"mean": mean, "std": std, "method": method}
+ else:
+ raise NotImplementedError
+
+ opts = opts or {}
+ opts_updated = {}
+ if not inverse:
+ transformed_values = []
+ for var in data_in.variables.values:
+ data_var = data_in.sel(**{transformation_dim: [var]})
+ var_opts = opts.get(var, {})
+ _apply = (var_opts.get("mean", None) is not None) or (var_opts.get("min") is not None)
+ values, new_var_opts = locals()["f_apply" if _apply else "f"](data_var, **var_opts)
+ opts_updated[var] = copy.deepcopy(new_var_opts)
+ transformed_values.append(values)
+ return xr.concat(transformed_values, dim=transformation_dim), opts_updated
+ else:
+ return self.inverse_transform(data_in, opts, transformation_dim)
+
class DataHandlerFirFilter(DataHandlerFilter):
"""Data handler using FIR filtered data."""
data_handler = DataHandlerFirFilterSingleStation
data_handler_transformation = DataHandlerFirFilterSingleStation
- _requirements = data_handler.requirements()
class DataHandlerKzFilterSingleStation(DataHandlerFilterSingleStation):
@@ -393,13 +450,8 @@ class DataHandlerClimateFirFilterSingleStation(DataHandlerFirFilterSingleStation
climate_filter.filtered_data]
# create input data with filter index
- input_data = xr.concat(climate_filter_data, pd.Index(self.create_filter_index(), name=self.filter_dim))
-
- # add unfiltered raw data
- if self._add_unfiltered is True:
- data_raw = self.shift(self.input_data, self.time_dim, -self.window_history_size)
- data_raw = data_raw.expand_dims({self.filter_dim: ["unfiltered"]}, -1)
- input_data = xr.concat([input_data, data_raw], self.filter_dim)
+ input_data = xr.concat(climate_filter_data, pd.Index(self.create_filter_index(add_unfiltered_index=False),
+ name=self.filter_dim))
self.input_data = input_data
@@ -410,7 +462,7 @@ class DataHandlerClimateFirFilterSingleStation(DataHandlerFirFilterSingleStation
# self.input_data.sel(filter="low", variables="temp", Stations="DEBW107").plot()
# self.input_data.sel(variables="temp", Stations="DEBW107").plot.line(hue="filter")
- def create_filter_index(self) -> pd.Index:
+ def create_filter_index(self, add_unfiltered_index=True) -> pd.Index:
"""
Round cut off periods in days and append 'res' for residuum index.
@@ -421,8 +473,6 @@ class DataHandlerClimateFirFilterSingleStation(DataHandlerFirFilterSingleStation
f = lambda x: int(np.round(x)) if x >= 10 else np.round(x, 1)
index = list(map(f, index.tolist()))
index = list(map(lambda x: str(x) + "d", index)) + ["res"]
- if self._add_unfiltered:
- index.append("unfiltered")
self.filter_dim_order = index
return pd.Index(index, name=self.filter_dim)
@@ -491,11 +541,3 @@ class DataHandlerClimateFirFilter(DataHandlerFilter):
_requirements = data_handler.requirements()
_store_attributes = data_handler.store_attributes()
- # def get_X_original(self):
- # X = []
- # for data in self._collection:
- # X_total = data.get_X()
- # filter_dim = data.filter_dim
- # for filter in data.filter_dim_order:
- # X.append(X_total.sel({filter_dim: filter}, drop=True))
- # return X
diff --git a/mlair/data_handler/default_data_handler.py b/mlair/data_handler/default_data_handler.py
index 1b4715a05d7e072add90137b89f7944299f449f5..3649eaf65ed7efe45c5fac54f892bd1e471e5838 100644
--- a/mlair/data_handler/default_data_handler.py
+++ b/mlair/data_handler/default_data_handler.py
@@ -7,6 +7,8 @@ import inspect
import gc
import logging
import os
+import pickle
+import random
import dill
import shutil
from functools import reduce
@@ -50,6 +52,7 @@ class DefaultDataHandler(AbstractDataHandler):
self._Y = None
self._X_extreme = None
self._Y_extreme = None
+ self._data_intersection = None
self._use_multiprocessing = use_multiprocessing
self._max_number_multiprocessing = max_number_multiprocessing
_name_affix = str(f"{str(self.id_class)}_{name_affix}" if name_affix is not None else id(self))
@@ -94,7 +97,7 @@ class DefaultDataHandler(AbstractDataHandler):
data = {"X": self._X, "Y": self._Y, "X_extreme": self._X_extreme, "Y_extreme": self._Y_extreme}
data = self._force_dask_computation(data)
with open(self._save_file, "wb") as f:
- dill.dump(data, f)
+ dill.dump(data, f, protocol=4)
logging.debug(f"save pickle data to {self._save_file}")
self._reset_data()
@@ -126,7 +129,7 @@ class DefaultDataHandler(AbstractDataHandler):
return X, Y
def __repr__(self):
- return ";".join(list(map(lambda x: str(x), self._collection)))
+ return str(self._collection[0])
def get_X_original(self):
X = []
@@ -165,10 +168,15 @@ class DefaultDataHandler(AbstractDataHandler):
else:
X = list(map(lambda x: x.sel({dim: intersect}), X_original))
Y = Y_original.sel({dim: intersect})
+ self._data_intersection = intersect
self._X, self._Y = X, Y
def get_observation(self):
- return self.id_class.observation.copy().squeeze()
+ dim = self.time_dim
+ if self._data_intersection is not None:
+ return self.id_class.observation.sel({dim: self._data_intersection}).copy().squeeze()
+ else:
+ return self.id_class.observation.copy().squeeze()
def apply_transformation(self, data, base="target", dim=0, inverse=False):
return self.id_class.apply_transformation(data, dim=dim, base=base, inverse=inverse)
@@ -208,30 +216,29 @@ class DefaultDataHandler(AbstractDataHandler):
raise TypeError(f"Elements of list extreme_values have to be {number.__args__}, but at least element "
f"{i} is type {type(i)}")
+ extremes_X, extremes_Y = None, None
for extr_val in sorted(extreme_values):
# check if some extreme values are already extracted
- if (self._X_extreme is None) or (self._Y_extreme is None):
- X = self._X
- Y = self._Y
+ if (extremes_X is None) or (extremes_Y is None):
+ X, Y = self._X, self._Y
+ extremes_X, extremes_Y = X, Y
else: # one extr value iteration is done already: self.extremes_label is NOT None...
- X = self._X_extreme
- Y = self._Y_extreme
+ X, Y = self._X_extreme, self._Y_extreme
# extract extremes based on occurrence in labels
other_dims = remove_items(list(Y.dims), dim)
if extremes_on_right_tail_only:
- extreme_idx = (Y > extr_val).any(dim=other_dims)
+ extreme_idx = (extremes_Y > extr_val).any(dim=other_dims)
else:
- extreme_idx = xr.concat([(Y < -extr_val).any(dim=other_dims[0]),
- (Y > extr_val).any(dim=other_dims[0])],
- dim=other_dims[1]).any(dim=other_dims[1])
+ extreme_idx = xr.concat([(extremes_Y < -extr_val).any(dim=other_dims[0]),
+ (extremes_Y > extr_val).any(dim=other_dims[0])],
+ dim=other_dims[0]).any(dim=other_dims[0])
- extremes_X = list(map(lambda x: x.sel(**{dim: extreme_idx}), X))
+ sel = extreme_idx[extreme_idx].coords[dim].values
+ extremes_X = list(map(lambda x: x.sel(**{dim: sel}), extremes_X))
self._add_timedelta(extremes_X, dim, timedelta)
- # extremes_X = list(map(lambda x: x.coords[dim].values + np.timedelta64(*timedelta), extremes_X))
-
- extremes_Y = Y.sel(**{dim: extreme_idx})
- extremes_Y.coords[dim].values += np.timedelta64(*timedelta)
+ extremes_Y = extremes_Y.sel(**{dim: extreme_idx})
+ self._add_timedelta([extremes_Y], dim, timedelta)
self._Y_extreme = xr.concat([Y, extremes_Y], dim=dim)
self._X_extreme = list(map(lambda x1, x2: xr.concat([x1, x2], dim=dim), X, extremes_X))
@@ -239,10 +246,10 @@ class DefaultDataHandler(AbstractDataHandler):
@staticmethod
def _add_timedelta(data, dim, timedelta):
for d in data:
- d.coords[dim].values += np.timedelta64(*timedelta)
+ d.coords[dim] = d.coords[dim].values + np.timedelta64(*timedelta)
@classmethod
- def transformation(cls, set_stations, **kwargs):
+ def transformation(cls, set_stations, tmp_path=None, dh_transformation=None, **kwargs):
"""
### supported transformation methods
@@ -272,53 +279,45 @@ class DefaultDataHandler(AbstractDataHandler):
If min and max are not None, the default data handler expects this parameters to match the data and applies
this values to the data. Make sure that all dimensions and/or coordinates are in agreement.
"""
+ if dh_transformation is None:
+ dh_transformation = cls.data_handler_transformation
- sp_keys = {k: copy.deepcopy(kwargs[k]) for k in cls._requirements if k in kwargs}
+ sp_keys = {k: copy.deepcopy(kwargs[k]) for k in dh_transformation.requirements() if k in kwargs}
if "transformation" not in sp_keys.keys():
return
transformation_dict = ({}, {})
- def _inner():
- """Inner method that is performed in both serial and parallel approach."""
- if dh is not None:
- for i, transformation in enumerate(dh._transformation):
- for var in transformation.keys():
- if var not in transformation_dict[i].keys():
- transformation_dict[i][var] = {}
- opts = transformation[var]
- if not transformation_dict[i][var].get("method", opts["method"]) == opts["method"]:
- # data handlers with filters are allowed to change transformation method to standardise
- assert hasattr(dh, "filter_dim") and opts["method"] == "standardise"
- transformation_dict[i][var]["method"] = opts["method"]
- for k in ["mean", "std", "min", "max"]:
- old = transformation_dict[i][var].get(k, None)
- new = opts.get(k)
- transformation_dict[i][var][k] = new if old is None else old.combine_first(new)
- if "feature_range" in opts.keys():
- transformation_dict[i][var]["feature_range"] = opts.get("feature_range", None)
-
max_process = kwargs.get("max_number_multiprocessing", 16)
n_process = min([psutil.cpu_count(logical=False), len(set_stations), max_process]) # use only physical cpus
if n_process > 1 and kwargs.get("use_multiprocessing", True) is True: # parallel solution
logging.info("use parallel transformation approach")
- pool = multiprocessing.Pool(
- min([psutil.cpu_count(logical=False), len(set_stations), 16])) # use only physical cpus
+ pool = multiprocessing.Pool(n_process) # use only physical cpus
logging.info(f"running {getattr(pool, '_processes')} processes in parallel")
+ sp_keys.update({"tmp_path": tmp_path, "return_strategy": "reference"})
output = [
- pool.apply_async(f_proc, args=(cls.data_handler_transformation, station), kwds=sp_keys)
+ pool.apply_async(f_proc, args=(dh_transformation, station), kwds=sp_keys)
for station in set_stations]
for p in output:
- dh, s = p.get()
- _inner()
+ _res_file, s = p.get()
+ with open(_res_file, "rb") as f:
+ dh = dill.load(f)
+ os.remove(_res_file)
+ transformation_dict = cls.update_transformation_dict(dh, transformation_dict)
pool.close()
else: # serial solution
logging.info("use serial transformation approach")
+ sp_keys.update({"return_strategy": "result"})
for station in set_stations:
- dh, s = f_proc(cls.data_handler_transformation, station, **sp_keys)
- _inner()
+ dh, s = f_proc(dh_transformation, station, **sp_keys)
+ transformation_dict = cls.update_transformation_dict(dh, transformation_dict)
# aggregate all information
iter_dim = sp_keys.get("iter_dim", cls.DEFAULT_ITER_DIM)
+ transformation_dict = cls.aggregate_transformation(transformation_dict, iter_dim)
+ return transformation_dict
+
+ @classmethod
+ def aggregate_transformation(cls, transformation_dict, iter_dim):
pop_list = []
for i, transformation in enumerate(transformation_dict):
for k in transformation.keys():
@@ -339,19 +338,47 @@ class DefaultDataHandler(AbstractDataHandler):
transformation_dict[i].pop(k)
return transformation_dict
+ @classmethod
+ def update_transformation_dict(cls, dh, transformation_dict):
+ """Inner method that is performed in both serial and parallel approach."""
+ if dh is not None:
+ for i, transformation in enumerate(dh._transformation):
+ for var in transformation.keys():
+ if var not in transformation_dict[i].keys():
+ transformation_dict[i][var] = {}
+ opts = transformation[var]
+ if not transformation_dict[i][var].get("method", opts["method"]) == opts["method"]:
+ # data handlers with filters are allowed to change transformation method to standardise
+ assert hasattr(dh, "filter_dim") and opts["method"] == "standardise"
+ transformation_dict[i][var]["method"] = opts["method"]
+ for k in ["mean", "std", "min", "max"]:
+ old = transformation_dict[i][var].get(k, None)
+ new = opts.get(k)
+ transformation_dict[i][var][k] = new if old is None else old.combine_first(new)
+ if "feature_range" in opts.keys():
+ transformation_dict[i][var]["feature_range"] = opts.get("feature_range", None)
+ return transformation_dict
+
def get_coordinates(self):
return self.id_class.get_coordinates()
-def f_proc(data_handler, station, **sp_keys):
+def f_proc(data_handler, station, return_strategy="", tmp_path=None, **sp_keys):
"""
Try to create a data handler for given arguments. If build fails, this station does not fulfil all requirements and
therefore f_proc will return None as indication. On a successful build, f_proc returns the built data handler and
the station that was used. This function must be implemented globally to work together with multiprocessing.
"""
+ assert return_strategy in ["result", "reference"]
try:
res = data_handler(station, **sp_keys)
except (AttributeError, EmptyQueryResult, KeyError, ValueError) as e:
logging.info(f"remove station {station} because it raised an error: {e}")
res = None
- return res, station
+ if return_strategy == "result":
+ return res, station
+ else:
+ _tmp_file = os.path.join(tmp_path, f"{station}_{'%032x' % random.getrandbits(128)}.pickle")
+ with open(_tmp_file, "wb") as f:
+ dill.dump(res, f, protocol=4)
+ return _tmp_file, station
diff --git a/mlair/helpers/filter.py b/mlair/helpers/filter.py
index a63cef975888162f335e4528c2f99bdfc7a892d5..543cff3624577ac617733b8b593c5f52f25196b3 100644
--- a/mlair/helpers/filter.py
+++ b/mlair/helpers/filter.py
@@ -173,8 +173,10 @@ class ClimateFIRFilter:
# visualize
if self.plot_path is not None:
- self.PlotClimateFirFilter(self.plot_path, self.plot_data, sampling, plot_name)
- # self._plot(sampling, new_dim=new_dim)
+ try:
+ self.PlotClimateFirFilter(self.plot_path, self.plot_data, sampling, plot_name)
+ except Exception as e:
+ logging.info(f"Could not plot climate fir filter due to following reason:\n{e}")
@staticmethod
def _minimum_length(order, minimum_length, pos, window):
@@ -382,7 +384,7 @@ class ClimateFIRFilter:
_end = pd.to_datetime(data.coords[time_dim].max().values).year
filt_coll = []
for _year in range(_start, _end + 1):
- logging.info(f"{data.coords['Stations'].values[0]} ({var}): year={_year}")
+ logging.debug(f"{data.coords['Stations'].values[0]} ({var}): year={_year}")
time_slice = self._create_time_range_extend(_year, sampling, extend_length_history)
d = data.sel({var_dim: [var], time_dim: time_slice})
@@ -504,137 +506,6 @@ class ClimateFIRFilter:
res.name = index_name
return res
- def _plot(self, sampling, new_dim="window"):
- h = None
- td_type = {"1d": "D", "1H": "h"}.get(sampling)
- if self.plot_path is None:
- return
- plot_folder = os.path.join(os.path.abspath(self.plot_path), "climFIR")
- if not os.path.exists(plot_folder):
- os.makedirs(plot_folder)
-
- # set plot parameter
- rc_params = {'axes.labelsize': 'large',
- 'xtick.labelsize': 'large',
- 'ytick.labelsize': 'large',
- 'legend.fontsize': 'medium',
- 'axes.titlesize': 'large',
- }
- plt.rcParams.update(rc_params)
-
- plot_dict = {}
- for i, o in enumerate(range(len(self.plot_data))):
- plot_data = self.plot_data[i]
- for p_d in plot_data:
- var = p_d.get("var")
- t0 = p_d.get("t0")
- filter_input = p_d.get("filter_input")
- filter_input_nc = p_d.get("filter_input_nc")
- valid_range = p_d.get("valid_range")
- time_range = p_d.get("time_range")
- new_dim = p_d.get("new_dim")
- h = p_d.get("h")
- plot_dict_var = plot_dict.get(var, {})
- plot_dict_t0 = plot_dict_var.get(t0, {})
- plot_dict_order = {"filter_input": filter_input,
- "filter_input_nc": filter_input_nc,
- "valid_range": valid_range,
- "time_range": time_range,
- "order": o, "h": h}
- plot_dict_t0[i] = plot_dict_order
- plot_dict_var[t0] = plot_dict_t0
- plot_dict[var] = plot_dict_var
-
- for var, viz_date_dict in plot_dict.items():
- for it0, t0 in enumerate(viz_date_dict.keys()):
- viz_data = viz_date_dict[t0]
- residuum_true = None
- for ifilter in sorted(viz_data.keys()):
- data = viz_data[ifilter]
- filter_input = data["filter_input"]
- filter_input_nc = data["filter_input_nc"] if residuum_true is None else residuum_true.sel(
- {new_dim: filter_input.coords[new_dim]})
- valid_range = data["valid_range"]
- time_axis = data["time_range"]
- # time_axis = pd.date_range(t_minus, t_plus, freq=sampling)
- filter_order = data["order"]
- h = data["h"]
- t_minus = t0 + np.timedelta64(-int(1.5 * valid_range.start), td_type)
- t_plus = t0 + np.timedelta64(int(0.5 * valid_range.start), td_type)
- fig, ax = plt.subplots()
- ax.axvspan(t0 + np.timedelta64(-valid_range.start, td_type),
- t0 + np.timedelta64(valid_range.stop - 1, td_type), color="whitesmoke",
- label="valid area")
- ax.axvline(t0, color="lightgrey", lw=6, label="time of interest ($t_0$)")
-
- # original data
- ax.plot(time_axis, filter_input_nc.values.flatten(), color="darkgrey", linestyle="dashed",
- label="original")
-
- # clim apriori
- if ifilter == 0:
- d_tmp = filter_input.sel(
- {new_dim: slice(0, filter_input.coords[new_dim].values.max())}).values.flatten()
- else:
- d_tmp = filter_input.values.flatten()
- ax.plot(time_axis[len(time_axis) - len(d_tmp):], d_tmp, color="darkgrey", linestyle="solid",
- label="estimated future")
-
- # clim filter response
- filt = xr.apply_ufunc(fir_filter_convolve, filter_input,
- input_core_dims=[[new_dim]],
- output_core_dims=[[new_dim]],
- vectorize=True,
- kwargs={"h": h},
- output_dtypes=[filter_input.dtype])
- ax.plot(time_axis, filt.values.flatten(), color="black", linestyle="solid",
- label="clim filter response", linewidth=2)
- residuum_estimated = filter_input - filt
-
- # ideal filter response
- filt = xr.apply_ufunc(fir_filter_convolve, filter_input_nc,
- input_core_dims=[[new_dim]],
- output_core_dims=[[new_dim]],
- vectorize=True,
- kwargs={"h": h},
- output_dtypes=[filter_input.dtype])
- ax.plot(time_axis, filt.values.flatten(), color="black", linestyle="dashed",
- label="ideal filter response", linewidth=2)
- residuum_true = filter_input_nc - filt
-
- # set title, legend, and save plot
- ax_start = max(t_minus, time_axis[0])
- ax_end = min(t_plus, time_axis[-1])
- ax.set_xlim((ax_start, ax_end))
- plt.title(f"Input of ClimFilter ({str(var)})")
- plt.legend()
- fig.autofmt_xdate()
- plt.tight_layout()
- plot_name = os.path.join(plot_folder,
- f"climFIR_{self.plot_name}_{str(var)}_{it0}_{ifilter}.pdf")
- plt.savefig(plot_name, dpi=300)
- plt.close('all')
-
- # plot residuum
- fig, ax = plt.subplots()
- ax.axvspan(t0 + np.timedelta64(-valid_range.start, td_type),
- t0 + np.timedelta64(valid_range.stop - 1, td_type), color="whitesmoke",
- label="valid area")
- ax.axvline(t0, color="lightgrey", lw=6, label="time of interest ($t_0$)")
- ax.plot(time_axis, residuum_true.values.flatten(), color="black", linestyle="dashed",
- label="ideal filter residuum", linewidth=2)
- ax.plot(time_axis, residuum_estimated.values.flatten(), color="black", linestyle="solid",
- label="clim filter residuum", linewidth=2)
- ax.set_xlim((ax_start, ax_end))
- plt.title(f"Residuum of ClimFilter ({str(var)})")
- plt.legend(loc="upper left")
- fig.autofmt_xdate()
- plt.tight_layout()
- plot_name = os.path.join(plot_folder,
- f"climFIR_{self.plot_name}_{str(var)}_{it0}_{ifilter}_residuum.pdf")
- plt.savefig(plot_name, dpi=300)
- plt.close('all')
-
@property
def filter_coefficients(self):
return self._h
diff --git a/mlair/helpers/helpers.py b/mlair/helpers/helpers.py
index 40fd98b93796c7b32a1e767e0c9179f429b9a0c9..ccf0250b0ac70e77b7b456b05883fbc7b8d2d34b 100644
--- a/mlair/helpers/helpers.py
+++ b/mlair/helpers/helpers.py
@@ -8,6 +8,9 @@ import json
import math
import os
+import sys
+
+
import numpy as np
import pandas as pd
import xarray as xr
@@ -84,8 +87,10 @@ def remove_items(obj: Union[List, Dict, Tuple], items: Any):
def remove_from_list(list_obj, item_list):
"""Remove implementation for lists."""
- if len(items) > 1:
+ if len(item_list) > 1:
return [e for e in list_obj if e not in item_list]
+ elif len(item_list) == 0:
+ return list_obj
else:
list_obj = list_obj.copy()
try:
@@ -220,3 +225,33 @@ if __name__ == '__main__':
print(123)
+# def convert_size(size_bytes):
+# if size_bytes == 0:
+# return "0B"
+# size_name = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB")
+# i = int(math.floor(math.log(size_bytes, 1024)))
+# p = math.pow(1024, i)
+# s = round(size_bytes / p, 2)
+# return "%s %s" % (s, size_name[i])
+#
+#
+# def get_size(obj, seen=None):
+# """Recursively finds size of objects"""
+# size = sys.getsizeof(obj)
+# if seen is None:
+# seen = set()
+# obj_id = id(obj)
+# if obj_id in seen:
+# return 0
+# # Important mark as seen *before* entering recursion to gracefully handle
+# # self-referential objects
+# seen.add(obj_id)
+# if isinstance(obj, dict):
+# size += sum([get_size(v, seen) for v in obj.values()])
+# size += sum([get_size(k, seen) for k in obj.keys()])
+# elif hasattr(obj, '__dict__'):
+# size += get_size(obj.__dict__, seen)
+# elif hasattr(obj, '__iter__') and not isinstance(obj, (str, bytes, bytearray)):
+# size += sum([get_size(i, seen) for i in obj])
+# return size
+
diff --git a/mlair/helpers/statistics.py b/mlair/helpers/statistics.py
index a1e713a8c135800d02ff7c27894485a5da7fae37..8b510f704396b35cf022089e7d94e037f4c62a2b 100644
--- a/mlair/helpers/statistics.py
+++ b/mlair/helpers/statistics.py
@@ -152,7 +152,10 @@ def min_max_apply(data: Data, _min: Data, _max: Data, feature_range: Data = (0,
:param feature_range: scale data to any interval given in feature range. Default is scaling on interval [0, 1].
:return: min/max scaled data
"""
- return (data - _min) / (_max - _min) * (max(feature_range) - min(feature_range)) + min(feature_range)
+ if not isinstance(feature_range, xr.DataArray):
+ return (data - _min) / (_max - _min) * (max(feature_range) - min(feature_range)) + min(feature_range)
+ else:
+ return (data - _min) / (_max - _min) * (feature_range.max() - feature_range.min()) + feature_range.min()
def log(data: Data, dim: Union[str, int]) -> Tuple[Data, Dict[(str, Data)]]:
@@ -284,7 +287,7 @@ class SkillScores:
combination_strings = [f"{first}-{second}" for (first, second) in combinations]
return combinations, combination_strings
- def skill_scores(self) -> pd.DataFrame:
+ def skill_scores(self) -> [pd.DataFrame, pd.DataFrame]:
"""
Calculate skill scores for all combinations of model names.
@@ -293,6 +296,7 @@ class SkillScores:
ahead_names = list(self.external_data[self.ahead_dim].data)
combinations, combination_strings = self.get_model_name_combinations()
skill_score = pd.DataFrame(index=combination_strings)
+ count = pd.DataFrame(index=combination_strings)
for iahead in ahead_names:
data = self.external_data.sel({self.ahead_dim: iahead})
skill_score[iahead] = [self.general_skill_score(data,
@@ -300,7 +304,12 @@ class SkillScores:
reference_name=second,
observation_name=self.observation_name)
for (first, second) in combinations]
- return skill_score
+ count[iahead] = [self.get_count(data,
+ forecast_name=first,
+ reference_name=second,
+ observation_name=self.observation_name)
+ for (first, second) in combinations]
+ return skill_score, count
def climatological_skill_scores(self, internal_data: Data, forecast_name: str) -> xr.DataArray:
"""
@@ -314,7 +323,10 @@ class SkillScores:
:return: all CASES as well as all terms
"""
- ahead_names = list(self.external_data[self.ahead_dim].data)
+ if self.external_data is not None:
+ ahead_names = list(self.external_data[self.ahead_dim].data)
+ else:
+ ahead_names = list(internal_data[self.ahead_dim].data)
all_terms = ['AI', 'AII', 'AIII', 'AIV', 'BI', 'BII', 'BIV', 'CI', 'CIV', 'CASE I', 'CASE II', 'CASE III',
'CASE IV']
@@ -370,6 +382,21 @@ class SkillScores:
skill_score = 1 - mse(observation, forecast) / mse(observation, reference)
return skill_score.values
+ def get_count(self, data: Data, forecast_name: str, reference_name: str,
+ observation_name: str = None) -> np.ndarray:
+ r"""
+ Calculate general skill score based on mean squared error.
+
+ :param data: internal data containing data for observation, forecast and reference
+ :param observation_name: name of observation
+ :param forecast_name: name of forecast
+ :param reference_name: name of reference
+
+ :return: skill score of forecast
+ """
+ data = data.dropna("index")
+ return data.count("index").max().values
+
def skill_score_pre_calculations(self, data: Data, observation_name: str, forecast_name: str) -> Tuple[np.ndarray,
np.ndarray,
np.ndarray,
diff --git a/mlair/plotting/data_insight_plotting.py b/mlair/plotting/data_insight_plotting.py
index 513f64f2c174d94cb7230b141387c9a850d678cb..6180493741c030d5dfdfcfa8972035619632c8aa 100644
--- a/mlair/plotting/data_insight_plotting.py
+++ b/mlair/plotting/data_insight_plotting.py
@@ -8,6 +8,7 @@ import os
import logging
import multiprocessing
import psutil
+import sys
import numpy as np
import pandas as pd
@@ -459,7 +460,7 @@ class PlotDataHistogram(AbstractPlotClass): # pragma: no cover
"""
def __init__(self, generators: Dict[str, DataCollection], plot_folder: str = ".", plot_name="histogram",
- variables_dim="variables", time_dim="datetime", window_dim="window"):
+ variables_dim="variables", time_dim="datetime", window_dim="window", upsampling=False):
super().__init__(plot_folder, plot_name)
self.variables_dim = variables_dim
self.time_dim = time_dim
@@ -468,6 +469,8 @@ class PlotDataHistogram(AbstractPlotClass): # pragma: no cover
self.bins = {}
self.interval_width = {}
self.bin_edges = {}
+ if upsampling is True:
+ self._handle_upsampling(generators)
# input plots
for branch_pos in range(number_of_branches):
@@ -483,6 +486,11 @@ class PlotDataHistogram(AbstractPlotClass): # pragma: no cover
self._plot(add_name="target", subset=subset)
self._plot_combined(add_name="target")
+ @staticmethod
+ def _handle_upsampling(generators):
+ if "train" in generators:
+ generators.update({"train_upsampled": generators["train"]})
+
@staticmethod
def _get_inputs_targets(gens, dim):
k = list(gens.keys())[0]
@@ -495,11 +503,15 @@ class PlotDataHistogram(AbstractPlotClass): # pragma: no cover
def _calculate_hist(self, generators, variables, input_data=True, branch_pos=0):
n_bins = 100
for set_type, generator in generators.items():
+ upsampling = "upsampled" in set_type
tmp_bins = {}
tmp_edges = {}
end = {}
start = {}
- f = lambda x: x.get_X(as_numpy=False)[branch_pos] if input_data is True else x.get_Y(as_numpy=False)
+ if input_data is True:
+ f = lambda x: x.get_X(as_numpy=False, upsampling=upsampling)[branch_pos]
+ else:
+ f = lambda x: x.get_Y(as_numpy=False, upsampling=upsampling)
for gen in generator:
w = min(abs(f(gen).coords[self.window_dim].values))
data = f(gen).sel({self.window_dim: w})
@@ -536,6 +548,7 @@ class PlotDataHistogram(AbstractPlotClass): # pragma: no cover
bin_edges = self.bin_edges[subset]
interval_width = self.interval_width[subset]
colors = self.get_dataset_colors()
+ colors.update({"train_upsampled": colors.get("train_val", "#000000")})
for var in bins.keys():
fig, ax = plt.subplots()
hist_var = bins[var]
@@ -555,6 +568,7 @@ class PlotDataHistogram(AbstractPlotClass): # pragma: no cover
pdf_pages = matplotlib.backends.backend_pdf.PdfPages(plot_path)
variables = self.bins[list(self.bins.keys())[0]].keys()
colors = self.get_dataset_colors()
+ colors.update({"train_upsampled": colors.get("train_val", "#000000")})
for var in variables:
fig, ax = plt.subplots()
for subset in self.bins.keys():
@@ -622,16 +636,17 @@ class PlotPeriodogram(AbstractPlotClass): # pragma: no cover
@staticmethod
def _has_filter_dimension(g, pos):
- # check if coords raw data differs from input / target data
- check_data = g.id_class
- if "filter" not in [check_data.input_data, check_data.target_data][pos].coords.dims:
+ """Inspect if filtered data is provided and return number and labels of filtered components."""
+ check_class = g.id_class
+ check_data = [check_class.get_X(as_numpy=False), check_class.get_Y(as_numpy=False)][pos]
+ if not hasattr(check_class, "filter_dim"): # data handler has no filtered data
return 1, []
else:
- if len(set(check_data._data[0].coords.dims).symmetric_difference(check_data.input_data.coords.dims)) > 0:
- return g.id_class.input_data.coords["filter"].shape[0], g.id_class.input_data.coords[
- "filter"].values.tolist()
- else:
+ filter_dim = check_class.filter_dim
+ if filter_dim not in check_data.coords.dims: # current data has no filter (e.g. target data)
return 1, []
+ else:
+ return check_data.coords[filter_dim].shape[0], check_data.coords[filter_dim].values.tolist()
@TimeTrackingWrapper
def _prepare_pgram(self, generator, pos, multiple=1, use_multiprocessing=False):
@@ -649,7 +664,6 @@ class PlotPeriodogram(AbstractPlotClass): # pragma: no cover
plot_data_mean_single = dict()
self.f_index = np.logspace(-3, 0 if self._sampling == "daily" else np.log10(24), 1000)
raw_data_single = self._prepare_pgram_parallel_gen(generator, m, pos, use_multiprocessing)
- # raw_data_single = self._prepare_pgram_parallel_var(generator, m, pos, use_multiprocessing)
for var in raw_data_single.keys():
pgram_com = []
pgram_mean = 0
@@ -844,9 +858,10 @@ def f_proc(var, d_var, f_index, time_dim="datetime"): # pragma: no cover
return var_str, f_index, pgram
-
def f_proc_2(g, m, pos, variables_dim, time_dim, f_index): # pragma: no cover
raw_data_single = dict()
+ if hasattr(g.id_class, "lazy"):
+ g.id_class.load_lazy() if g.id_class.lazy is True else None
if m == 0:
d = g.id_class._data
else:
@@ -858,6 +873,8 @@ def f_proc_2(g, m, pos, variables_dim, time_dim, f_index): # pragma: no cover
d_var = d.loc[{variables_dim: var}].squeeze().dropna(time_dim)
var_str, f, pgram = f_proc(var, d_var, f_index)
raw_data_single[var_str] = [(f, pgram)]
+ if hasattr(g.id_class, "lazy"):
+ g.id_class.clean_up() if g.id_class.lazy is True else None
return raw_data_single
@@ -888,6 +905,8 @@ class PlotClimateFirFilter(AbstractPlotClass):
from mlair.helpers.filter import fir_filter_convolve
+ logging.info(f"start PlotClimateFirFilter for ({name})")
+
# adjust default plot parameters
rc_params = {
'axes.labelsize': 'large',
@@ -951,59 +970,63 @@ class PlotClimateFirFilter(AbstractPlotClass):
for it0, t0 in enumerate(viz_date_dict.keys()):
viz_data = viz_date_dict[t0]
residuum_true = None
- for ifilter in sorted(viz_data.keys()):
- data = viz_data[ifilter]
- filter_input = data["filter_input"]
- filter_input_nc = data["filter_input_nc"] if residuum_true is None else residuum_true.sel(
- {new_dim: filter_input.coords[new_dim]})
- valid_range = data["valid_range"]
- time_axis = data["time_range"]
- filter_order = data["order"]
- h = data["h"]
- fig, ax = plt.subplots()
-
- # plot backgrounds
- self._plot_valid_area(ax, t0, valid_range, td_type)
- self._plot_t0(ax, t0)
-
- # original data
- self._plot_original_data(ax, time_axis, filter_input_nc)
-
- # clim apriori
- self._plot_apriori(ax, time_axis, filter_input, new_dim, ifilter)
-
- # clim filter response
- residuum_estimated = self._plot_clim_filter(ax, time_axis, filter_input, new_dim, h,
+ try:
+ for ifilter in sorted(viz_data.keys()):
+ data = viz_data[ifilter]
+ filter_input = data["filter_input"]
+ filter_input_nc = data["filter_input_nc"] if residuum_true is None else residuum_true.sel(
+ {new_dim: filter_input.coords[new_dim]})
+ valid_range = data["valid_range"]
+ time_axis = data["time_range"]
+ filter_order = data["order"]
+ h = data["h"]
+ fig, ax = plt.subplots()
+
+ # plot backgrounds
+ self._plot_valid_area(ax, t0, valid_range, td_type)
+ self._plot_t0(ax, t0)
+
+ # original data
+ self._plot_original_data(ax, time_axis, filter_input_nc)
+
+ # clim apriori
+ self._plot_apriori(ax, time_axis, filter_input, new_dim, ifilter)
+
+ # clim filter response
+ residuum_estimated = self._plot_clim_filter(ax, time_axis, filter_input, new_dim, h,
+ output_dtypes=filter_input.dtype)
+
+ # ideal filter response
+ residuum_true = self._plot_ideal_filter(ax, time_axis, filter_input_nc, new_dim, h,
output_dtypes=filter_input.dtype)
- # ideal filter response
- residuum_true = self._plot_ideal_filter(ax, time_axis, filter_input_nc, new_dim, h,
- output_dtypes=filter_input.dtype)
-
- # set title, legend, and save plot
- xlims = self._set_xlim(ax, t0, filter_order, valid_range, td_type, time_axis)
-
- plt.title(f"Input of ClimFilter ({str(var)})")
- plt.legend()
- fig.autofmt_xdate()
- plt.tight_layout()
- self.plot_name = f"climFIR_{self._name}_{str(var)}_{it0}_{ifilter}"
- self._save()
-
- # plot residuum
- fig, ax = plt.subplots()
- self._plot_valid_area(ax, t0, valid_range, td_type)
- self._plot_t0(ax, t0)
- self._plot_series(ax, time_axis, residuum_true.values.flatten(), style="ideal")
- self._plot_series(ax, time_axis, residuum_estimated.values.flatten(), style="clim")
- ax.set_xlim(xlims)
- plt.title(f"Residuum of ClimFilter ({str(var)})")
- plt.legend(loc="upper left")
- fig.autofmt_xdate()
- plt.tight_layout()
-
- self.plot_name = f"climFIR_{self._name}_{str(var)}_{it0}_{ifilter}_residuum"
- self._save()
+ # set title, legend, and save plot
+ xlims = self._set_xlim(ax, t0, filter_order, valid_range, td_type, time_axis)
+
+ plt.title(f"Input of ClimFilter ({str(var)})")
+ plt.legend()
+ fig.autofmt_xdate()
+ plt.tight_layout()
+ self.plot_name = f"climFIR_{self._name}_{str(var)}_{it0}_{ifilter}"
+ self._save()
+
+ # plot residuum
+ fig, ax = plt.subplots()
+ self._plot_valid_area(ax, t0, valid_range, td_type)
+ self._plot_t0(ax, t0)
+ self._plot_series(ax, time_axis, residuum_true.values.flatten(), style="ideal")
+ self._plot_series(ax, time_axis, residuum_estimated.values.flatten(), style="clim")
+ ax.set_xlim(xlims)
+ plt.title(f"Residuum of ClimFilter ({str(var)})")
+ plt.legend(loc="upper left")
+ fig.autofmt_xdate()
+ plt.tight_layout()
+
+ self.plot_name = f"climFIR_{self._name}_{str(var)}_{it0}_{ifilter}_residuum"
+ self._save()
+ except Exception as e:
+ logging.info(f"Could not create plot because of:\n{sys.exc_info()[0]}\n{sys.exc_info()[1]}\n{sys.exc_info()[2]}")
+ pass
def _set_xlim(self, ax, t0, order, valid_range, td_type, time_axis):
"""
diff --git a/mlair/plotting/postprocessing_plotting.py b/mlair/plotting/postprocessing_plotting.py
index 50c2e6d1fc2ecb641b3fbb861d7ff455d8c2cc97..5fb4e320a1dcf8d3105193dd4de8b7c79b902652 100644
--- a/mlair/plotting/postprocessing_plotting.py
+++ b/mlair/plotting/postprocessing_plotting.py
@@ -491,15 +491,23 @@ class PlotCompetitiveSkillScore(AbstractPlotClass):
"""
+#<<<<<<< HEAD
def __init__(self, data: pd.DataFrame, plot_folder=".", model_setup="NN", sampling="daily",
model_name_for_plots=None):
+#=======
+# def __init__(self, data: Dict[str, pd.DataFrame], plot_folder=".", model_setup="NN"):
+#>>>>>>> develop
"""Initialise."""
super().__init__(plot_folder, f"skill_score_competitive_{model_setup}")
self._model_setup = model_setup
self._sampling = self._get_sampling(sampling)
self._labels = None
+#<<<<<<< HEAD
self._model_name_for_plots = model_name_for_plots
self._data = self._prepare_data(data)
+#=======
+# self._data = self._prepare_data(helpers.remove_items(data, "total"))
+#>>>>>>> develop
default_plot_name = self.plot_name
# draw full detail plot
self.plot_name = default_plot_name + "_full_detail"
@@ -974,13 +982,14 @@ class PlotTimeSeries:
"""
def __init__(self, stations: List, data_path: str, name: str, window_lead_time: int = None, plot_folder: str = ".",
- sampling="daily", model_name="nn", obs_name="obs"):
+ sampling="daily", model_name="nn", obs_name="obs", ahead_dim="ahead"):
"""Initialise."""
self._data_path = data_path
self._data_name = name
self._stations = stations
self._model_name = model_name
self._obs_name = obs_name
+ self._ahead_dim = ahead_dim
self._window_lead_time = self._get_window_lead_time(window_lead_time)
self._sampling = self._get_sampling(sampling)
self._plot(plot_folder)
@@ -1003,7 +1012,7 @@ class PlotTimeSeries:
:param window_lead_time: lead time from arguments to validate
:return: validated lead time, comes either from given argument or from data itself
"""
- ahead_steps = len(self._load_data(self._stations[0]).ahead)
+ ahead_steps = len(self._load_data(self._stations[0]).coords[self._ahead_dim])
if window_lead_time is None:
window_lead_time = ahead_steps
return min(ahead_steps, window_lead_time)
@@ -1025,10 +1034,13 @@ class PlotTimeSeries:
data_year = data.sel(index=f"{start + i_year}")
for i_half_of_year in range(factor):
pos = factor * i_year + i_half_of_year
- plot_data = self._create_plot_data(data_year, factor, i_half_of_year)
- self._plot_obs(axes[pos], plot_data)
- self._plot_ahead(axes[pos], plot_data)
- if np.isnan(plot_data.values).all():
+ try:
+ plot_data = self._create_plot_data(data_year, factor, i_half_of_year)
+ self._plot_obs(axes[pos], plot_data)
+ self._plot_ahead(axes[pos], plot_data)
+ if np.isnan(plot_data.values).all():
+ nan_list.append(pos)
+ except Exception:
nan_list.append(pos)
self._clean_up_axes(nan_list, axes, fig)
self._save_page(station, pdf_pages)
@@ -1065,9 +1077,8 @@ class PlotTimeSeries:
def _plot_ahead(self, ax, data):
color = sns.color_palette("Blues_d", self._window_lead_time).as_hex()
- for ahead in data.coords["ahead"].values:
- plot_data = data.sel(type=self._model_name, ahead=ahead).drop(["type", "ahead"]).squeeze().shift(
- index=ahead)
+ for ahead in data.coords[self._ahead_dim].values:
+ plot_data = data.sel({"type": self._model_name, self._ahead_dim: ahead}).drop(["type", self._ahead_dim]).squeeze().shift(index=ahead)
label = f"{ahead}{self._sampling}"
ax.plot(plot_data, color=color[ahead - 1], label=label)
diff --git a/mlair/run_modules/experiment_setup.py b/mlair/run_modules/experiment_setup.py
index 09d47ef9888ddbbff563a58055d9cea29933c4cb..41c1cd4e69c0e6e4069559ca01affc4e0c8f8462 100644
--- a/mlair/run_modules/experiment_setup.py
+++ b/mlair/run_modules/experiment_setup.py
@@ -21,7 +21,7 @@ from mlair.configuration.defaults import DEFAULT_STATIONS, DEFAULT_VAR_ALL_DICT,
DEFAULT_USE_ALL_STATIONS_ON_ALL_DATA_SETS, DEFAULT_EVALUATE_BOOTSTRAPS, DEFAULT_CREATE_NEW_BOOTSTRAPS, \
DEFAULT_NUMBER_OF_BOOTSTRAPS, DEFAULT_PLOT_LIST, DEFAULT_SAMPLING, DEFAULT_DATA_ORIGIN, DEFAULT_ITER_DIM, \
DEFAULT_USE_MULTIPROCESSING, DEFAULT_USE_MULTIPROCESSING_ON_DEBUG, DEFAULT_MAX_NUMBER_MULTIPROCESSING, \
- DEFAULT_BOOTSTRAP_TYPE, DEFAULT_BOOTSTRAP_METHOD
+ DEFAULT_BOOTSTRAP_TYPE, DEFAULT_BOOTSTRAP_METHOD, DEFAULT_OVERWRITE_LAZY_DATA
from mlair.data_handler import DefaultDataHandler
from mlair.run_modules.run_environment import RunEnvironment
@@ -219,7 +219,8 @@ class ExperimentSetup(RunEnvironment):
hpc_hosts=None, model=None, batch_size=None, epochs=None, data_handler=None,
data_origin: Dict = None, competitors: list = None, competitor_path: str = None,
use_multiprocessing: bool = None, use_multiprocessing_on_debug: bool = None,
- max_number_multiprocessing: int = None, start_script: Union[Callable, str] = None, **kwargs):
+ max_number_multiprocessing: int = None, start_script: Union[Callable, str] = None,
+ overwrite_lazy_data: bool = None, **kwargs):
# create run framework
super().__init__()
@@ -288,6 +289,10 @@ class ExperimentSetup(RunEnvironment):
self._set_param("logging_path", None, os.path.join(experiment_path, "logging"))
path_config.check_path_and_create(self.data_store.get("logging_path"))
+ # set tmp path
+ self._set_param("tmp_path", None, os.path.join(experiment_path, "tmp"))
+ path_config.check_path_and_create(self.data_store.get("tmp_path"), remove_existing=True)
+
# setup for data
self._set_param("stations", stations, default=DEFAULT_STATIONS, apply=helpers.to_list)
self._set_param("statistics_per_var", statistics_per_var, default=DEFAULT_VAR_ALL_DICT)
@@ -298,6 +303,8 @@ class ExperimentSetup(RunEnvironment):
self._set_param("window_history_size", window_history_size, default=DEFAULT_WINDOW_HISTORY_SIZE)
self._set_param("overwrite_local_data", overwrite_local_data, default=DEFAULT_OVERWRITE_LOCAL_DATA,
scope="preprocessing")
+ self._set_param("overwrite_lazy_data", overwrite_lazy_data, default=DEFAULT_OVERWRITE_LAZY_DATA,
+ scope="preprocessing")
self._set_param("transformation", transformation, default={})
self._set_param("transformation", None, scope="preprocessing")
self._set_param("data_handler", data_handler, default=DefaultDataHandler)
diff --git a/mlair/run_modules/post_processing.py b/mlair/run_modules/post_processing.py
index bbe31c2a2246079aac3f5ddd79077097b22fd16c..ac4f19c70ea3be6ce5431d687bcda5d092af572f 100644
--- a/mlair/run_modules/post_processing.py
+++ b/mlair/run_modules/post_processing.py
@@ -6,6 +6,8 @@ __date__ = '2019-12-11'
import inspect
import logging
import os
+import sys
+import traceback
from typing import Dict, Tuple, Union, List, Callable
import keras
@@ -85,8 +87,14 @@ class PostProcessing(RunEnvironment):
self.competitor_path = self.data_store.get("competitor_path")
self.competitors = to_list(self.data_store.get_default("competitors", default=[]))
self.forecast_indicator = "nn"
+#<<<<<<< HEAD
self.model_name_for_plots = self.data_store.get_default("model_name_for_plots", default=None)
+#=======
+ self.observation_indicator = "obs"
+#>>>>>>> develop
self.ahead_dim = "ahead"
+ self.boot_var_dim = "boot_var"
+ self.model_type_dim = "type"
self._run()
def _run(self):
@@ -108,13 +116,16 @@ class PostProcessing(RunEnvironment):
bootstrap_type = self.data_store.get("bootstrap_type", "postprocessing")
self.bootstrap_postprocessing(create_new_bootstraps, bootstrap_type=bootstrap_type,
bootstrap_method=bootstrap_method)
+ if self.bootstrap_skill_scores is not None:
+ self.report_bootstrap_results(self.bootstrap_skill_scores)
# skill scores and error metrics
with TimeTracking(name="calculate skill scores"):
- skill_score_competitive, skill_score_climatological, errors = self.calculate_error_metrics()
+ skill_score_competitive, _, skill_score_climatological, errors = self.calculate_error_metrics()
self.skill_scores = (skill_score_competitive, skill_score_climatological)
self.report_error_metrics(errors)
- self.report_error_metrics(skill_score_climatological)
+ self.report_error_metrics({self.forecast_indicator: skill_score_climatological})
+ self.report_error_metrics({"skill_score": skill_score_competitive})
# plotting
self.plot()
@@ -138,7 +149,7 @@ class PostProcessing(RunEnvironment):
except (FileNotFoundError, KeyError):
logging.debug(f"No competitor found for combination '{station_name}' and '{competitor_name}'.")
continue
- return xr.concat(competing_predictions, "type") if len(competing_predictions) > 0 else None
+ return xr.concat(competing_predictions, self.model_type_dim) if len(competing_predictions) > 0 else None
def bootstrap_postprocessing(self, create_new_bootstraps: bool, _iter: int = 0, bootstrap_type="singleinput",
bootstrap_method="shuffle") -> None:
@@ -154,7 +165,8 @@ class PostProcessing(RunEnvironment):
:param _iter: internal counter to reduce unnecessary recursive calls (maximum number is 2, otherwise something
went wrong).
"""
- self.bootstrap_skill_scores = {}
+ if _iter == 0:
+ self.bootstrap_skill_scores = {}
for boot_type in to_list(bootstrap_type):
self.bootstrap_skill_scores[boot_type] = {}
for boot_method in to_list(bootstrap_method):
@@ -185,7 +197,7 @@ class PostProcessing(RunEnvironment):
# extract all requirements from data store
forecast_path = self.data_store.get("forecast_path")
number_of_bootstraps = self.data_store.get("number_of_bootstraps", "postprocessing")
- dims = ["index", self.ahead_dim, "type"]
+ dims = ["index", self.ahead_dim, self.model_type_dim]
for station in self.test_data:
X, Y = None, None
bootstraps = BootStraps(station, number_of_bootstraps, bootstrap_type=bootstrap_type,
@@ -209,7 +221,7 @@ class PostProcessing(RunEnvironment):
# store also true labels for each station
labels = np.expand_dims(Y, axis=-1)
file_name = os.path.join(forecast_path, f"bootstraps_{station}_{bootstrap_method}_labels.nc")
- labels = xr.DataArray(labels, coords=(*coords, ["obs"]), dims=dims)
+ labels = xr.DataArray(labels, coords=(*coords, [self.observation_indicator]), dims=dims)
labels.to_netcdf(file_name)
def calculate_bootstrap_skill_scores(self, bootstrap_type, bootstrap_method) -> Dict[str, xr.DataArray]:
@@ -245,7 +257,7 @@ class PostProcessing(RunEnvironment):
orig = self.get_orig_prediction(forecast_path, forecast_file % str(station), number_of_bootstraps)
orig = orig.reshape(shape)
coords = (range(shape[0]), range(1, shape[1] + 1), ["orig"])
- orig = xr.DataArray(orig, coords=coords, dims=["index", self.ahead_dim, "type"])
+ orig = xr.DataArray(orig, coords=coords, dims=["index", self.ahead_dim, self.model_type_dim])
# calculate skill scores for each variable
skill = pd.DataFrame(columns=range(1, self.window_lead_time + 1))
@@ -264,7 +276,7 @@ class PostProcessing(RunEnvironment):
skill.loc[boot_var] = np.array(boot_scores)
# collect all results in single dictionary
- score[str(station)] = xr.DataArray(skill, dims=["boot_var", self.ahead_dim])
+ score[str(station)] = xr.DataArray(skill, dims=[self.boot_var_dim, self.ahead_dim])
return score
def get_orig_prediction(self, path, file_name, number_of_bootstraps, prediction_name=None):
@@ -338,7 +350,8 @@ class PostProcessing(RunEnvironment):
PlotSeparationOfScales(self.test_data, plot_folder=self.plot_path, time_dim=time_dim,
window_dim=window_dim, target_dim=target_dim, **{"filter_dim": filter_dim})
except Exception as e:
- logging.error(f"Could not create plot PlotSeparationOfScales due to the following error: {e}")
+ logging.error(f"Could not create plot PlotSeparationOfScales due to the following error:"
+ f"\n{sys.exc_info()[0]}\n{sys.exc_info()[1]}\n{sys.exc_info()[2]}")
try:
if (self.bootstrap_skill_scores is not None) and ("PlotBootstrapSkillScore" in plot_list):
@@ -351,7 +364,8 @@ class PostProcessing(RunEnvironment):
bootstrap_type=boot_type, bootstrap_method=boot_method)
except Exception as e:
logging.error(f"Could not create plot PlotBootstrapSkillScore ({boot_type}, {boot_method}) "
- f"due to the following error: {e}")
+ f"due to the following error:\n{sys.exc_info()[0]}\n{sys.exc_info()[1]}\n"
+ f"{sys.exc_info()[2]}")
except Exception as e:
logging.error(f"Could not create plot PlotBootstrapSkillScore due to the following error: {e}")
@@ -360,14 +374,16 @@ class PostProcessing(RunEnvironment):
if "PlotConditionalQuantiles" in plot_list:
PlotConditionalQuantiles(self.test_data.keys(), data_pred_path=path, plot_folder=self.plot_path)
except Exception as e:
- logging.error(f"Could not create plot PlotConditionalQuantiles due to the following error: {e}")
+ logging.error(f"Could not create plot PlotConditionalQuantiles due to the following error:"
+ f"\n{sys.exc_info()[0]}\n{sys.exc_info()[1]}\n{sys.exc_info()[2]}")
try:
if "PlotMonthlySummary" in plot_list:
PlotMonthlySummary(self.test_data.keys(), path, r"forecasts_%s_test.nc", self.target_var,
plot_folder=self.plot_path)
except Exception as e:
- logging.error(f"Could not create plot PlotMonthlySummary due to the following error: {e}")
+ logging.error(f"Could not create plot PlotMonthlySummary due to the following error:"
+ f"\n{sys.exc_info()[0]}\n{sys.exc_info()[1]}\n{sys.exc_info()[2]}")
try:
if "PlotClimatologicalSkillScore" in plot_list:
@@ -376,7 +392,8 @@ class PostProcessing(RunEnvironment):
PlotClimatologicalSkillScore(self.skill_scores[1], plot_folder=self.plot_path, score_only=False,
extra_name_tag="all_terms_", model_setup=self.forecast_indicator)
except Exception as e:
- logging.error(f"Could not create plot PlotClimatologicalSkillScore due to the following error: {e}")
+ logging.error(f"Could not create plot PlotClimatologicalSkillScore due to the following error: {e}"
+ f"\n{sys.exc_info()[0]}\n{sys.exc_info()[1]}\n{sys.exc_info()[2]}")
try:
if "PlotCompetitiveSkillScore" in plot_list:
@@ -384,14 +401,16 @@ class PostProcessing(RunEnvironment):
model_setup=self.forecast_indicator, sampling=self._sampling,
model_name_for_plots=self.model_name_for_plots)
except Exception as e:
- logging.error(f"Could not create plot PlotCompetitiveSkillScore due to the following error: {e}")
+ logging.error(f"Could not create plot PlotCompetitiveSkillScore due to the following error: {e}"
+ f"\n{sys.exc_info()[0]}\n{sys.exc_info()[1]}\n{sys.exc_info()[2]}")
try:
if "PlotTimeSeries" in plot_list:
PlotTimeSeries(self.test_data.keys(), path, r"forecasts_%s_test.nc", plot_folder=self.plot_path,
- sampling=self._sampling)
+ sampling=self._sampling, ahead_dim=self.ahead_dim)
except Exception as e:
- logging.error(f"Could not create plot PlotTimeSeries due to the following error: {e}")
+ logging.error(f"Could not create plot PlotTimeSeries due to the following error:\n{sys.exc_info()[0]}\n"
+ f"{sys.exc_info()[1]}\n{sys.exc_info()[2]}\n{traceback.format_exc()}")
try:
if "PlotStationMap" in plot_list:
@@ -408,7 +427,8 @@ class PostProcessing(RunEnvironment):
(self.test_data, {"marker": 9, "ms": 9})]
PlotStationMap(generators=gens, plot_folder=self.plot_path, plot_name="station_map_var")
except Exception as e:
- logging.error(f"Could not create plot PlotStationMap due to the following error: {e}")
+ logging.error(f"Could not create plot PlotStationMap due to the following error: {e}"
+ f"\n{sys.exc_info()[0]}\n{sys.exc_info()[1]}\n{sys.exc_info()[2]}")
try:
if "PlotAvailability" in plot_list:
@@ -416,7 +436,8 @@ class PostProcessing(RunEnvironment):
PlotAvailability(avail_data, plot_folder=self.plot_path, time_dimension=time_dim,
window_dimension=window_dim)
except Exception as e:
- logging.error(f"Could not create plot PlotAvailability due to the following error: {e}")
+ logging.error(f"Could not create plot PlotAvailability due to the following error: {e}"
+ f"\n{sys.exc_info()[0]}\n{sys.exc_info()[1]}\n{sys.exc_info()[2]}")
try:
if "PlotAvailabilityHistogram" in plot_list:
@@ -424,7 +445,8 @@ class PostProcessing(RunEnvironment):
PlotAvailabilityHistogram(avail_data, plot_folder=self.plot_path, station_dim=iter_dim,
history_dim=window_dim)
except Exception as e:
- logging.error(f"Could not create plot PlotAvailabilityHistogram due to the following error: {e}")
+ logging.error(f"Could not create plot PlotAvailabilityHistogram due to the following error: {e}"
+ f"\n{sys.exc_info()[0]}\n{sys.exc_info()[1]}\n{sys.exc_info()[2]}")
try:
if "PlotPeriodogram" in plot_list:
@@ -432,14 +454,18 @@ class PostProcessing(RunEnvironment):
variables_dim=target_dim, sampling=self._sampling,
use_multiprocessing=use_multiprocessing)
except Exception as e:
- logging.error(f"Could not create plot PlotPeriodogram due to the following error: {e}")
+ logging.error(f"Could not create plot PlotPeriodogram due to the following error: {e}"
+ f"\n{sys.exc_info()[0]}\n{sys.exc_info()[1]}\n{sys.exc_info()[2]}")
try:
if "PlotDataHistogram" in plot_list:
+ upsampling = self.data_store.get_default("upsampling", scope="train", default=False)
gens = {"train": self.train_data, "val": self.val_data, "test": self.test_data}
- PlotDataHistogram(gens, plot_folder=self.plot_path, time_dim=time_dim, variables_dim=target_dim)
+ PlotDataHistogram(gens, plot_folder=self.plot_path, time_dim=time_dim, variables_dim=target_dim,
+ upsampling=upsampling)
except Exception as e:
- logging.error(f"Could not create plot PlotDataHistogram due to the following error: {e}")
+ logging.error(f"Could not create plot PlotDataHistogram due to the following error: {e}"
+ f"\n{sys.exc_info()[0]}\n{sys.exc_info()[1]}\n{sys.exc_info()[2]}")
def calculate_test_score(self):
"""Evaluate test score of model and save locally."""
@@ -500,7 +526,7 @@ class PostProcessing(RunEnvironment):
full_index = self.create_fullindex(observation_data.indexes[time_dimension], self._get_frequency())
prediction_dict = {self.forecast_indicator: nn_prediction,
"persi": persistence_prediction,
- "obs": observation,
+ self.observation_indicator: observation,
"ols": ols_prediction}
all_predictions = self.create_forecast_arrays(full_index, list(target_data.indexes[window_dim]),
time_dimension, ahead_dim=self.ahead_dim,
@@ -533,7 +559,7 @@ class PostProcessing(RunEnvironment):
with xr.open_dataarray(file) as da:
data = da.load()
forecast = data.sel(type=[self.forecast_indicator])
- forecast.coords["type"] = [competitor_name]
+ forecast.coords[self.model_type_dim] = [competitor_name]
return forecast
def _create_observation(self, data, _, transformation_func: Callable, normalised: bool) -> xr.DataArray:
@@ -715,18 +741,19 @@ class PostProcessing(RunEnvironment):
except (IndexError, KeyError, FileNotFoundError):
return None
- @staticmethod
- def _combine_forecasts(forecast, competitor, dim="type"):
+ def _combine_forecasts(self, forecast, competitor, dim=None):
"""
Combine forecast and competitor if both are xarray. If competitor is None, this returns forecasts and vise
versa.
"""
+ if dim is None:
+ dim = self.model_type_dim
try:
return xr.concat([forecast, competitor], dim=dim)
except (TypeError, AttributeError):
return forecast if competitor is None else competitor
- def calculate_error_metrics(self) -> Tuple[Dict, Dict, Dict]:
+ def calculate_error_metrics(self) -> Tuple[Dict, Dict, Dict, Dict]:
"""
Calculate error metrics and skill scores of NN forecast.
@@ -739,6 +766,7 @@ class PostProcessing(RunEnvironment):
path = self.data_store.get("forecast_path")
all_stations = self.data_store.get("stations")
skill_score_competitive = {}
+ skill_score_competitive_count = {}
skill_score_climatological = {}
errors = {}
for station in all_stations:
@@ -746,24 +774,61 @@ class PostProcessing(RunEnvironment):
# test errors
if external_data is not None:
- errors[station] = statistics.calculate_error_metrics(*map(lambda x: external_data.sel(type=x),
- [self.forecast_indicator, "obs"]),
- dim="index")
- # skill score
+ model_type_list = external_data.coords[self.model_type_dim].values.tolist()
+ for model_type in remove_items(model_type_list, self.observation_indicator):
+ if model_type not in errors.keys():
+ errors[model_type] = {}
+ errors[model_type][station] = statistics.calculate_error_metrics(
+ *map(lambda x: external_data.sel(**{self.model_type_dim: x}),
+ [model_type, self.observation_indicator]), dim="index")
+
+ # load competitors
competitor = self.load_competitors(station)
- combined = self._combine_forecasts(external_data, competitor, dim="type")
- model_list = remove_items(list(combined.type.values), "obs") if combined is not None else None
+ combined = self._combine_forecasts(external_data, competitor, dim=self.model_type_dim)
+ if combined is not None:
+ model_list = remove_items(combined.coords[self.model_type_dim].values.tolist(),
+ self.observation_indicator)
+ else:
+ model_list = None
+
+ # test errors of competitors
+ for model_type in remove_items(model_list or [], list(errors.keys())):
+ if self.observation_indicator not in combined.coords[self.model_type_dim]:
+ continue
+ if model_type not in errors.keys():
+ errors[model_type] = {}
+ errors[model_type][station] = statistics.calculate_error_metrics(
+ *map(lambda x: combined.sel(**{self.model_type_dim: x}),
+ [model_type, self.observation_indicator]), dim="index")
+
+ # skill score
skill_score = statistics.SkillScores(combined, models=model_list, ahead_dim=self.ahead_dim)
if external_data is not None:
- skill_score_competitive[station] = skill_score.skill_scores()
+ skill_score_competitive[station], skill_score_competitive_count[station] = skill_score.skill_scores()
internal_data = self._get_internal_data(station, path)
if internal_data is not None:
skill_score_climatological[station] = skill_score.climatological_skill_scores(
internal_data, forecast_name=self.forecast_indicator)
- errors.update({"total": self.calculate_average_errors(errors)})
- return skill_score_competitive, skill_score_climatological, errors
+ for model_type in errors.keys():
+ errors[model_type].update({"total": self.calculate_average_errors(errors[model_type])})
+ skill_score_competitive.update({"total": self.calculate_average_skill_scores(skill_score_competitive,
+ skill_score_competitive_count)})
+ return skill_score_competitive, skill_score_competitive_count, skill_score_climatological, errors
+
+ @staticmethod
+ def calculate_average_skill_scores(scores, counts):
+ avg_skill_score = 0
+ n_total = None
+ for vals in counts.values():
+ n_total = vals if n_total is None else n_total.add(vals, fill_value=0)
+ for station, station_scores in scores.items():
+ n = counts.get(station)
+ avg_skill_score = station_scores.mul(n.div(n_total, fill_value=0), fill_value=0).add(avg_skill_score,
+ fill_value=0)
+ return avg_skill_score
+
@staticmethod
def calculate_average_errors(errors):
@@ -776,28 +841,55 @@ class PostProcessing(RunEnvironment):
avg_error[error_metric] = new_val
return avg_error
+ def report_bootstrap_results(self, results):
+ """Create a csv file containing all results from bootstrapping."""
+ report_path = os.path.join(self.data_store.get("experiment_path"), "latex_report")
+ path_config.check_path_and_create(report_path)
+ res = [[self.model_type_dim, "method", "station", self.boot_var_dim, self.ahead_dim, "vals"]]
+ for boot_type, d0 in results.items():
+ for boot_method, d1 in d0.items():
+ for station_name, vals in d1.items():
+ for boot_var in vals.coords[self.boot_var_dim].values.tolist():
+ for ahead in vals.coords[self.ahead_dim].values.tolist():
+ res.append([boot_type, boot_method, station_name, boot_var, ahead,
+ float(vals.sel({self.boot_var_dim: boot_var, self.ahead_dim: ahead}))])
+ col_names = res.pop(0)
+ df = pd.DataFrame(res, columns=col_names)
+ file_name = "bootstrap_skill_score_report_raw.csv"
+ df.to_csv(os.path.join(report_path, file_name), sep=";")
+
def report_error_metrics(self, errors):
report_path = os.path.join(self.data_store.get("experiment_path"), "latex_report")
path_config.check_path_and_create(report_path)
- metric_collection = {}
- for station, station_errors in errors.items():
- if isinstance(station_errors, xr.DataArray):
- dim = station_errors.dims[0]
- sel_index = [sel for sel in station_errors.coords[dim] if "CASE" in str(sel)]
- station_errors = {str(i.values): station_errors.sel(**{dim: i}) for i in sel_index}
- for metric, vals in station_errors.items():
- if metric == "n":
- continue
- pd_vals = pd.DataFrame.from_dict({station: vals}).T
- pd_vals.columns = [f"{metric}(t+{x})" for x in vals.coords["ahead"].values]
- mc = metric_collection.get(metric, pd.DataFrame())
- mc = mc.append(pd_vals)
- metric_collection[metric] = mc
- for metric, error_df in metric_collection.items():
- df = error_df.sort_index()
- if "total" in df.index:
- df.reindex(df.index.drop(["total"]).to_list() + ["total"], )
- column_format = tables.create_column_format_for_tex(df)
- file_name = f"error_report_{metric}.%s".replace(' ', '_')
- tables.save_to_tex(report_path, file_name % "tex", column_format=column_format, df=df)
- tables.save_to_md(report_path, file_name % "md", df=df)
+ for model_type in errors.keys():
+ metric_collection = {}
+ for station, station_errors in errors[model_type].items():
+ if isinstance(station_errors, xr.DataArray):
+ dim = station_errors.dims[0]
+ sel_index = [sel for sel in station_errors.coords[dim] if "CASE" in str(sel)]
+ station_errors = {str(i.values): station_errors.sel(**{dim: i}) for i in sel_index}
+ elif isinstance(station_errors, pd.DataFrame):
+ sel_index = station_errors.index.tolist()
+ ahead = station_errors.columns.values
+ station_errors = {k: xr.DataArray(station_errors[station_errors.index == k].values.flatten(),
+ dims=["ahead"], coords={"ahead": ahead}).astype(float)
+ for k in sel_index}
+ for metric, vals in station_errors.items():
+ if metric == "n":
+ metric = "count"
+ pd_vals = pd.DataFrame.from_dict({station: vals}).T
+ pd_vals.columns = [f"{metric}(t+{x})" for x in vals.coords["ahead"].values]
+ mc = metric_collection.get(metric, pd.DataFrame())
+ mc = mc.append(pd_vals)
+ metric_collection[metric] = mc
+ for metric, error_df in metric_collection.items():
+ df = error_df.sort_index()
+ if "total" in df.index:
+ df.reindex(df.index.drop(["total"]).to_list() + ["total"], )
+ column_format = tables.create_column_format_for_tex(df)
+ if model_type == "skill_score":
+ file_name = f"error_report_{model_type}_{metric}.%s".replace(' ', '_')
+ else:
+ file_name = f"error_report_{metric}_{model_type}.%s".replace(' ', '_')
+ tables.save_to_tex(report_path, file_name % "tex", column_format=column_format, df=df)
+ tables.save_to_md(report_path, file_name % "md", df=df)
diff --git a/mlair/run_modules/pre_processing.py b/mlair/run_modules/pre_processing.py
index db9e9e940f7675e618e48f41f0cffb4e4d9ac644..f0192e469c0a52a68fc5b29a5254583d5207cc32 100644
--- a/mlair/run_modules/pre_processing.py
+++ b/mlair/run_modules/pre_processing.py
@@ -11,6 +11,8 @@ import multiprocessing
import requests
import psutil
import traceback
+import random
+import dill
import pandas as pd
@@ -246,6 +248,7 @@ class PreProcessing(RunEnvironment):
valid_stations = []
kwargs = self.data_store.create_args_dict(data_handler.requirements(), scope=set_name)
use_multiprocessing = self.data_store.get("use_multiprocessing")
+ tmp_path = self.data_store.get("tmp_path")
max_process = self.data_store.get("max_number_multiprocessing")
n_process = min([psutil.cpu_count(logical=False), len(set_stations), max_process]) # use only physical cpus
@@ -253,18 +256,23 @@ class PreProcessing(RunEnvironment):
logging.info("use parallel validate station approach")
pool = multiprocessing.Pool(n_process)
logging.info(f"running {getattr(pool, '_processes')} processes in parallel")
+ kwargs.update({"tmp_path": tmp_path, "return_strategy": "reference"})
output = [
pool.apply_async(f_proc, args=(data_handler, station, set_name, store_processed_data), kwds=kwargs)
for station in set_stations]
for i, p in enumerate(output):
- dh, s = p.get()
+ _res_file, s = p.get()
logging.info(f"...finished: {s} ({int((i + 1.) / len(output) * 100)}%)")
+ with open(_res_file, "rb") as f:
+ dh = dill.load(f)
+ os.remove(_res_file)
if dh is not None:
collection.add(dh)
valid_stations.append(s)
pool.close()
else: # serial solution
logging.info("use serial validate station approach")
+ kwargs.update({"return_strategy": "result"})
for station in set_stations:
dh, s = f_proc(data_handler, station, set_name, store_processed_data, **kwargs)
if dh is not None:
@@ -272,7 +280,7 @@ class PreProcessing(RunEnvironment):
valid_stations.append(s)
logging.info(f"run for {t_outer} to check {len(set_stations)} station(s). Found {len(collection)}/"
- f"{len(set_stations)} valid stations.")
+ f"{len(set_stations)} valid stations ({set_name}).")
if set_name == "train":
self.store_data_handler_attributes(data_handler, collection)
return collection, valid_stations
@@ -292,7 +300,8 @@ class PreProcessing(RunEnvironment):
def transformation(self, data_handler: AbstractDataHandler, stations):
if hasattr(data_handler, "transformation"):
kwargs = self.data_store.create_args_dict(data_handler.requirements(), scope="train")
- transformation_dict = data_handler.transformation(stations, **kwargs)
+ tmp_path = self.data_store.get_default("tmp_path", default=None)
+ transformation_dict = data_handler.transformation(stations, tmp_path=tmp_path, **kwargs)
if transformation_dict is not None:
self.data_store.set("transformation", transformation_dict)
@@ -316,12 +325,13 @@ class PreProcessing(RunEnvironment):
logging.info("No preparation required because no competitor was provided to the workflow.")
-def f_proc(data_handler, station, name_affix, store, **kwargs):
+def f_proc(data_handler, station, name_affix, store, return_strategy="", tmp_path=None, **kwargs):
"""
Try to create a data handler for given arguments. If build fails, this station does not fulfil all requirements and
- therefore f_proc will return None as indication. On a successfull build, f_proc returns the built data handler and
+ therefore f_proc will return None as indication. On a successful build, f_proc returns the built data handler and
the station that was used. This function must be implemented globally to work together with multiprocessing.
"""
+ assert return_strategy in ["result", "reference"]
try:
res = data_handler.build(station, name_affix=name_affix, store_processed_data=store, **kwargs)
except (AttributeError, EmptyQueryResult, KeyError, requests.ConnectionError, ValueError, IndexError) as e:
@@ -330,7 +340,15 @@ def f_proc(data_handler, station, name_affix, store, **kwargs):
#f"remove station {station} because it raised an error: {e} -> {' | '.join(f_inspect_error(formatted_lines))}")
f"remove station {station} because it raised an error: {e} -> {' | '.join(formatted_lines)}")
res = None
- return res, station
+ if return_strategy == "result":
+ return res, station
+ else:
+ if tmp_path is None:
+ tmp_path = os.getcwd()
+ _tmp_file = os.path.join(tmp_path, f"{station}_{'%032x' % random.getrandbits(128)}.pickle")
+ with open(_tmp_file, "wb") as f:
+ dill.dump(res, f, protocol=4)
+ return _tmp_file, station
def f_inspect_error(formatted):
diff --git a/run_climate_filter.py b/run_climate_filter.py
new file mode 100755
index 0000000000000000000000000000000000000000..5577375e2fc135676f71151791c1d564dcb25a2e
--- /dev/null
+++ b/run_climate_filter.py
@@ -0,0 +1,91 @@
+__author__ = "Lukas Leufen"
+__date__ = '2019-11-14'
+
+import argparse
+
+from mlair.workflows import DefaultWorkflow
+from mlair.data_handler.data_handler_mixed_sampling import DataHandlerMixedSamplingWithClimateFirFilter
+from mlair.model_modules.fully_connected_networks import BranchedInputFCN as NN
+from mlair.configuration.defaults import DEFAULT_PLOT_LIST, DEFAULT_TRAIN_END
+
+
+STATS = {'o3': 'dma8eu', 'relhum': 'average_values', 'temp': 'maximum', 'u': 'average_values',
+ 'v': 'average_values', 'no': 'dma8eu', 'no2': 'dma8eu', 'cloudcover': 'average_values',
+ 'pblheight': 'maximum'}
+
+DATA_ORIGIN = {"no": "", "no2": "", "o3": "",
+ "cloudcover": "REA", "pblheight": "REA", "relhum": "REA",
+ "temp": "REA", "u": "REA", "v": "REA"}
+
+
+def main(parser_args):
+ # plots = remove_items(DEFAULT_PLOT_LIST, "PlotConditionalQuantiles")
+ args = dict(
+ sampling=("hourly", "daily"), # T1A, T1D
+ stations=["DEBW107", "DEBW013"], # [:5], # T1B, TODO: remove indexing for meaningful experiments
+ network="UBA", # T1B
+ variables=["o3", "no", "no2", "cloudcover", "pblheight", "relhum", "temp", "u", "v"], # T1B
+ statistics_per_var=STATS, # T1A, T1D, T1F
+
+ data_origin=DATA_ORIGIN,
+ # data_handler=DataHandlerMixedSampling,
+ # window_history_offset=16,
+ # window_history_size=6 * 24 + 16, # T1D
+ data_handler=DataHandlerMixedSamplingWithClimateFirFilter,
+ filter_cutoff_period=[21],
+ filter_order=[42],
+ filter_window_type=("kaiser", 5),
+ filter_add_unfiltered=True,
+ apriori_sel_opts=slice(DEFAULT_TRAIN_END),
+ apriori_type="residuum_stats",
+ apriori_diurnal=True,
+ use_filter_branches=True,
+
+ window_history_size=2 * 24 + 16,
+ window_history_offset=16, # T1F
+ window_lead_time=4, # T1D
+ target_var="o3", # T1D
+ interpolation_limit=(24, 2), # T1F
+ transformation={
+ "o3": {"method": "log"},
+ "no": {"method": "log"},
+ "no2": {"method": "log"},
+ "cloudcover": {"method": "min_max", "feature_range": [-1, 1]},
+ "pblheight": {"method": "log"},
+ "relhum": {"method": "min_max", "feature_range": [-1, 1]},
+ "temp": {"method": "standardise"},
+ "u": {"method": "standardise"},
+ "v": {"method": "standardise"}, }, # T1F, also apply same target transformation
+
+ start="2006-01-01",
+ train_start="2006-01-01",
+ end="2011-12-31",
+ test_end="2011-12-31",
+
+ train_model=False, create_new_model=True,
+ epochs=1,
+ model=NN,
+ evaluate_bootstraps=False,
+ bootstrap_type=["singleinput", "branch", "variable"],
+ bootstrap_method=["shuffle", "zero_mean"],
+ plot_list=DEFAULT_PLOT_LIST,
+ # competitors=["IntelliO3-ts-v1", "MFCN_64_32_16_climFIR", "FCN_1449_512_32_4"], # "FCN_1449_16_8_4",],
+ lazy_preprocessing=True,
+ use_multiprocessing=True,
+ use_multiprocessing_on_debug=False,
+ overwrite_local_data=False,
+ overwrite_lazy_data=False,
+ max_number_multiprocessing=2,
+ **parser_args.__dict__
+ )
+ print(parser_args.__dict__)
+ workflow = DefaultWorkflow(**args, start_script=__file__)
+ workflow.run()
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser()
+ parser.add_argument('--experiment_date', metavar='--exp_date', type=str, default=None,
+ help="set experiment date as string")
+ args = parser.parse_args(["--experiment_date", "testrun"])
+ main(args)