A dataflow language for deep weather forecasting: Riverflow | Formatting

RiverFlow data loader, part 1 of 2. Next: From columns to training pairs.

RiverFlow is a project which aims to forecast the water level of rivers in Northern Italy from their own recent history and from the weather upstream. The river data is an hourly sensor export labelled "Sesia @ Campertogno". The weather comes from eight stations in the valley, one daily CSV each: Cellio, Carcoforo, Lozzolo, Rima, Bocchetta delle Pisse, Alagna, Rassa and Boccioleto. No two station files have the same columns. The headers are in Italian, the decimals use commas (2,6), and the dates are written 11/08/2006 while the river uses 2018-03-22 02:00:00.

Loading this with one Python script per experiment works for about a week. After that, the decisions that shape the dataset are spread across files: which columns count, how a gap is filled, what gets normalized, how far back a window reaches. So the repository has a small language for it instead. One file describes the whole path from raw CSV to a saved .npz of training pairs, and an interpreter written in Python runs it.

This series walks through that language using the code in the RiverFlow repository [1]. This first part covers the four sections that load and transform variables. Part 2 covers the plans that turn variables into samples, and part 3 covers the assumptions the interpreter makes about the data it receives.

A file reads top to bottom, in five sections

Every file (the examples use the extension .makefile, although GNU make is not involved) has five sections, always in this order: .decl declares sources, .res resolves types, .act transforms variables, .sap saves intermediate files and .make builds datasets. The interpreter's entry point is a straight pipeline, one method per section:

def interpret(self):
    self.create_data()

    self.parse_part_one()  # .decl
    self.parse_part_two()  # .res
    self.act()  # .act
    self.parse_sap()  # .sap
    self.parse_make()  # .make

create_data() reads the file and removes blank lines and lines that start with #. Each later stage locates its section with self.rows.index('.res\n') and similar calls, so every marker has to exist on its own line, even when the section is empty. The Iris example shows all five in about thirty lines (the paths are placeholders in the repository):

.decl
source_file IrisData = "Your/Shuffled/Data/File"
{SLength},{SWidth},{PLength},{PWidth},{Category}

.res
SLength: numeric
SWidth: numeric
PLength: numeric
PWidth: numeric
Category: categorical

.act
import numpy as np

new length = np.size(SLength)
new Indexes = np.arange(0, length)

SLength = SLength - media(SLength)
SWidth = SWidth - media(SWidth)
PLength = PLength - media(PLength)
PWidth = PWidth - media(PWidth)

new Categories = da_categorico_a_numero(Category, {'Iris-setosa':0, 'Iris-versicolor':1, 'Iris-virginica':2})

.sap

save_file SaveScaled = "/Save/path"
save "{SLength},{SWidth},{PLength},{PWidth}" into SaveScaled

.make
...

The order is also the order of commitment. After .decl the interpreter knows which variables exist, after .res it knows what they are, and after .act their values are final. The rest of this post follows the same order.

Declarations describe a row, not a schema

A declaration is always two lines. The first names a source and gives its path. The second is a format string that describes one row of the file:

source_file MeteoCellio = EXAMPLESROOT + '/Meteo/CELLIO_giornalieri_2006_2022Pad.csv'
{DataC};{PNoveC};{PZeroC};{TempAvgC};{TempMaxC};{TempMinC};

source_file RiverFile = EXAMPLESROOT + '/River Height/sesia-hourly-packed-padded-aligned.csv'
{RData};{VectorRH}

The path after = is a Python expression. The interpreter evaluates it against the module dictionary of Config.py, which is where EXAMPLESROOT comes from:

path_statement = statement.split('=')[1].strip()
file_path = eval(path_statement, FileDeclGetter.glob_env)

The format string is split on braces. Names sit at odd positions and the literal text between them sits at even positions:

split_arg_list = [a for arg in arg_list.split('{') for a in arg.split('}')]

variable_names = split_arg_list[1::2]

self.formats[file_label] = split_arg_list[::2]
self.files_arglists[file_label] = variable_names

For {DataC};{PNoveC}; this gives the names DataC and PNoveC, and the separators '', ';' and ';'. When a row is parsed, each separator is matched literally and the text before it becomes a value. The final separator decides what happens to the rest of the line. The Cellio file ends every row with a semicolon, so the format string ends with one too. The river file stores a whole bracketed list of 24 hourly readings in its second column, [14.0, 12.0, 13.0, ...]. Its format string ends with an empty separator, so the parser takes everything after the first semicolon as one value, commas included.

This is why the format string is more useful than a column list. It describes the row the way it looks on disk, and one parser handles semicolon files, comma files and embedded vectors. The Iris declaration uses commas, the station files use semicolons, and neither needs a flag.

Headers are handled by guessing. Before parsing, the interpreter reads two lines and asks csv.Sniffer().has_header() whether the first one looks like a header. The Python documentation describes this method as a heuristic [2], and it can be wrong for files whose first data row looks like text. None of the station files in the repository caused problems, but a declaration has no way to state "this file has a header", which would be more reliable.

Resolution gives every column a type

After .decl, all values are strings. The .res section assigns each variable one of four types, and the interpreter maps them onto NumPy dtypes:

recognized_v_types = {
    "categorical": str,
    "boolean": str,
    "numeric": np.float64,
    "integer": np.int32
}

A variable that is declared but not resolved keeps the placeholder type "N/A", and loading it raises a TypeError. The large river example therefore types every column, including the ones it never uses. The file even labels them for the reader:

# ************ Fiume ************
RData: categorical
VectorRH: categorical

# > Date di ogni meteo
# !NOT PRESENT IN MODEL!
DataC: categorical
...
PNoveC: numeric
PZeroC: numeric

Dates and the river vectors are categorical on purpose. A date stays a string so that it can be compared later, and the river vector stays a string until .act parses it. .res also accepts copies such as Backup = PNoveC, which make a deep copy of an already resolved variable.

The type conversion has one silent rule that matters for weather data. If a column does not convert, the interpreter replaces empty strings with '0' and tries again:

try:
    self.variables[var_name] = np.array(py_data).astype(as_type)
# Solve missing datapoint by just putting them equal to zero.
except ValueError:
    intermediate = np.array(py_data)
    # Naive attempt to counter missing data, should be more modular
    # in this sense.
    intermediate[intermediate == ''] = '0'

For a rainfall column, zero is a valid measurement, so a missing day becomes a dry day and nothing downstream can tell the difference. The comment in the code already flags this as naive. In practice the repository avoids the rule by padding the station files before the language sees them, with the Pad.csv files shown above. Part 3 looks at that padding in detail.

Actions are Python in a controlled namespace

The .act section is where the language stops being declarative. Each line is a Python expression that the interpreter runs with eval() in an environment that contains the resolved variables as NumPy arrays and a set of primitives. Most of the primitives have Italian names:

self.env = {
    # Primitive functions
    "discretizza": vec_discrete,
    "aggiungi_rumore": vec_add_noise,
    "da_categorico_a_numero": vec_bool_to_num,
    "shuffle": vec_shuffle,
    "media_zero": vec_zero_mean,
    "media": vec_mean,
    ...
    "one_hot_encode": vec_one_hot,
    "load_vec": vec_load,
    "z_score":  vec_zscore,
    "max_linear": vec_max_linear,
    ...
}
# Enhance environment with user-defined variables.
self.env.update(self.variables)

Three rules keep this from turning into an ordinary script. First, a new name needs the keyword new. Assigning to an unknown name raises NameError on purpose, and the error handler uses difflib.get_close_matches to suggest the variable you probably meant, so a typo like PNvoeC = z_score(PNoveC) fails with a suggestion instead of creating a second variable:

# Enforce 'new' syntax.
if ref_var not in self.env.keys():
    raise NameError(ref_var)

Second, imports are limited to numpy, math and datetime, with an optional alias. Third, every line must be an assignment or a print(...), because variables are taken by value and an expression without assignment has no effect.

In the river examples, .act does the numeric preparation. It parses the bracketed strings into a 1839 × 24 array and standardizes every column:

.act

VectorRH = load_vec(VectorRH, sep=',')

VectorRH = z_score(VectorRH)
PNoveC = z_score(PNoveC)
PZeroC = z_score(PZeroC)
TempMaxC = z_score(TempMaxC)

z_score flattens its input before computing the mean and standard deviation, so the 24 hourly columns share one mean and one scale. That keeps the relative shape of a day intact, which is the right choice for a vector of the same quantity measured every hour.

The primitives are thin NumPy wrappers, and some carry their sources. interpola_outlier points to the NIST/SEMATECH handbook page on outlier detection [3], which describes the modified z-score: the distance from the median, divided by the median absolute deviation (MAD). The implementation follows that test, but the replacement differs from its docstring:

a = np.copy(x)
d = np.abs(a - np.median(a))
m_dev = np.median(d)
s = d / m_dev if m_dev else np.zero(len(d))
a[s > m] = m * m_dev

The docstring says outliers are "brought to the mean". The code sets them to m * m_dev, a positive spread value that is not centred on the data. Also, np.zero does not exist in NumPy (np.zeros does), so a column with a MAD of zero, such as a mostly dry rainfall series, raises AttributeError. None of the example files calls this primitive, which explains why these two problems went unnoticed.

One more detail matters when you write .act lines. The interpreter decides what a line is by searching for substrings: 'new ', then 'import', then 'plot'. A line such as Subplot = x contains plot and is sent to the plotting branch. Variable names need to avoid these words.

Saving uses the same format strings as loading

The .sap section writes variables back to text, and it reuses the declaration syntax in reverse. A format string with braces becomes the template of each output row. The interpreter refuses to save variables of different lengths into one file. The smallest complete pipeline in the repository uses only this section and the three before it:

.decl
# Import each original IrisData data row as a whole
source_file IrisData = "Your/Shuffled/Data/File"
{IrisData}

.res
# Instruct the interpreter that IrisData[i] is a string
IrisData: categorical

.act
new ShuffledData = shuffle(IrisData)

.sap
save_file SaveShuffled = "/your/save/path"
save "{ShuffledData}" into SaveShuffled

.make

A format string with a single name and no separator reads the whole line as one value. shuffle permutes the lines and save writes them out unchanged, so every row keeps its fields together. The file exists because of something in the .make section: its train/test split is sequential. The Iris data is sorted by species. When I ran the main Iris file on the unshuffled data, all 44 test samples were Iris virginica [4]. Part 2 shows where that split happens.

What kind of language this is

Martin Fowler separates external domain-specific languages, which have their own syntax and parser, from internal ones, which are written inside a host language [5]. RiverFlow's loader language is both. .decl, .res, .sap and .make are external: fixed keywords, parsed line by line. .act and the file paths are internal: they are Python, evaluated by the host.

That split explains most of the design. The external sections hold the decisions that should be visible when you read the file: which columns, which types, which windows. The internal section holds the arithmetic, where a new keyword for every NumPy operation would add nothing. The cost is that a makefile is code, not configuration. The Python documentation notes that when the globals dictionary passed to eval() has no __builtins__ key, Python inserts the built-in module [6]. A makefile can therefore call open() or __import__() despite the import whitelist. That is acceptable for files you wrote yourself, and it means a makefile from someone else needs the same review as a script.

At the end of .act, each variable is a clean array with a type and a length. Nothing so far says what one training sample is. That is the job of the .make section, which is the subject of part 2.