Updates for cesium library refactor

* Uses new .npz featureset serialization instead of xarray/netCDF

* Uses standard scikit-learn models/GridSearchCV instead of special `cesium` functions

Author: bnaul

.travis.yml

@@ -50,8 +50,6 @@ addons:
     packages:
     - ccache
     - wget
-    - libhdf5-serial-dev
-    - libnetcdf-dev
     - nodejs
     - supervisor
     - nginx

cesium_app/ext/sklearn_models.py

@@ -1,5 +1,37 @@
-from cesium.util import make_list
-from cesium.build_model import MODELS_TYPE_DICT
+import collections
+from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor
+from sklearn.linear_model import (LinearRegression, SGDClassifier,
+                                  RidgeClassifierCV, ARDRegression,
+                                  BayesianRidge)
+
+MODELS_TYPE_DICT = {'RandomForestClassifier': RandomForestClassifier,
+                    'RandomForestRegressor': RandomForestRegressor,
+                    'LinearSGDClassifier': SGDClassifier,
+                    'LinearRegressor': LinearRegression,
+                    'RidgeClassifierCV': RidgeClassifierCV,
+                    'BayesianARDRegressor': ARDRegression,
+                    'BayesianRidgeRegressor': BayesianRidge}
+
+
+def make_list(x):
+    """Wrap `x` in a list if it isn't already a list or tuple.
+
+    Parameters
+    ----------
+    x : any valid object
+        The parameter to be wrapped in a list.
+
+    Returns
+    -------
+    list or tuple
+        Returns `[x]` if `x` is not already a list or tuple, otherwise
+        returns `x`.
+
+    """
+    if isinstance(x, collections.Iterable) and not isinstance(x, (str, dict)):
+        return x
+    else:
+        return [x]
 
 
 model_descriptions = [

cesium_app/handlers/dataset.py

@@ -61,8 +61,7 @@ def post(self):
             zipfile_path,
             cfg['paths']['ts_data_folder'],
             headerfile_path)
-        meta_features = list(time_series.from_netcdf(ts_paths[0])
-                             .meta_features.keys())
+        meta_features = list(time_series.load(ts_paths[0]).meta_features.keys())
         unique_ts_paths = [os.path.join(os.path.dirname(ts_path),
                                         str(uuid.uuid4()) + "_" +
                                         util.secure_filename(ts_path))

cesium_app/handlers/feature.py

@@ -1,9 +1,7 @@
 import tornado.ioloop
 
-import xarray as xr
 from cesium import featurize, time_series
 from cesium.features import dask_feature_graph
-from cesium import featureset
 
 from .base import BaseHandler, AccessError
 from ..models import Dataset, Featureset, Project, File
@@ -79,7 +77,7 @@ def post(self):
             return self.error('Cannot access dataset')
 
         fset_path = pjoin(cfg['paths']['features_folder'],
-                          '{}_featureset.nc'.format(uuid.uuid4()))
+                          '{}_featureset.npz'.format(uuid.uuid4()))
 
         fset = Featureset.create(name=featureset_name,
                                  file=File.create(uri=fset_path),
@@ -89,15 +87,18 @@ def post(self):
 
         executor = yield self._get_executor()
 
-        all_time_series = executor.map(time_series.from_netcdf, dataset.uris)
+        all_time_series = executor.map(time_series.load, dataset.uris)
+        all_labels = executor.map(lambda ts: ts.label, all_time_series)
         all_features = executor.map(featurize.featurize_single_ts,
                                     all_time_series,
                                     features_to_use=features_to_use,
                                     custom_script_path=custom_script_path)
         computed_fset = executor.submit(featurize.assemble_featureset,
                                         all_features, all_time_series)
-        imputed_fset = executor.submit(featureset.Featureset.impute, computed_fset)
-        future = executor.submit(xr.Dataset.to_netcdf, imputed_fset, fset_path)
+        imputed_fset = executor.submit(featurize.impute_featureset,
+                                       computed_fset, inplace=False)
+        future = executor.submit(featurize.save_featureset, imputed_fset,
+                                 fset_path, labels=all_labels)
         fset.task_id = future.key
         fset.save()
 

cesium_app/handlers/model.py

@@ -4,19 +4,19 @@
 from ..models import Project, Model, Featureset, File
 from ..ext.sklearn_models import (
     model_descriptions as sklearn_model_descriptions,
-    check_model_param_types
+    check_model_param_types, MODELS_TYPE_DICT
     )
 from ..util import robust_literal_eval
 from ..config import cfg
+from cesium import featurize
 
 from os.path import join as pjoin
 import uuid
 import datetime
 
-from cesium import build_model, featureset
 import tornado.ioloop
+from sklearn.model_selection import GridSearchCV
 import joblib
-import xarray as xr
 from distributed.client import _wait
 
 
@@ -27,7 +27,7 @@ def _build_model_compute_statistics(fset_path, model_type, model_params,
     Parameters
     ----------
     fset_path : str
-        Path to feature set NetCDF file.
+        Path to feature set .npz file.
     model_type : str
         Type of model to be built, e.g. 'RandomForestClassifier'.
     model_params : dict
@@ -57,15 +57,16 @@ def _build_model_compute_statistics(fset_path, model_type, model_params,
         `params_to_optimize` is None or is an empty dict, this will be an empty
         dict.
     '''
-    fset = featureset.from_netcdf(fset_path)
-    computed_model = build_model.build_model_from_featureset(
-        featureset=fset, model_type=model_type,
-        model_parameters=model_params,
-        params_to_optimize=params_to_optimize)
-    score = build_model.score_model(computed_model, fset)
-    best_params = computed_model.best_params_ if params_to_optimize else {}
-    joblib.dump(computed_model, model_path)
-    fset.close()
+    fset, data = featurize.load_featureset(fset_path)
+    if len(data['labels']) != len(fset):
+        raise ValueError("Cannot build model for unlabeled feature set.")
+    model = MODELS_TYPE_DICT[model_type](**model_params)
+    if params_to_optimize:
+        model = GridSearchCV(model, params_to_optimize)
+    model.fit(fset, data['labels'])
+    score = model.score(fset, data['labels'])
+    best_params = model.best_params_ if params_to_optimize else {}
+    joblib.dump(model, model_path)
 
     return score, best_params
 

cesium_app/handlers/prediction.py

@@ -7,19 +7,17 @@
 from tornado.web import RequestHandler
 from tornado.escape import json_decode
 
-import cesium.time_series
-import cesium.featurize
-import cesium.predict
-import cesium.featureset
+from cesium import featurize, time_series
 from cesium.features import CADENCE_FEATS, GENERAL_FEATS, LOMB_SCARGLE_FEATS
 
-import xarray as xr
 import joblib
 from os.path import join as pjoin
 import uuid
 import datetime
 import os
 import tempfile
+import numpy as np
+import pandas as pd
 
 
 class PredictionHandler(BaseHandler):
@@ -82,27 +80,39 @@ def post(self):
         if (model.finished is None) or (fset.finished is None):
             return self.error('Computation of model or feature set still in progress')
 
-        prediction_path = pjoin(cfg['paths']['predictions_folder'],
-                                '{}_prediction.nc'.format(uuid.uuid4()))
-        prediction_file = File.create(uri=prediction_path)
+        pred_path = pjoin(cfg['paths']['predictions_folder'],
+                          '{}_prediction.npz'.format(uuid.uuid4()))
+        prediction_file = File.create(uri=pred_path)
         prediction = Prediction.create(file=prediction_file, dataset=dataset,
                                        project=dataset.project, model=model)
 
         executor = yield self._get_executor()
 
-        all_time_series = executor.map(cesium.time_series.from_netcdf,
-                                       dataset.uris)
-        all_features = executor.map(cesium.featurize.featurize_single_ts,
+        all_time_series = executor.map(time_series.load, dataset.uris)
+        all_labels = executor.map(lambda ts: ts.label, all_time_series)
+        all_features = executor.map(featurize.featurize_single_ts,
                                     all_time_series,
                                     features_to_use=fset.features_list,
                                     custom_script_path=fset.custom_features_script)
-        fset_data = executor.submit(cesium.featurize.assemble_featureset,
+        fset_data = executor.submit(featurize.assemble_featureset,
                                     all_features, all_time_series)
-        fset_data = executor.submit(cesium.featureset.Featureset.impute, fset_data)
-        model_data = executor.submit(joblib.load, model.file.uri)
-        predset = executor.submit(cesium.predict.model_predictions,
-                                  fset_data, model_data)
-        future = executor.submit(xr.Dataset.to_netcdf, predset, prediction_path)
+        imputed_fset = executor.submit(featurize.impute_featureset,
+                                       fset_data, inplace=False)
+        model_or_gridcv = executor.submit(joblib.load, model.file.uri)
+        model_data = executor.submit(lambda model: model.best_estimator_
+                                     if hasattr(model, 'best_estimator_') else model,
+                                     model_or_gridcv)
+        preds = executor.submit(lambda fset, model: model.predict(fset),
+                                imputed_fset, model_data)
+        pred_probs = executor.submit(lambda fset, model: model.predict_proba(fset)
+                                     if hasattr(model, 'predict_proba') else [],
+                                     imputed_fset, model_data)
+        all_classes = executor.submit(lambda model: model.classes_
+                                      if hasattr(model, 'classes_') else [],
+                                      model_data)
+        future = executor.submit(featurize.save_featureset, imputed_fset,
+                                 pred_path, labels=all_labels, preds=preds,
+                                 pred_probs=pred_probs, all_classes=all_classes)
 
         prediction.task_id = future.key
         prediction.save()
@@ -114,14 +124,18 @@ def post(self):
 
     def get(self, prediction_id=None, action=None):
         if action == 'download':
-            prediction = cesium.featureset.from_netcdf(self._get_prediction(prediction_id).file.uri)
-            with tempfile.NamedTemporaryFile() as tf:
-                util.prediction_to_csv(prediction, tf.name)
-                with open(tf.name) as f:
-                    self.set_header("Content-Type", 'text/csv; charset="utf-8"')
-                    self.set_header("Content-Disposition",
-                                    "attachment; filename=cesium_prediction_results.csv")
-                    self.write(f.read())
+            pred_path = self._get_prediction(prediction_id).file.uri
+            fset, data = featurize.load_featureset(pred_path)
+            result = pd.DataFrame({'ts_name': fset.index,
+                                   'label': data['labels'],
+                                   'prediction': data['preds']},
+                                  columns=['ts_name', 'label', 'prediction'])
+            if data.get('pred_probs'):
+                result['probability'] = np.max(data['pred_probs'], axis=1)
+            self.set_header("Content-Type", 'text/csv; charset="utf-8"')
+            self.set_header("Content-Disposition", "attachment; "
+                            "filename=cesium_prediction_results.csv")
+            self.write(result.to_csv(index=False))
         else:
             if prediction_id is None:
                 predictions = [prediction
@@ -144,20 +158,22 @@ class PredictRawDataHandler(BaseHandler):
     def post(self):
         ts_data = json_decode(self.get_argument('ts_data'))
         model_id = json_decode(self.get_argument('modelID'))
-        meta_feats = json_decode(
-            self.get_argument('meta_features', 'null'))
-        impute_kwargs = json_decode(
-            self.get_argument('impute_kwargs', '{}'))
+        meta_feats = json_decode(self.get_argument('meta_features', 'null'))
+        impute_kwargs = json_decode(self.get_argument('impute_kwargs', '{}'))
 
         model = Model.get(Model.id == model_id)
-        computed_model = joblib.load(model.file.uri)
+        model_data = joblib.load(model.file.uri)
+        if hasattr(model_data, 'best_estimator_'):
+            model_data = model_data.best_estimator_
         features_to_use = model.featureset.features_list
 
-        fset_data = cesium.featurize.featurize_time_series(
-            *ts_data, features_to_use=features_to_use, meta_features=meta_feats)
-        fset = cesium.featureset.Featureset(fset_data).impute(**impute_kwargs)
-
-        predset = cesium.predict.model_predictions(fset, computed_model)
-        predset['name'] = predset.name.astype('str')
-
-        return self.success(predset)
+        fset = featurize.featurize_time_series(*ts_data,
+                                               features_to_use=features_to_use,
+                                               meta_features=meta_feats)
+        fset = featurize.impute_featureset(fset, **impute_kwargs)
+        data = {'preds': model_data.predict(fset),
+                'all_classes': model_data.classes_}
+        if hasattr(model_data, 'predict_proba'):
+            data['pred_probs'] = model_data.predict_proba(fset)
+        pred_info = Prediction.format_pred_data(fset, data)
+        return self.success(pred_info)

cesium_app/json_util.py

@@ -1,9 +1,9 @@
 from datetime import datetime
 import simplejson as json
 import numpy as np
+import pandas as pd
 import peewee
 import six
-import xarray as xr
 
 
 data_types = {
@@ -16,28 +16,6 @@
     }
 
 
-def dataset_row_to_dict(row):
-    """Semi-hacky helper function for extracting JSON for a single time series
-    of a featureset. For now assumes single-channel data since that's what the
-    front end can display.
-    """
-    out = {}
-    out['target'] = row.target.values.item() if 'target' in row else None
-    if 'prediction' in row:
-        if 'class_label' in row:  # {class label: probability}
-            out['prediction'] = {six.u(label): value for label, value
-                                 in zip(row.class_label.values,
-                                        row.prediction.values)}
-        else: # just a single predicted label or target
-            out['prediction'] = row.prediction.values.item()
-    else:
-        out['prediction'] = None
-    out['features'] = {f: row[f].item()
-                       for f in row.data_vars if f != 'prediction'}
-
-    return out
-
-
 class Encoder(json.JSONEncoder):
     """Extends json.JSONEncoder with additional capabilities/configurations."""
     def default(self, o):
@@ -62,9 +40,9 @@ def default(self, o):
         elif isinstance(o, np.ndarray):
             return o.tolist()
 
-        elif isinstance(o, xr.Dataset):
-            return {ts_name: dataset_row_to_dict(o.sel(name=ts_name))
-                    for ts_name in o.name.values}
+        elif isinstance(o, pd.DataFrame):
+            o.columns = o.columns.droplevel('channel')  # flatten MultiIndex
+            return o.to_dict(orient='index')
 
         elif type(o) is type and o in data_types:
             return data_types[o]