122 lines
3.6 KiB
Python
122 lines
3.6 KiB
Python
#!/usr/bin/env python
|
|
# file jurse2.py
|
|
# author Florent Guiotte <florent.guiotte@irisa.fr>
|
|
# version 0.0
|
|
# date 26 mai 2020
|
|
"""Abstract
|
|
|
|
doc.
|
|
"""
|
|
|
|
import importlib
|
|
import numpy as np
|
|
from sklearn import metrics
|
|
from .protocol import Protocol, TestError
|
|
|
|
|
|
class Jurse2(Protocol):
|
|
"""Second JURSE test protocol for LiDAR classification with 2D maps.
|
|
|
|
"""
|
|
|
|
def __init__(self, expe):
|
|
super().__init__(expe, self.__class__.__name__)
|
|
self._results = {}
|
|
|
|
def _run(self):
|
|
self._log.info('Load data')
|
|
try:
|
|
data = self._load_data()
|
|
except Exception:
|
|
raise TestError('Error occured during data loading')
|
|
|
|
self._log.info('Compute descriptors')
|
|
try:
|
|
descriptors = self._compute_descriptors(data)
|
|
except Exception:
|
|
raise TestError('Error occured during description')
|
|
|
|
self._log.info('Classify descriptors')
|
|
try:
|
|
classification = self._compute_classification(descriptors)
|
|
except Exception:
|
|
raise TestError('Error occured during classification')
|
|
|
|
self._log.info('Run metrics')
|
|
metrics = self._run_metrics(classification, descriptors)
|
|
|
|
self._results['metrics'] = metrics
|
|
|
|
def _load_data(self):
|
|
data_loader = self._expe['data_loader']
|
|
|
|
loader = importlib.import_module(data_loader['name'])
|
|
data = loader.run(**data_loader['parameters'])
|
|
|
|
return data
|
|
|
|
def _compute_descriptors(self, data):
|
|
script = self._expe['descriptors_script']
|
|
|
|
desc = importlib.import_module(script['name'])
|
|
att = desc.run(*data, **script['parameters'])
|
|
|
|
return att
|
|
|
|
def _compute_classification(self, descriptors):
|
|
X, y, groups, Xn = descriptors
|
|
|
|
# CrossVal and ML
|
|
cv = self._expe['cross_validation']
|
|
cl = self._expe['classifier']
|
|
|
|
cross_val = getattr(importlib.import_module(cv['package']), cv['name'])
|
|
classifier = getattr(importlib.import_module(cl['package']), cl['name'])
|
|
|
|
y_pred = np.zeros_like(y)
|
|
|
|
cl_feature_importances = []
|
|
cvi = cross_val(**cv['parameters'])
|
|
for train_index, test_index in cvi.split(X, y, groups):
|
|
cli = classifier(**cl['parameters'])
|
|
|
|
self._log.info(' - fit')
|
|
cli.fit(X[train_index], y[train_index])
|
|
|
|
self._log.info(' - predict')
|
|
y_pred[test_index] = cli.predict(X[test_index])
|
|
|
|
cl_feature_importances += [cli.feature_importances_.copy()]
|
|
|
|
cl_feature_importances = np.array(cl_feature_importances)
|
|
self._results['features'] = {
|
|
'name': Xn,
|
|
'importance': cl_feature_importances.tolist()
|
|
}
|
|
|
|
return y_pred
|
|
|
|
def _get_results(self):
|
|
return self._results
|
|
|
|
def _run_metrics(self, classification, descriptors):
|
|
X, y_true, groups, Xn = descriptors
|
|
y_pred = classification
|
|
|
|
self._log.info(' - Scores')
|
|
self.oa = metrics.accuracy_score(y_true, y_pred)
|
|
self.aa = metrics.balanced_accuracy_score(y_true, y_pred)
|
|
self.k = metrics.cohen_kappa_score(y_true, y_pred)
|
|
|
|
self._log.info(' - Additional results')
|
|
p, r, f, s = metrics.precision_recall_fscore_support(y_true, y_pred)
|
|
cm = metrics.confusion_matrix(y_true, y_pred)
|
|
results = {'dimensions': X.shape[-1],
|
|
'precision': p.tolist(),
|
|
'recall': r.tolist(),
|
|
'f1score': f.tolist(),
|
|
'support': s.tolist(),
|
|
'confusion': cm.tolist()}
|
|
|
|
return results
|