Skip to content
Snippets Groups Projects
Commit 240596d5 authored by Baptiste Bauvin's avatar Baptiste Bauvin
Browse files

formatted late fusion clfs that will keep for sure

parent c6783173
Branches
Tags
No related merge requests found
Showing
with 172 additions and 943 deletions
...@@ -6,17 +6,19 @@ import os ...@@ -6,17 +6,19 @@ import os
import numpy as np import numpy as np
from ...multiview.multiview_utils import ConfigGenerator, \ from ...multiview.multiview_utils import ConfigGenerator, \
get_monoview_classifier, get_examples_views_indices, \ get_examples_views_indices, get_available_monoview_classifiers, \
get_available_monoview_classifiers, BaseMultiviewClassifier BaseMultiviewClassifier
from .fusion_utils import BaseLateFusionClassifier
class DiversityFusion(BaseMultiviewClassifier): class DiversityFusionClassifier(BaseMultiviewClassifier,
BaseLateFusionClassifier):
"""This is the base class for all the diversity fusion based classifiers.""" """This is the base class for all the diversity fusion based classifiers."""
def __init__(self, random_state=None, classifier_names=None, def __init__(self, random_state=None, classifier_names=None,
monoview_estimators=None, classifiers_configs=None): monoview_estimators=None, classifiers_configs=None):
"""Used to init the instances""" """Used to init the instances"""
super(DiversityFusion, self).__init__(random_state) super(DiversityFusionClassifier, self).__init__(random_state)
if classifier_names is None: if classifier_names is None:
classifier_names = get_available_monoview_classifiers() classifier_names = get_available_monoview_classifiers()
self.classifier_names = classifier_names self.classifier_names = classifier_names
...@@ -33,18 +35,7 @@ class DiversityFusion(BaseMultiviewClassifier): ...@@ -33,18 +35,7 @@ class DiversityFusion(BaseMultiviewClassifier):
self.estimator_pool = [] self.estimator_pool = []
for classifier_idx, classifier_name in enumerate(self.classifier_names): for classifier_idx, classifier_name in enumerate(self.classifier_names):
self.estimator_pool.append([]) self.estimator_pool.append([])
if self.classifiers_configs is not None and classifier_name in self.classifiers_configs: estimator = self.init_monoview_estimator(classifier_name)
if 'random_state' in inspect.getfullargspec(get_monoview_classifier(classifier_name).__init__).args:
estimator = get_monoview_classifier(classifier_name)(random_state=self.random_state,
**self.classifiers_configs[classifier_name])
else:
estimator = get_monoview_classifier(classifier_name)(
**self.classifiers_configs[classifier_name])
else:
if 'random_state' in inspect.getfullargspec(get_monoview_classifier(classifier_name).__init__).args:
estimator = get_monoview_classifier(classifier_name)(random_state=self.random_state)
else:
estimator = get_monoview_classifier(classifier_name)()
for idx, view_idx in enumerate(views_indices): for idx, view_idx in enumerate(views_indices):
estimator.fit(X.get_v(view_idx, train_indices), y[train_indices]) estimator.fit(X.get_v(view_idx, train_indices), y[train_indices])
self.estimator_pool[classifier_idx].append(estimator) self.estimator_pool[classifier_idx].append(estimator)
...@@ -96,7 +87,7 @@ class DiversityFusion(BaseMultiviewClassifier): ...@@ -96,7 +87,7 @@ class DiversityFusion(BaseMultiviewClassifier):
return combinations, combis, div_measure, classifiers_decisions, nb_views return combinations, combis, div_measure, classifiers_decisions, nb_views
class GlobalDiversityFusion(DiversityFusion): class GlobalDiversityFusionClassifier(DiversityFusionClassifier):
def choose_combination(self, X, y, examples_indices, view_indices): def choose_combination(self, X, y, examples_indices, view_indices):
combinations, combis, div_measure, classifiers_decisions, nb_views = self.init_combinations( combinations, combis, div_measure, classifiers_decisions, nb_views = self.init_combinations(
...@@ -114,7 +105,7 @@ class GlobalDiversityFusion(DiversityFusion): ...@@ -114,7 +105,7 @@ class GlobalDiversityFusion(DiversityFusion):
in enumerate(best_combination)] in enumerate(best_combination)]
class CoupleDiversityFusion(DiversityFusion): class CoupleDiversityFusionClassifier(DiversityFusionClassifier):
def choose_combination(self, X, y, examples_indices, view_indices): def choose_combination(self, X, y, examples_indices, view_indices):
combinations, combis, div_measure, classifiers_decisions, nb_views = self.init_combinations( combinations, combis, div_measure, classifiers_decisions, nb_views = self.init_combinations(
......
import inspect
from ...multiview.multiview_utils import BaseMultiviewClassifier, get_monoview_classifier
class BaseLateFusionClassifier(BaseMultiviewClassifier):
def init_monoview_estimator(self, classifier_name, classifier_index=None):
if classifier_index is not None:
classifier_configs = self.classifier_configs[classifier_index]
else:
classifier_configs = self.classifier_configs
if classifier_configs is not None and classifier_name in classifier_configs:
if 'random_state' in inspect.getfullargspec(
get_monoview_classifier(classifier_name).__init__).args:
estimator = get_monoview_classifier(classifier_name)(
random_state=self.random_state,
**classifier_configs[classifier_name])
else:
estimator = get_monoview_classifier(classifier_name)(
**classifier_configs[classifier_name])
else:
if 'random_state' in inspect.getfullargspec(
get_monoview_classifier(classifier_name).__init__).args:
estimator = get_monoview_classifier(classifier_name)(
random_state=self.random_state)
else:
estimator = get_monoview_classifier(classifier_name)()
return estimator
import numpy as np
import warnings
from scipy.stats import uniform
from ...multiview.multiview_utils import BaseMultiviewClassifier, get_available_monoview_classifiers, get_monoview_classifier, get_examples_views_indices, ConfigGenerator
from .fusion_utils import BaseLateFusionClassifier
class ClassifierCombinator:
def __init__(self, nb_view):
self.nb_view = nb_view
self.available_classifiers = get_available_monoview_classifiers()
def rvs(self, random_state=None):
classifier_names = random_state.choice(self.available_classifiers,
size=self.nb_view, replace=True)
return classifier_names
class MultipleConfigGenerator(ConfigGenerator):
def __init__(self, nb_view):
super(MultipleConfigGenerator, self).__init__(get_available_monoview_classifiers())
self.nb_view = nb_view
self.multiple_distribs = [self.distribs for _ in range(nb_view)]
def rvs(self, random_state=None):
config_samples = [super(MultipleConfigGenerator, self).rvs(random_state)
for _ in range(self.nb_view)]
return config_samples
class WeightsGenerator:
def __init__(self, nb_view):
self.nb_view=nb_view
self.uniform = uniform(loc=0, state=1)
def rvs(self, random_state=None):
return np.array([uniform.rvs(random_state=random_state)
for _ in range(self.nb_view)])
class LateFusionClassifier(BaseMultiviewClassifier, BaseLateFusionClassifier):
def __init__(self, random_state=None, classifier_names=None,
classifier_configs=None, nb_cores=1, nb_view=None, weights=None):
super(LateFusionClassifier, self).__init__(random_state)
self.verif_clf_views(classifier_names, nb_view)
self.nb_view = len(classifier_names)
self.classifiers_names = classifier_names
self.classifier_configs = classifier_configs
self.monoview_estimators = [self.init_monoview_estimator(classifier_name, classifier_index)
for classifier_index, classifier_name in self.classifiers_names]
self.nb_cores = nb_cores
self.accuracies = np.zeros(len(classifier_names))
self.needProbas = False
if weights is None:
self.weights = np.ones(nb_view)/nb_view
else:
self.weights = weights
self.param_names = ["classifier_names", "classifier_configs", "weights"]
self.distribs =[ClassifierCombinator(self.nb_view),
MultipleConfigGenerator(self.nb_view),
WeightsGenerator(nb_view)]
def fit(self, X, y, train_indices=None, views_indices=None):
train_indices, views_indices = get_examples_views_indices(X,
train_indices,
views_indices)
self.monoview_estimators = [monoview_estimator.fit(X.get_v(view_index, train_indices), y[train_indices]) for view_index, monoview_estimator in zip(views_indices, self.monoview_estimators)]
return self
def verif_clf_views(self, classifier_names, nb_view):
if classifier_names is None:
if nb_view is None:
raise AttributeError(self.__class__.__name__+" must have either classifier_names or nb_views provided.")
else:
self.classifiers_names = self.get_classifiers(get_available_monoview_classifiers(), nb_view)
else:
if nb_view is None:
self.classifiers_names = classifier_names
else:
if len(classifier_names)==nb_view:
self.classifiers_names = classifier_names
else:
warnings.warn("nb_view and classifier_names not matching, choosing nb_view random classifiers in classifier_names.", UserWarning)
self.classifiers_names = self.get_classifiers(classifier_names, nb_view)
def get_classifiers(self, classifiers_names, nb_choices):
return self.random_state.choice(classifiers_names, size=nb_choices, replace=True)
import numpy as np
from ..multiview_classifiers.additions.late_fusion_utils import \
LateFusionClassifier
from ..multiview.multiview_utils import get_examples_views_indices
classifier_class_name = "BayesianInferenceClassifier"
class BayesianInferenceClassifier(LateFusionClassifier):
def __init__(self, random_state, classifier_names=None,
classifier_configs=None, nb_view=None, nb_cores=1):
super(BayesianInferenceClassifier, self).__init__(random_state=random_state,
classifier_names=classifier_names,
classifier_configs=classifier_configs,
nb_cores=nb_cores,
nb_view=nb_view)
def predict(self, X, example_indices=None, views_indices=None):
example_indices, views_indices = get_examples_views_indices(X, example_indices, views_indices)
if sum(self.weights) != 1.0:
self.weights = self.weights / sum(self.weights)
view_scores = []
for index, view_index in enumerate(views_indices):
view_scores.append(np.power(
self.monoviewClassifiers[index].predict_proba(X.get_v(view_index,
example_indices)),
self.weights[index]))
view_scores = np.array(view_scores)
predicted_labels = np.argmax(np.prod(view_scores, axis=0), axis=1)
return predicted_labels
import numpy as np import numpy as np
from multiview_platform.mono_multi_view_classifiers.multiview_classifiers.additions.diversity_utils import GlobalDiversityFusion from multiview_platform.mono_multi_view_classifiers.multiview_classifiers.additions.diversity_utils import GlobalDiversityFusionClassifier
classifier_class_name = "DifficultyFusion" classifier_class_name = "DifficultyFusion"
class DifficultyFusion(GlobalDiversityFusion): class DifficultyFusion(GlobalDiversityFusionClassifier):
def diversity_measure(self, classifiers_decisions, combination, y): def diversity_measure(self, classifiers_decisions, combination, y):
_, nb_view, nb_examples = classifiers_decisions.shape _, nb_view, nb_examples = classifiers_decisions.shape
......
import numpy as np import numpy as np
from multiview_platform.mono_multi_view_classifiers.multiview_classifiers.additions.diversity_utils import CoupleDiversityFusion from multiview_platform.mono_multi_view_classifiers.multiview_classifiers.additions.diversity_utils import CoupleDiversityFusionClassifier
classifier_class_name = "DisagreeFusion" classifier_class_name = "DisagreeFusion"
class DisagreeFusion(CoupleDiversityFusion): class DisagreeFusion(CoupleDiversityFusionClassifier):
def diversity_measure(self, first_classifier_decision, second_classifier_decision, _): def diversity_measure(self, first_classifier_decision, second_classifier_decision, _):
return np.logical_xor(first_classifier_decision, second_classifier_decision) return np.logical_xor(first_classifier_decision, second_classifier_decision)
import numpy as np import numpy as np
from multiview_platform.mono_multi_view_classifiers.multiview_classifiers.additions.diversity_utils import \ from multiview_platform.mono_multi_view_classifiers.multiview_classifiers.additions.diversity_utils import \
CoupleDiversityFusion CoupleDiversityFusionClassifier
classifier_class_name = "DoubleFaultFusion" classifier_class_name = "DoubleFaultFusion"
class DoubleFaultFusion(CoupleDiversityFusion): class DoubleFaultFusion(CoupleDiversityFusionClassifier):
def diversity_measure(self, first_classifier_decision, def diversity_measure(self, first_classifier_decision,
second_classifier_decision, y): second_classifier_decision, y):
......
import numpy as np import numpy as np
from multiview_platform.mono_multi_view_classifiers.multiview_classifiers.additions.diversity_utils import GlobalDiversityFusion from multiview_platform.mono_multi_view_classifiers.multiview_classifiers.additions.diversity_utils import GlobalDiversityFusionClassifier
classifier_class_name = "EntropyFusion" classifier_class_name = "EntropyFusion"
class EntropyFusion(GlobalDiversityFusion): class EntropyFusion(GlobalDiversityFusionClassifier):
def diversity_measure(self, classifiers_decisions, combination, y): def diversity_measure(self, classifiers_decisions, combination, y):
_, nb_view, nb_examples = classifiers_decisions.shape _, nb_view, nb_examples = classifiers_decisions.shape
......
#!/usr/bin/env python
# -*- encoding: utf-8
import numpy as np
from ....utils.dataset import get_v
from .... import monoview_classifiers
class EarlyFusionClassifier(object):
def __init__(self, randomState, monoviewClassifierName, monoviewClassifierConfig, NB_CORES=1):
self.monoviewClassifierName = monoviewClassifierName
if type(monoviewClassifierConfig) == dict:
pass
elif monoviewClassifierConfig is None:
pass
else:
monoviewClassifierConfig = dict((str(configIndex), config[0]) for configIndex, config in
enumerate(monoviewClassifierConfig
))
monoviewClassifierModule = getattr(monoview_classifiers, self.monoviewClassifierName)
if monoviewClassifierConfig is not None:
self.monoviewClassifier = getattr(monoviewClassifierModule, self.monoviewClassifierName)(
random_state=randomState, **monoviewClassifierConfig)
else:
self.monoviewClassifier = getattr(monoviewClassifierModule, self.monoviewClassifierName)(
random_state=randomState)
self.monoviewClassifiersConfig = monoviewClassifierConfig
self.nbCores = NB_CORES
self.monoviewData = None
self.randomState = randomState
def makeMonoviewData_hdf5(self, DATASET, weights=None, usedIndices=None, viewsIndices=None):
if type(viewsIndices) == type(None):
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
nbView = len(viewsIndices)
if usedIndices is None:
usedIndices = range(DATASET.get("Metadata").attrs["datasetLength"])
if type(weights) == type(None):
weights = np.array([1 / nbView for i in range(nbView)])
if sum(weights) != 1:
weights = weights / sum(weights)
self.monoviewData = np.concatenate([get_v(DATASET, viewIndex, usedIndices)
for index, viewIndex in enumerate(viewsIndices)], axis=1)
import numpy as np
from ...Methods.EarlyFusion import EarlyFusionClassifier
from ..... import monoview_classifiers
def genParamsSets(classificationKWARGS, randomState, nIter=1):
nbView = classificationKWARGS["nbView"]
if classificationKWARGS["classifiersConfigs"] is None:
monoviewClassifierModule = getattr(monoview_classifiers, classificationKWARGS["classifiersNames"])
paramsMonoview = monoviewClassifierModule.paramsToSet(nIter, randomState)
paramsSets = []
for iterIndex in range(nIter):
randomWeightsArray = randomState.random_sample(nbView)
normalizedArray = randomWeightsArray / np.sum(randomWeightsArray)
paramsSets.append([normalizedArray, paramsMonoview[iterIndex]])
return paramsSets
def getArgs(benchmark, args, views, viewsIndices, directory, resultsMonoview, classificationIndices):
argumentsList = []
if args.FU_E_cl_names != ['']:
pass
else:
monoviewClassifierModulesNames = benchmark["monoview"]
args.FU_E_cl_names = monoviewClassifierModulesNames
args.FU_E_cl_config = [None for _ in monoviewClassifierModulesNames]
for classifierName, classifierConfig in zip(args.FU_E_cl_names, args.FU_E_cl_config):
monoviewClassifierModule = getattr(monoview_classifiers, classifierName)
if classifierConfig is not None:
arguments = {"CL_type": "fusion",
"views": views,
"NB_VIEW": len(views),
"viewsIndices": viewsIndices,
"NB_CLASS": len(args.CL_classes),
"LABELS_NAMES": args.CL_classes,
"FusionKWARGS": {"fusionType": "EarlyFusion",
"fusionMethod": "WeightedLinear",
"classifiersNames": classifierName,
"classifiersConfigs": monoviewClassifierModule.getKWARGS([arg.split(":")
for arg in
classifierConfig.split(
",")]),
'fusionMethodConfig': args.FU_E_method_configs,
"nbView": (len(viewsIndices))}}
else:
arguments = {"CL_type": "fusion",
"views": views,
"NB_VIEW": len(views),
"viewsIndices": viewsIndices,
"NB_CLASS": len(args.CL_classes),
"LABELS_NAMES": args.CL_classes,
"FusionKWARGS": {"fusionType": "EarlyFusion",
"fusionMethod": "WeightedLinear",
"classifiersNames": classifierName,
"classifiersConfigs": None,
'fusionMethodConfig': args.FU_E_method_configs,
"nbView": (len(viewsIndices))}}
argumentsList.append(arguments)
return argumentsList
class WeightedLinear(EarlyFusionClassifier):
def __init__(self, randomState, NB_CORES=1, **kwargs):
EarlyFusionClassifier.__init__(self, randomState, kwargs['classifiersNames'], kwargs['classifiersConfigs'],
NB_CORES=NB_CORES)
if kwargs['fusionMethodConfig'] is None:
self.weights = np.ones(len(kwargs["classifiersNames"]), dtype=float)
elif kwargs['fusionMethodConfig'] == ['']:
self.weights = np.ones(len(kwargs["classifiersNames"]), dtype=float)
else:
self.weights = np.array(map(float, kwargs['fusionMethodConfig']))
self.weights /= float(max(self.weights))
def fit_hdf5(self, DATASET, labels, trainIndices=None, viewsIndices=None):
if type(viewsIndices) == type(None):
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
if trainIndices is None:
trainIndices = range(DATASET.get("Metadata").attrs["datasetLength"])
self.weights /= float(max(self.weights))
self.makeMonoviewData_hdf5(DATASET, weights=self.weights, usedIndices=trainIndices, viewsIndices=viewsIndices)
# monoviewClassifierModule = getattr(monoview_classifiers, self.monoviewClassifierName)
self.monoviewClassifier.fit(self.monoviewData, labels[trainIndices])
# self.randomState,
# NB_CORES=self.nbCores,
# **self.monoviewClassifiersConfig)
def setParams(self, paramsSet):
self.weights = paramsSet[0]
self.monoviewClassifiersConfig = paramsSet[1]
def predict_hdf5(self, DATASET, usedIndices=None, viewsIndices=None):
if type(viewsIndices) == type(None):
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
self.weights /= float(np.sum(self.weights))
if usedIndices is None:
usedIndices = range(DATASET.get("Metadata").attrs["datasetLength"])
self.makeMonoviewData_hdf5(DATASET, weights=self.weights, usedIndices=usedIndices, viewsIndices=viewsIndices)
predictedLabels = self.monoviewClassifier.predict(self.monoviewData)
return predictedLabels
def predict_proba_hdf5(self, DATASET, usedIndices=None):
if usedIndices is None:
usedIndices = range(DATASET.get("Metadata").attrs["datasetLength"])
self.makeMonoviewData_hdf5(DATASET, weights=self.weights, usedIndices=usedIndices)
predictedLabels = self.monoviewClassifier.predict_proba(self.monoviewData)
return predictedLabels
def getConfig(self, fusionMethodConfig, monoviewClassifiersNames, monoviewClassifiersConfigs):
configString = "with weighted concatenation, using weights : " + ", ".join(map(str, self.weights)) + \
" with monoview classifier : "
# monoviewClassifierModule = getattr(monoview_classifiers, monoviewClassifiersNames)
configString += self.monoviewClassifier.getConfig()
return configString
def gridSearch(self, classificationKWARGS):
return
import os
for module in os.listdir(os.path.dirname(os.path.realpath(__file__))):
if module == '__init__.py' or module[-3:] != '.py':
continue
__import__(module[:-3], locals(), globals(), [], 1)
del module
del os
#!/usr/bin/env python
# -*- encoding: utf-8
import numpy as np
import itertools
from joblib import Parallel, delayed
import sys
import math
from .... import monoview_classifiers
from .... import metrics
from ....utils.dataset import get_v
# def canProbasClassifier(classifierConfig):
# try:
# _ = getattr(classifierConfig, "predict_proba")
# return True
# except AttributeError:
# return False
def fitMonoviewClassifier(monoviewClassifier, data, labels, needProbas, randomState, nbCores=1):
if needProbas and not monoviewClassifier.canProbas():
DTConfig = {"max_depth": 300, "criterion": "entropy", "splitter": "random"}
monoviewClassifier = getattr(monoview_classifiers, "DecisionTree")(random_state=randomState, **DTConfig)
classifier = monoviewClassifier.fit(data, labels)
return classifier
else:
# if type(classifierConfig) is dict:
# pass
# else:
# classifierConfig = dict((str(configIndex), config)
# for configIndex, config in enumerate(classifierConfig))
classifier = monoviewClassifier.fit(data, labels)
return classifier
def getScores(LateFusionClassifiers):
return ""
def intersect(allClassifersNames, directory, viewsIndices, resultsMonoview, classificationIndices):
wrongSets = [[] for _ in viewsIndices]
# wrongSets = [0 for _ in allClassifersNames]
classifiersNames = [[] for _ in viewsIndices]
nbViews = len(viewsIndices)
trainLabels = np.genfromtxt(directory + "train_labels.csv", delimiter=",").astype(np.int16)
length = len(trainLabels)
for resultMonoview in resultsMonoview:
if resultMonoview.classifier_name in classifiersNames[viewsIndices.index(resultMonoview.view_index)]:
classifierIndex = classifiersNames.index(resultMonoview.classifier_name)
wrongSets[resultMonoview.view_index][classifierIndex] = np.where(
trainLabels + resultMonoview.full_labels_pred[classificationIndices[0]] == 1)[0]
else:
classifiersNames[viewsIndices.index(resultMonoview.view_index)].append(resultMonoview.classifier_name)
wrongSets[viewsIndices.index(resultMonoview.view_index)].append(
np.where(trainLabels + resultMonoview.full_labels_pred[classificationIndices[0]] == 1)[0])
combinations = itertools.combinations_with_replacement(range(len(classifiersNames[0])), nbViews)
bestLen = length
bestCombination = None
for combination in combinations:
intersect = np.arange(length, dtype=np.int16)
for viewIndex, classifierIndex in enumerate(combination):
intersect = np.intersect1d(intersect, wrongSets[viewIndex][classifierIndex])
if len(intersect) < bestLen:
bestLen = len(intersect)
bestCombination = combination
return [classifiersNames[viewIndex][index] for viewIndex, index in enumerate(bestCombination)]
def bestScore(allClassifersNames, directory, viewsIndices, resultsMonoview, classificationIndices):
nbViews = len(viewsIndices)
nbClassifiers = len(allClassifersNames)
scores = np.zeros((nbViews, nbClassifiers))
classifiersNames = [[] for _ in viewsIndices]
metricName = resultsMonoview[0].metrics_scores.keys()[0]
metricModule = getattr(metrics, metricName)
if metricModule.getConfig()[-14] == "h":
betterHigh = True
else:
betterHigh = False
for resultMonoview in resultsMonoview:
if resultMonoview.classifier_name not in classifiersNames[resultMonoview.view_index]:
classifiersNames[resultMonoview.view_index].append(resultMonoview.classifier_name)
classifierIndex = classifiersNames[resultMonoview.view_index].index(resultMonoview.classifier_name)
scores[resultMonoview.view_index, classifierIndex] = resultMonoview.metrics_scores.values()[0][0]
if betterHigh:
classifierIndices = np.argmax(scores, axis=1)
else:
classifierIndices = np.argmin(scores, axis=1)
return [classifiersNames[viewIndex][index] for viewIndex, index in enumerate(classifierIndices)]
def getClassifiers(selectionMethodName, allClassifiersNames, directory, viewsIndices, resultsMonoview,
classificationIndices):
thismodule = sys.modules[__name__]
selectionMethod = getattr(thismodule, selectionMethodName)
classifiersNames = selectionMethod(allClassifiersNames, directory, viewsIndices, resultsMonoview,
classificationIndices)
return classifiersNames
def getConfig(classifiersNames, resultsMonoview, viewsIndices):
classifiersConfigs = [0 for _ in range(len(classifiersNames))]
for classifierIndex, classifierName in enumerate(classifiersNames):
for resultMonoview in resultsMonoview:
if resultMonoview.view_index == viewsIndices[classifierIndex] and resultMonoview.classifier_name == classifierName:
classifiersConfigs[classifierIndex] = resultMonoview.classifier_config
return classifiersConfigs
class LateFusionClassifier(object):
def __init__(self, randomState, monoviewClassifiersNames, monoviewClassifiersConfigs, monoviewSelection,
NB_CORES=1):
self.monoviewClassifiersNames = monoviewClassifiersNames
monoviewClassifiersModules = [getattr(monoview_classifiers, classifierName)
for classifierName in self.monoviewClassifiersNames]
if type(monoviewClassifiersConfigs[0]) == dict:
self.monoviewClassifiers = [
getattr(monoviewClassifiersModule, classifierName)(random_state=randomState, **config)
for monoviewClassifiersModule, config, classifierName
in zip(monoviewClassifiersModules, monoviewClassifiersConfigs, monoviewClassifiersNames)]
else:
self.monoviewClassifiers = [
getattr(monoviewClassifiersModule, classifierName)(random_state=randomState,)
for monoviewClassifiersModule, config, classifierName
in zip(monoviewClassifiersModules, monoviewClassifiersConfigs, monoviewClassifiersNames)]
self.nbCores = NB_CORES
self.accuracies = np.zeros(len(monoviewClassifiersNames))
self.needProbas = False
self.monoviewSelection = monoviewSelection
self.randomState = randomState
def fit_hdf5(self, DATASET, labels, trainIndices=None, viewsIndices=None):
if type(viewsIndices) == type(None):
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
if trainIndices is None:
trainIndices = range(DATASET.get("Metadata").attrs["datasetLength"])
self.monoviewClassifiers = Parallel(n_jobs=self.nbCores)(
delayed(fitMonoviewClassifier)(self.monoviewClassifiers[index],
get_v(DATASET, viewIndex, trainIndices),
labels[trainIndices],
self.needProbas, self.randomState)
for index, viewIndex in enumerate(viewsIndices))
import numpy as np
from sklearn.metrics import accuracy_score
import pkgutil
from .....utils.dataset import get_v
from ..... import monoview_classifiers
from ..LateFusion import LateFusionClassifier, getClassifiers, getConfig
def genParamsSets(classificationKWARGS, randomState, nIter=1):
nbView = classificationKWARGS["nbView"]
paramsSets = []
for _ in range(nIter):
randomWeightsArray = randomState.random_sample(nbView)
normalizedArray = randomWeightsArray / np.sum(randomWeightsArray)
paramsSets.append([normalizedArray])
return paramsSets
def getArgs(benchmark, args, views, viewsIndices, directory, resultsMonoview, classificationIndices):
if args.FU_L_cl_names != ['']:
args.FU_L_select_monoview = "user_defined"
else:
monoviewClassifierModulesNames = benchmark["monoview"]
args.FU_L_cl_names = getClassifiers(args.FU_L_select_monoview, monoviewClassifierModulesNames, directory,
viewsIndices, resultsMonoview, classificationIndices)
monoviewClassifierModules = [getattr(monoview_classifiers, classifierName)
for classifierName in args.FU_L_cl_names]
if args.FU_L_cl_names == [""] and args.CL_type == ["multiview"]:
raise AttributeError("You must perform monoview classification or specify "
"which monoview classifier to use Late fusion")
if args.FU_L_cl_config != ['']:
classifiersConfigs = [
monoviewClassifierModule.getKWARGS([arg.split(":") for arg in classifierConfig.split(",")])
for monoviewClassifierModule, classifierConfig
in zip(monoviewClassifierModules, args.FU_L_cl_config)]
else:
classifiersConfigs = getConfig(args.FU_L_cl_names, resultsMonoview, viewsIndices)
arguments = {"CL_type": "fusion",
"views": views,
"NB_VIEW": len(views),
"viewsIndices": viewsIndices,
"NB_CLASS": len(args.CL_classes),
"LABELS_NAMES": args.CL_classes,
"FusionKWARGS": {"fusionType": "LateFusion",
"fusionMethod": "BayesianInference",
"classifiersNames": args.FU_L_cl_names,
"classifiersConfigs": classifiersConfigs,
'fusionMethodConfig': args.FU_L_method_config,
'monoviewSelection': args.FU_L_select_monoview,
"nbView": (len(viewsIndices))}}
return [arguments]
class BayesianInference(LateFusionClassifier):
def __init__(self, randomState, NB_CORES=1, **kwargs):
LateFusionClassifier.__init__(self, randomState, kwargs['classifiersNames'], kwargs['classifiersConfigs'],
kwargs["monoviewSelection"],
NB_CORES=NB_CORES)
if kwargs['fusionMethodConfig'][0] is None or kwargs['fusionMethodConfig'] == ['']:
self.weights = np.array([1.0 for _ in kwargs['classifiersNames']])
else:
self.weights = np.array(map(float, kwargs['fusionMethodConfig'][0]))
self.needProbas = True
def setParams(self, paramsSet):
self.weights = paramsSet[0]
def predict_hdf5(self, DATASET, usedIndices=None, viewsIndices=None):
if viewsIndices is None:
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
nbView = len(viewsIndices)
if usedIndices is None:
usedIndices = range(DATASET.get("Metadata").attrs["datasetLength"])
if sum(self.weights) != 1.0:
self.weights = self.weights / sum(self.weights)
viewScores = []#np.zeros((nbView, len(usedIndices), DATASET.get("Metadata").attrs["nbClass"]))
for index, viewIndex in enumerate(viewsIndices):
viewScores.append(np.power(
self.monoviewClassifiers[index].predict_proba(get_v(DATASET, viewIndex, usedIndices)),
self.weights[index]))
viewScores = np.array(viewScores)
predictedLabels = np.argmax(np.prod(viewScores, axis=0), axis=1)
return predictedLabels
def getConfig(self, fusionMethodConfig, monoviewClassifiersNames, monoviewClassifiersConfigs):
configString = "with Bayesian Inference using a weight for each view : " + ", ".join(map(str, self.weights)) + \
"\n\t-With monoview classifiers : "
for monoviewClassifier in self.monoviewClassifiers:
configString += monoviewClassifier.getConfig()
configString += "\n\t -Method used to select monoview classifiers : " + self.monoviewSelection
return configString
import numpy as np
# from sklearn.metrics import accuracy_score
# import pkgutil
from .....utils.dataset import get_v
from ..LateFusion import LateFusionClassifier, getClassifiers, getConfig
from ..... import monoview_classifiers
def genParamsSets(classificationKWARGS, randomState, nIter=1):
nbView = classificationKWARGS["nbView"]
paramsSets = []
for _ in range(nIter):
randomWeightsArray = randomState.random_sample(nbView)
normalizedArray = randomWeightsArray / np.sum(randomWeightsArray)
paramsSets.append([normalizedArray])
return paramsSets
def getArgs(benchmark, args, views, viewsIndices, directory, resultsMonoview, classificationIndices):
if args.FU_L_cl_names != ['']:
pass
else:
monoviewClassifierModulesNames = benchmark["monoview"]
args.FU_L_cl_names = getClassifiers(args.FU_L_select_monoview, monoviewClassifierModulesNames, directory,
viewsIndices, resultsMonoview, classificationIndices)
monoviewClassifierModules = [getattr(monoview_classifiers, classifierName)
for classifierName in args.FU_L_cl_names]
if args.FU_L_cl_names == [""] and args.CL_type == ["multiview"]:
raise AttributeError("You must perform monoview classification or specify "
"which monoview classifier to use Late fusion")
if args.FU_L_cl_config != ['']:
classifiersConfigs = [
monoviewClassifierModule.getKWARGS([arg.split(":") for arg in classifierConfig.split(",")])
for monoviewClassifierModule, classifierConfig
in zip(monoviewClassifierModules, args.FU_L_cl_config)]
else:
classifiersConfigs = getConfig(args.FU_L_cl_names, resultsMonoview, viewsIndices)
arguments = {"CL_type": "fusion",
"views": views,
"NB_VIEW": len(views),
"viewsIndices": viewsIndices,
"NB_CLASS": len(args.CL_classes),
"LABELS_NAMES": args.CL_classes,
"FusionKWARGS": {"fusionType": "LateFusion",
"fusionMethod": "MajorityVoting",
"classifiersNames": args.FU_L_cl_names,
"classifiersConfigs": classifiersConfigs,
'fusionMethodConfig': args.FU_L_method_config,
'monoviewSelection': args.FU_L_select_monoview,
"nbView": (len(viewsIndices))}}
return [arguments]
class MajorityVoting(LateFusionClassifier):
def __init__(self, randomState, NB_CORES=1, **kwargs):
LateFusionClassifier.__init__(self, randomState, kwargs['classifiersNames'], kwargs['classifiersConfigs'],
kwargs["monoviewSelection"],
NB_CORES=NB_CORES)
if kwargs['fusionMethodConfig'][0] is None or kwargs['fusionMethodConfig'] == ['']:
self.weights = np.ones(len(kwargs["classifiersNames"]), dtype=float)
else:
self.weights = np.array(map(float, kwargs['fusionMethodConfig'][0]))
def setParams(self, paramsSet):
self.weights = np.array(paramsSet[0])
def predict_hdf5(self, DATASET, usedIndices=None, viewsIndices=None):
if type(viewsIndices) == type(None):
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
nbView = len(viewsIndices)
self.weights /= float(sum(self.weights))
if usedIndices is None:
usedIndices = range(DATASET.get("Metadata").attrs["datasetLength"])
datasetLength = len(usedIndices)
votes = np.zeros((datasetLength, DATASET.get("Metadata").attrs["nbClass"]), dtype=float)
monoViewDecisions = np.zeros((len(usedIndices), nbView), dtype=int)
for index, viewIndex in enumerate(viewsIndices):
monoViewDecisions[:, index] = self.monoviewClassifiers[index].predict(
get_v(DATASET, viewIndex, usedIndices))
for exampleIndex in range(datasetLength):
for viewIndex, featureClassification in enumerate(monoViewDecisions[exampleIndex, :]):
votes[exampleIndex, featureClassification] += self.weights[viewIndex]
nbMaximum = len(np.where(votes[exampleIndex] == max(votes[exampleIndex]))[0])
try:
assert nbMaximum != nbView
except:
print("Majority voting can't decide, each classifier has voted for a different class")
raise
predictedLabels = np.argmax(votes, axis=1)
# Can be upgraded by restarting a new classification process if
# there are multiple maximums ?:
# while nbMaximum>1:
# relearn with only the classes that have a maximum number of vote
# votes = revote
# nbMaximum = len(np.where(votes==max(votes))[0])
return predictedLabels
def getConfig(self, fusionMethodConfig, monoviewClassifiersNames, monoviewClassifiersConfigs):
configString = "with Majority Voting \n\t-With weights : " + str(
self.weights) + "\n\t-With monoview classifiers : "
for monoviewClassifier in self.monoviewClassifiers:
configString += monoviewClassifier.getConfig()
return configString
import numpy as np
from pyscm.scm import SetCoveringMachineClassifier as scm
from sklearn.base import BaseEstimator, ClassifierMixin
from sklearn.externals.six import iteritems, iterkeys, itervalues
from sklearn.metrics import accuracy_score
import itertools
from ..LateFusion import LateFusionClassifier, getClassifiers, getConfig
from ..... import monoview_classifiers
from .....utils.dataset import get_v
class DecisionStumpSCMNew(BaseEstimator, ClassifierMixin):
"""docstring for SCM
A hands on class of SCM using decision stump, built with sklearn format in order to use sklearn function on SCM like
CV, gridsearch, and so on ..."""
def __init__(self, model_type='conjunction', p=0.1, max_rules=10, random_state=42):
super(DecisionStumpSCMNew, self).__init__()
self.model_type = model_type
self.p = p
self.max_rules = max_rules
self.random_state = random_state
def fit(self, X, y):
self.clf = scm(model_type=self.model_type, max_rules=self.max_rules, p=self.p, random_state=self.random_state)
self.clf.fit(X=X, y=y)
def predict(self, X):
return self.clf.predict(X)
def set_params(self, **params):
for key, value in iteritems(params):
if key == 'p':
self.p = value
if key == 'model_type':
self.model_type = value
if key == 'max_rules':
self.max_rules = value
def get_stats(self):
return {"Binary_attributes": self.clf.model_.rules}
def genParamsSets(classificationKWARGS, randomState, nIter=1):
paramsSets = []
for _ in range(nIter):
max_attributes = randomState.randint(1, 20)
p = randomState.random_sample()
model = randomState.choice(["conjunction", "disjunction"])
order = randomState.randint(1, 10)
paramsSets.append([p, max_attributes, model, order])
return paramsSets
def getArgs(benchmark, args, views, viewsIndices, directory, resultsMonoview, classificationIndices):
if args.FU_L_cl_names != ['']:
pass
else:
monoviewClassifierModulesNames = benchmark["monoview"]
args.FU_L_cl_names = getClassifiers(args.FU_L_select_monoview, monoviewClassifierModulesNames, directory,
viewsIndices, resultsMonoview, classificationIndices)
monoviewClassifierModules = [getattr(monoview_classifiers, classifierName)
for classifierName in args.FU_L_cl_names]
if args.FU_L_cl_names == [""] and args.CL_type == ["multiview"]:
raise AttributeError("You must perform monoview classification or specify "
"which monoview classifier to use Late fusion")
if args.FU_L_cl_config != ['']:
classifiersConfigs = [
monoviewClassifierModule.getKWARGS([arg.split(":") for arg in classifierConfig.split(",")])
for monoviewClassifierModule, classifierConfig
in zip(monoviewClassifierModules, args.FU_L_cl_config)]
else:
classifiersConfigs = getConfig(args.FU_L_cl_names, resultsMonoview, viewsIndices)
arguments = {"CL_type": "fusion",
"views": views,
"NB_VIEW": len(views),
"viewsIndices": viewsIndices,
"NB_CLASS": len(args.CL_classes),
"LABELS_NAMES": args.CL_classes,
"FusionKWARGS": {"fusionType": "LateFusion",
"fusionMethod": "SCMForLinear",
"classifiersNames": args.FU_L_cl_names,
"classifiersConfigs": classifiersConfigs,
'fusionMethodConfig': args.FU_L_method_config,
'monoviewSelection': args.FU_L_select_monoview,
"nbView": (len(viewsIndices))}}
return [arguments]
class SCMForLinear(LateFusionClassifier):
def __init__(self, randomState, NB_CORES=1, **kwargs):
LateFusionClassifier.__init__(self, randomState, kwargs['classifiersNames'], kwargs['classifiersConfigs'],
kwargs["monoviewSelection"],
NB_CORES=NB_CORES)
self.SCMClassifier = None
if kwargs['fusionMethodConfig'][0] is None or kwargs['fusionMethodConfig'] == ['']:
self.p = 1
self.maxAttributes = 5
self.order = 1
self.modelType = "conjunction"
else:
self.p = int(kwargs['fusionMethodConfig'][0])
self.maxAttributes = int(kwargs['fusionMethodConfig'][1])
self.order = int(kwargs['fusionMethodConfig'][2])
self.modelType = kwargs['fusionMethodConfig'][3]
def setParams(self, paramsSet):
self.p = paramsSet[0]
self.maxAttributes = paramsSet[1]
self.order = paramsSet[3]
self.modelType = paramsSet[2]
def fit_hdf5(self, DATASET, labels, trainIndices=None, viewsIndices=None):
if viewsIndices is None:
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
if trainIndices is None:
trainIndices = range(DATASET.get("Metadata").attrs["datasetLength"])
for index, viewIndex in enumerate(viewsIndices):
self.monoviewClassifiers[index].fit(get_v(DATASET, viewIndex, trainIndices),
labels[trainIndices])
self.SCMForLinearFusionFit(DATASET, labels, usedIndices=trainIndices, viewsIndices=viewsIndices)
def predict_hdf5(self, DATASET, usedIndices=None, viewsIndices=None):
if viewsIndices is None:
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
nbView = len(viewsIndices)
if usedIndices is None:
usedIndices = range(DATASET.get("Metadata").attrs["datasetLength"])
monoviewDecisions = np.zeros((len(usedIndices), nbView), dtype=int)
# accus = []
for index, viewIndex in enumerate(viewsIndices):
monoviewDecision = self.monoviewClassifiers[index].predict(
get_v(DATASET, viewIndex, usedIndices))
# accus.append(accuracy_score(DATASET.get("Labels").value[usedIndices], monoviewDecision))
monoviewDecisions[:, index] = monoviewDecision
features = self.generateInteractions(monoviewDecisions)
predictedLabels = self.SCMClassifier.predict(features)
return predictedLabels
def SCMForLinearFusionFit(self, DATASET, labels, usedIndices=None, viewsIndices=None):
if type(viewsIndices) == type(None):
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
nbView = len(viewsIndices)
self.SCMClassifier = DecisionStumpSCMNew(p=self.p, max_rules=self.maxAttributes, model_type=self.modelType,
random_state=self.randomState)
monoViewDecisions = np.zeros((len(usedIndices), nbView), dtype=int)
for index, viewIndex in enumerate(viewsIndices):
monoViewDecisions[:, index] = self.monoviewClassifiers[index].predict(
get_v(DATASET, viewIndex, usedIndices))
features = self.generateInteractions(monoViewDecisions)
features = np.array([np.array([feat for feat in feature]) for feature in features])
self.SCMClassifier.fit(features, labels[usedIndices].astype(int))
def generateInteractions(self, monoViewDecisions):
if type(self.order) == type(None):
self.order = monoViewDecisions.shape[1]
if self.order == 1:
return monoViewDecisions
else:
genratedIntercations = [monoViewDecisions[:, i] for i in range(monoViewDecisions.shape[1])]
for orderIndex in range(self.order - 1):
combins = itertools.combinations(range(monoViewDecisions.shape[1]), orderIndex + 2)
for combin in combins:
generatedDecision = monoViewDecisions[:, combin[0]]
for index in range(len(combin) - 1):
if self.modelType == "disjunction":
generatedDecision = np.logical_and(generatedDecision,
monoViewDecisions[:, combin[index + 1]])
else:
generatedDecision = np.logical_or(generatedDecision,
monoViewDecisions[:, combin[index + 1]])
genratedIntercations.append(generatedDecision)
return np.transpose(np.array(genratedIntercations))
def getConfig(self, fusionMethodConfig, monoviewClassifiersNames, monoviewClassifiersConfigs):
configString = "with SCM for linear with max_attributes : " + str(self.maxAttributes) + ", p : " + str(self.p) + \
" model_type : " + str(self.modelType) + " order : " + str(self.order)+ " has chosen " + \
str(0.1) + " rule(s) \n\t-With monoview classifiers : "
for monoviewClassifier in self.monoviewClassifiers:
configString += monoviewClassifier.getConfig()
return configString
import numpy as np
from sklearn.multiclass import OneVsOneClassifier
from sklearn.svm import SVC
import pkgutil
from ..LateFusion import LateFusionClassifier, getClassifiers, getConfig
from .....utils.dataset import get_v
from ..... import monoview_classifiers
def genParamsSets(classificationKWARGS, randomState, nIter=1):
paramsSets = []
for _ in range(nIter):
paramsSets.append([])
return paramsSets
def getArgs(benchmark, args, views, viewsIndices, directory, resultsMonoview, classificationIndices):
if args.FU_L_cl_names != ['']:
pass
else:
monoviewClassifierModulesNames = benchmark["monoview"]
args.FU_L_cl_names = getClassifiers(args.FU_L_select_monoview, monoviewClassifierModulesNames, directory,
viewsIndices, resultsMonoview, classificationIndices)
monoviewClassifierModules = [getattr(monoview_classifiers, classifierName)
for classifierName in args.FU_L_cl_names]
if args.FU_L_cl_names == [""] and args.CL_type == ["multiview"]:
raise AttributeError("You must perform monoview classification or specify "
"which monoview classifier to use Late fusion")
if args.FU_L_cl_config != ['']:
classifiersConfigs = [
monoviewClassifierModule.getKWARGS([arg.split(":") for arg in classifierConfig.split(",")])
for monoviewClassifierModule, classifierConfig
in zip(monoviewClassifierModules, args.FU_L_cl_config)]
else:
classifiersConfigs = getConfig(args.FU_L_cl_names, resultsMonoview, viewsIndices)
arguments = {"CL_type": "fusion",
"views": views,
"NB_VIEW": len(views),
"viewsIndices": viewsIndices,
"NB_CLASS": len(args.CL_classes),
"LABELS_NAMES": args.CL_classes,
"FusionKWARGS": {"fusionType": "LateFusion",
"fusionMethod": "SVMForLinear",
"classifiersNames": args.FU_L_cl_names,
"classifiersConfigs": classifiersConfigs,
'fusionMethodConfig': args.FU_L_method_config,
'monoviewSelection': args.FU_L_select_monoview,
"nbView": (len(viewsIndices))}}
return [arguments]
class SVMForLinear(LateFusionClassifier):
def __init__(self, randomState, NB_CORES=1, **kwargs):
LateFusionClassifier.__init__(self, randomState, kwargs['classifiersNames'], kwargs['classifiersConfigs'],
kwargs["monoviewSelection"],
NB_CORES=NB_CORES)
self.SVMClassifier = None
def fit_hdf5(self, DATASET, labels, trainIndices=None, viewsIndices=None):
if viewsIndices is None:
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
if trainIndices is None:
trainIndices = range(DATASET.get("Metadata").attrs["datasetLength"])
for index, viewIndex in enumerate(viewsIndices):
self.monoviewClassifiers[index].fit(get_v(DATASET, viewIndex, trainIndices),
labels[trainIndices])
self.SVMForLinearFusionFit(DATASET, labels, usedIndices=trainIndices, viewsIndices=viewsIndices)
def setParams(self, paramsSet):
pass
def predict_hdf5(self, DATASET, usedIndices=None, viewsIndices=None):
if viewsIndices is None:
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
nbView = len(viewsIndices)
if usedIndices is None:
usedIndices = range(DATASET.get("Metadata").attrs["datasetLength"])
monoviewDecisions = np.zeros((len(usedIndices), nbView), dtype=int)
for index, viewIndex in enumerate(viewsIndices):
monoviewDecisions[:, index] = self.monoviewClassifiers[index].predict(
get_v(DATASET, viewIndex, usedIndices))
predictedLabels = self.SVMClassifier.predict(monoviewDecisions)
return predictedLabels
def SVMForLinearFusionFit(self, DATASET, labels, usedIndices=None, viewsIndices=None):
if type(viewsIndices) == type(None):
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
nbView = len(viewsIndices)
self.SVMClassifier = OneVsOneClassifier(SVC())
monoViewDecisions = np.zeros((len(usedIndices), nbView), dtype=int)
for index, viewIndex in enumerate(viewsIndices):
monoViewDecisions[:, index] = self.monoviewClassifiers[index].predict(
get_v(DATASET, viewIndex, usedIndices))
self.SVMClassifier.fit(monoViewDecisions, labels[usedIndices])
def getConfig(self, fusionMethodConfig, monoviewClassifiersNames, monoviewClassifiersConfigs):
configString = "with SVM for linear \n\t-With monoview classifiers : "
for monoviewClassifier in self.monoviewClassifiers:
configString += monoviewClassifier.getConfig()
return configString
import numpy as np
from sklearn.metrics import accuracy_score
import pkgutil
from ..... import monoview_classifiers
from ..LateFusion import LateFusionClassifier, getClassifiers, getConfig
from .....utils.dataset import get_v
def genParamsSets(classificationKWARGS, randomState, nIter=1):
nbView = classificationKWARGS["nbView"]
paramsSets = []
for _ in range(nIter):
randomWeightsArray = randomState.random_sample(nbView)
normalizedArray = randomWeightsArray / np.sum(randomWeightsArray)
paramsSets.append([normalizedArray])
return paramsSets
def getArgs(benchmark, args, views, viewsIndices, directory, resultsMonoview, classificationIndices):
if args.FU_L_cl_names != ['']:
pass
else:
monoviewClassifierModulesNames = benchmark["monoview"]
args.FU_L_cl_names = getClassifiers(args.FU_L_select_monoview, monoviewClassifierModulesNames, directory,
viewsIndices, resultsMonoview, classificationIndices)
monoviewClassifierModules = [getattr(monoview_classifiers, classifierName)
for classifierName in args.FU_L_cl_names]
if args.FU_L_cl_names == [""] and args.CL_type == ["multiview"]:
raise AttributeError("You must perform monoview classification or specify "
"which monoview classifier to use Late fusion")
if args.FU_L_cl_config != ['']:
classifiersConfigs = [
monoviewClassifierModule.getKWARGS([arg.split(":") for arg in classifierConfig.split(",")])
for monoviewClassifierModule, classifierConfig
in zip(monoviewClassifierModules, args.FU_L_cl_config)]
else:
classifiersConfigs = getConfig(args.FU_L_cl_names, resultsMonoview, viewsIndices)
arguments = {"CL_type": "fusion",
"views": views,
"NB_VIEW": len(views),
"viewsIndices": viewsIndices,
"NB_CLASS": len(args.CL_classes),
"LABELS_NAMES": args.CL_classes,
"FusionKWARGS": {"fusionType": "LateFusion",
"fusionMethod": "WeightedLinear",
"classifiersNames": args.FU_L_cl_names,
"classifiersConfigs": classifiersConfigs,
'fusionMethodConfig': args.FU_L_method_config,
'monoviewSelection': args.FU_L_select_monoview,
"nbView": (len(viewsIndices))}}
return [arguments]
class WeightedLinear(LateFusionClassifier):
def __init__(self, randomState, NB_CORES=1, **kwargs):
LateFusionClassifier.__init__(self, randomState, kwargs['classifiersNames'], kwargs['classifiersConfigs'],
kwargs["monoviewSelection"],
NB_CORES=NB_CORES)
if kwargs['fusionMethodConfig'][0] is None or kwargs['fusionMethodConfig'] == ['']:
self.weights = np.ones(len(kwargs["classifiersNames"]), dtype=float)
else:
self.weights = np.array(map(float, kwargs['fusionMethodConfig'][0]))
self.needProbas = True
def setParams(self, paramsSet):
self.weights = paramsSet[0]
def predict_hdf5(self, DATASET, usedIndices=None, viewsIndices=None):
if viewsIndices is None:
viewsIndices = np.arange(DATASET.get("Metadata").attrs["nbView"])
nbView = len(viewsIndices)
self.weights /= float(sum(self.weights))
if usedIndices is None:
usedIndices = range(DATASET.get("Metadata").attrs["datasetLength"])
viewScores = []#np.zeros((nbView, len(usedIndices), DATASET.get("Metadata").attrs["nbClass"]))
for index, viewIndex in enumerate(viewsIndices):
viewScores.append(np.array(self.monoviewClassifiers[index].predict_proba(
get_v(DATASET, viewIndex, usedIndices))) * self.weights[index])
viewScores = np.array(viewScores)
predictedLabels = np.argmax(np.sum(viewScores, axis=0), axis=1)
return predictedLabels
def getConfig(self, fusionMethodConfig, monoviewClassifiersNames, monoviewClassifiersConfigs):
configString = "with Weighted linear using a weight for each view : " + ", ".join(map(str, self.weights)) + \
"\n\t-With monoview classifiers : "
for monoviewClassifier in self.monoviewClassifiers:
configString += monoviewClassifier.getConfig()
return configString
import os
for module in os.listdir(os.path.dirname(os.path.realpath(__file__))):
if module == '__init__.py' or module[-3:] != '.py':
continue
__import__(module[:-3], locals(), globals(), [], 1)
del module
del os
# import pdb;pdb.set_trace()
from . import EarlyFusion
from . import LateFusion
from . import LateFusionPackage
from . import EarlyFusionPackage
__all__ = ["EarlyFusionPackage", "LateFusionPackage"]
from . import fusion, analyze_results, Methods
__all__ = ["fusion.py", "Methods"]
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment