eFlows4HPC WP2
ModelRepository

Repository

import mlair

# just give it a dry run without any modification
mlair.run()
INFO: DefaultWorkflow started
INFO: ExperimentSetup started
INFO: Experiment path is: /home/<usr>/mlair/testrun_network
...
INFO: load data for DEBW107 from JOIN
INFO: load data for DEBY081 from JOIN
INFO: load data for DEBW013 from JOIN
INFO: load data for DEBW076 from JOIN
INFO: load data for DEBW087 from JOIN
...
INFO: Training started
...
INFO: DefaultWorkflow finished after 0:03:04 (hh:mm:ss)
import mlair

# our new stations to use
stations = ['DEBW030', 'DEBW037', 'DEBW031', 'DEBW015', 'DEBW107']

# expanded temporal context to 14 (days, because of default sampling="daily")
window_history_size = 14

# restart the experiment with little customisation
mlair.run(stations=stations,
          window_history_size=window_history_size)
INFO: DefaultWorkflow started
INFO: ExperimentSetup started
...
INFO: load data for DEBW030 from JOIN
INFO: load data for DEBW037 from JOIN
INFO: load data for DEBW031 from JOIN
INFO: load data for DEBW015 from JOIN
...
INFO: Training started
...
INFO: DefaultWorkflow finished after 00:02:03 (hh:mm:ss)
import mlair

# our new stations to use
stations = ['DEBY002', 'DEBY079']

# same setting for window_history_size
window_history_size = 14

# run experiment without training
mlair.run(stations=stations,
          window_history_size=window_history_size,
          create_new_model=False,
          train_model=False)
INFO: DefaultWorkflow started
...
INFO: No training has started, because train_model parameter was false.
...
INFO: DefaultWorkflow finished after 0:01:27 (hh:mm:ss)
import mlair

# create your custom MLAir workflow
DefaultWorkflow = mlair.DefaultWorkflow()
# execute default workflow
DefaultWorkflow.run()
INFO: DefaultWorkflow started
INFO: ExperimentSetup started
...
INFO: ExperimentSetup finished after 00:00:01 (hh:mm:ss)
INFO: PreProcessing started
...
INFO: PreProcessing finished after 00:00:11 (hh:mm:ss)
INFO: ModelSetup started
...
INFO: ModelSetup finished after 00:00:01 (hh:mm:ss)
INFO: Training started
...
INFO: Training finished after 00:02:15 (hh:mm:ss)
INFO: PostProcessing started
...
INFO: PostProcessing finished after 00:01:37 (hh:mm:ss)
INFO: DefaultWorkflow finished after 00:04:05 (hh:mm:ss)
import mlair
import logging

class CustomStage(mlair.RunEnvironment):
    """A custom MLAir stage for demonstration."""

    def __init__(self, test_string):
        super().__init__()  # always call super init method
        self._run(test_string)  # call a class method

    def _run(self, test_string):
        logging.info("Just running a custom stage.")
        logging.info("test_string = " + test_string)
        epochs = self.data_store.get("epochs")
        logging.info("epochs = " + str(epochs))
# create your custom MLAir workflow
CustomWorkflow = mlair.Workflow()
# provide stages without initialisation
CustomWorkflow.add(mlair.ExperimentSetup, epochs=128)
# add also keyword arguments for a specific stage
CustomWorkflow.add(CustomStage, test_string="Hello World")
# finally execute custom workflow in order of adding
CustomWorkflow.run()
INFO: Workflow started
...
INFO: ExperimentSetup finished after 00:00:12 (hh:mm:ss)
INFO: CustomStage started
INFO: Just running a custom stage.
INFO: test_string = Hello World
INFO: epochs = 128
INFO: CustomStage finished after 00:00:01 (hh:mm:ss)
INFO: Workflow finished after 00:00:13 (hh:mm:ss)
from mlair import AbstractModelClass

class MyCustomisedModel(AbstractModelClass):

    def __init__(self, input_shape: list, output_shape: list):

        # set attributes shape_inputs and shape_outputs
        super().__init__(input_shape[0], output_shape[0])

        # apply to model
        self.set_model()
        self.set_compile_options()
        self.set_custom_objects(loss=self.compile_options['loss'])
import tensorflow.keras as keras
from tensorflow.keras.layers import PReLU, Input, Conv2D, Flatten, Dropout, Dense

class MyCustomisedModel(AbstractModelClass):

    def set_model(self):
        x_input = Input(shape=self._input_shape)
        x_in = Conv2D(4, (1, 1))(x_input)
        x_in = PReLU()(x_in)
        x_in = Flatten()(x_in)
        x_in = Dropout(0.1)(x_in)
        x_in = Dense(16)(x_in)
        x_in = PReLU()(x_in)
        x_in = Dense(self._output_shape)(x_in)
        out = PReLU()(x_in)
        self.model = keras.Model(inputs=x_input, outputs=[out])
from tensorflow.keras.losses import mean_squared_error as mse

class MyCustomisedModel(AbstractModelClass):

    def set_compile_options(self):
        self.initial_lr = 1e-2
        self.optimizer = keras.optimizers.SGD(lr=self.initial_lr, momentum=0.9)
        self.loss = mse
        self.compile_options = {"metrics": ["mse", "mae"]}
def set_compile_options(self):
    self.optimizer = keras.optimizers.SGD()
    self.loss = keras.losses.mean_squared_error
    self.compile_options = {"optimizer" = keras.optimizers.Adam()}
from mlair.workflows import DefaultWorkflow

workflow = DefaultWorkflow(model=MyCustomisedModel)
workflow.run()
class MyCustomisedModel(AbstractModelClass):

    def set_model(self):
        ...
        self.model = keras.Model(inputs=x_input, outputs=[out_minor_1, out_minor_2, out_main])

    def set_compile_options(self):
        self.loss = [keras.losses.mean_absolute_error] +  # for out_minor_1
                    [keras.losses.mean_squared_error] +   # for out_minor_2
                    [keras.losses.mean_squared_error]     # for out_main
>>> MyCustomisedModel().model
<your custom model>
>>> MyCustomisedModel().loss
<your custom loss>
>>> MyCustomisedModel().model.compile(**kwargs) == MyCustomisedModel().compile(**kwargs)
True
import datetime as dt
import numpy as np
import pandas as pd
import xarray as xr

from mlair.data_handler import AbstractDataHandler

class DummyDataHandler(AbstractDataHandler):

    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
        number between 100 and 150 is set."""
        super().__init__()
        self.name = name
        self.number_of_samples = number_of_samples if number_of_samples is not None else np.random.randint(100, 150)
        self._X = self.create_X()
        self._Y = self.create_Y()

    def create_X(self):
        """Inputs are random numbers between 0 and 10 with shape (no_samples, window=14, variables=5)."""
        X = np.random.randint(0, 10, size=(self.number_of_samples, 14, 5))  # samples, window, variables
        datelist = pd.date_range(dt.datetime.today().date(), periods=self.number_of_samples, freq="H").tolist()
        return xr.DataArray(X, dims=['datetime', 'window', 'variables'], coords={"datetime": datelist,
                                                                                 "window": range(14),
                                                                                 "variables": range(5)})

    def create_Y(self):
        """Targets are normal distributed random numbers with shape (no_samples, window=5, variables=1)."""
        Y = np.round(0.5 * np.random.randn(self.number_of_samples, 5, 1), 1)  # samples, window, variables
        datelist = pd.date_range(dt.datetime.today().date(), periods=self.number_of_samples, freq="H").tolist()
        return xr.DataArray(Y, dims=['datetime', 'window', 'variables'], coords={"datetime": datelist,
                                                                                 "window": range(5),
                                                                                 "variables": range(1)})

    def get_X(self, upsampling=False, as_numpy=False):
        """Upsampling parameter is not used for X."""
        return np.copy(self._X) if as_numpy is True else self._X

    def get_Y(self, upsampling=False, as_numpy=False):
        """Upsampling parameter is not used for Y."""
        return np.copy(self._Y) if as_numpy is True else self._Y

    def __str__(self):
        return self.name

from mlair.helpers import remove_items
from mlair.configuration.defaults import DEFAULT_PLOT_LIST

mlair.run(plot_list=remove_items(DEFAULT_PLOT_LIST, "PlotStationMap"))

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",)'