swallow testerrors and cleanup core

saqc/core/core.py

 #! /usr/bin/env python
 # -*- coding: utf-8 -*-
+import logging
 
 import numpy as np
 import pandas as pd
@@ -11,7 +12,7 @@ from ..lib.plotting import plot_hook, plotall_hook
 from ..flagger import BaseFlagger, CategoricalBaseFlagger, SimpleFlagger, DmpFlagger
 
 
-def collectVariables(meta, flagger, data, flags):
+def _collectVariables(meta, flagger, data, flags):
     """ find every requested variable in the meta, that is not already present in flags """
     if flags is None:
         ignore = list()
@@ -27,7 +28,7 @@ def collectVariables(meta, flagger, data, flags):
     return varnames
 
 
-def _check_input(data, flags, flagger):
+def _checkInput(data, flags, flagger):
     if not isinstance(data, pd.DataFrame):
         raise TypeError('data must be of type pd.DataFrame')
 
@@ -60,9 +61,9 @@ def _setup():
     pd.set_option('mode.chained_assignment', 'warn')
 
 
-def runner(metafname, flagger, data, flags=None, nodata=np.nan):
+def runner(metafname, flagger, data, flags=None, nodata=np.nan, error_policy='raise'):
     _setup()
-    _check_input(data, flags, flagger)
+    _checkInput(data, flags, flagger)
     config = prepareConfig(readConfig(metafname), data)
 
     # split config into the test and some 'meta' data
@@ -70,7 +71,7 @@ def runner(metafname, flagger, data, flags=None, nodata=np.nan):
     meta = config[config.columns.difference(tests.columns)]
 
     # prepapre the flags
-    varnames = collectVariables(meta, flagger, data, flags)
+    varnames = _collectVariables(meta, flagger, data, flags)
     fresh = flagger.initFlags(pd.DataFrame(index=data.index, columns=varnames))
     flags = fresh if flags is None else flags.join(fresh)
 
@@ -87,6 +88,8 @@ def runner(metafname, flagger, data, flags=None, nodata=np.nan):
             continue
 
         for idx, configrow in meta.iterrows():
+
+            # store config params in some handy variables
             varname = configrow[Fields.VARNAME]
             start_date = configrow[Fields.START]
             end_date = configrow[Fields.END]
@@ -98,17 +101,25 @@ def runner(metafname, flagger, data, flags=None, nodata=np.nan):
             if varname not in data and varname not in flags:
                 continue
 
+            # prepare the data for the tests
             dchunk = data.loc[start_date:end_date]
+            fchunk = flags.loc[start_date:end_date]
             if dchunk.empty:
                 continue
 
-            fchunk = flags.loc[start_date:end_date]
-
-            dchunk, ffchunk = evalExpression(
-                flag_test,
-                data=dchunk, flags=fchunk.copy(), field=varname,
-                flagger=flagger, nodata=nodata)
-
+            # actually run the tests
+            try:
+                dchunk, ffchunk = evalExpression(
+                    flag_test,
+                    data=dchunk, flags=fchunk.copy(), field=varname,
+                    flagger=flagger, nodata=nodata)
+            except Exception as e:
+                if _handleErrors(e, configrow, flag_test, error_policy):
+                    raise e
+                else:
+                    continue
+
+            # write back the (new) flagged data
             data.loc[start_date:end_date] = dchunk
             flags.loc[start_date:end_date] = ffchunk.squeeze()
 
@@ -120,3 +131,17 @@ def runner(metafname, flagger, data, flags=None, nodata=np.nan):
     plotall_hook(data, flags, flagger)
 
     return data, flags
+
+
+def _handleErrors(err, configrow, test, policy):
+    line = configrow[Fields.LINENUMBER]
+    msg = f" config, line {line}, test: `{test}` failed with `{type(err).__name__}: {err}`"
+    if policy == 'raise':
+        return True
+    elif policy == 'ignore':
+        logging.debug(msg)
+        return False
+    elif policy == 'warn':
+        logging.warning(msg)
+        return False
+    return True