Source code for autogluon.mxnet.task.classifier
import copy
import math
import os
import warnings
from collections import OrderedDict, defaultdict
import cloudpickle as pkl
import matplotlib.pyplot as plt
import mxnet as mx
from PIL import Image
from mxnet.gluon.data.vision import transforms
from .metrics import get_metric_instance
from .nets import get_network
from .utils import *
from ..utils import collect_params, update_params
from autogluon.core import AutoGluonObject
from autogluon.core.utils import save, load, tqdm
from autogluon.core.task.base import BasePredictor
__all__ = ['Classifier']
[docs]class Classifier(BasePredictor):
"""Trained Image Classifier returned by fit() that can be used to make predictions on new images.
Deprecated: please use autogluon.vision.ImagePredictor starting v0.1.0.
Attributes
----------
Examples
--------
>>> import autogluon.core as ag
>>> from autogluon.vision import ImagePredictor
>>> dataset = ImagePredictor.Dataset(train_path='data/train',
>>> test_path='data/test')
>>> classifier = ImagePredictor().fit(dataset,
>>> nets=ag.space.Categorical['resnet18_v1', 'resnet34_v1'],
>>> time_limit=time_limit,
>>> ngpus_per_trial=1)
>>> image = 'data/test/BabyShirt/BabyShirt_323.jpg'
>>> ind, prob = classifier.predict(image)
"""
def __init__(self, model, results, eval_func, scheduler_checkpoint,
args, ensemble=0, format_results=True, **kwargs):
warnings.warn('Classifier is deprecated starting v0.1.0, please use `autogluon.vision.ImagePredictor`.')
self.model = model
self.eval_func = eval_func
self.results = self._format_results(results) if format_results else results
self.scheduler_checkpoint = scheduler_checkpoint
self.args = args
self.ensemble = ensemble
[docs] @classmethod
def load(cls, checkpoint):
"""Load trained Image Classifier from directory specified by `checkpoint`.
"""
state_dict = load(checkpoint)
args = state_dict['args']
results = pkl.loads(state_dict['results'])
eval_func = state_dict['eval_func']
scheduler_checkpoint = state_dict['scheduler_checkpoint']
model_params = state_dict['model_params']
ensemble = state_dict['ensemble']
if ensemble <= 1:
model_args = copy.deepcopy(args)
model_args.update(results['best_config'])
model = get_network(args.net, num_classes=results['num_classes'], ctx=mx.cpu(0))
update_params(model, model_params)
else:
raise NotImplemented
return cls(model, results, eval_func, scheduler_checkpoint, args,
ensemble, format_results=False)
def state_dict(self, destination=None):
if destination is None:
destination = OrderedDict()
destination._metadata = OrderedDict()
model_params = collect_params(self.model)
destination['model_params'] = model_params
destination['eval_func'] = self.eval_func
destination['results'] = pkl.dumps(self.results)
destination['scheduler_checkpoint'] = self.scheduler_checkpoint
destination['args'] = self.args
destination['ensemble'] = self.ensemble
return destination
[docs] def save(self, checkpoint):
""" Save image classifier to folder specified by `checkpoint`.
"""
state_dict = self.state_dict()
save(state_dict, checkpoint)
[docs] def predict(self, X, input_size=224, crop_ratio=0.875, set_prob_thresh=0.001, plot=False):
"""Predict class-index and associated class probability for each image in a given dataset (or just a single image).
Parameters
----------
X : str or :class:`autogluon.vision.ImagePredictor.Dataset` or list of `autogluon.vision.ImagePredictor.Dataset`
If str, should be path to the input image (when we just want to predict on single image).
If class:`autogluon.vision.ImagePredictor.Dataset`, should be dataset of multiple images in same format as training dataset.
If list of `autogluon.vision.ImagePredictor.Dataset`, should be a set of test dataset with different scales of origin images.
input_size : int
Size of the images (pixels).
plot : bool
Whether to plot the image being classified.
set_prob_thresh: float
Results with probability below threshold are set to 0 by default.
Examples
--------
>>> import autogluon.core as ag
>>> from autogluon.vision import ImagePredictor
>>> train_data = ImagePredictor.Dataset(train_path='~/data/train')
>>> classifier = ImagePredictor().fit(train_data,
>>> nets=ag.space.Categorical['resnet18_v1', 'resnet34_v1'],
>>> time_limit=600, ngpus_per_trial=1)
>>> test_data = ImagePredictor.Dataset('~/data/test', train=False)
>>> class_index, class_probability = classifier.predict('example.jpg')
"""
input_size = self.model.input_size if hasattr(self.model, 'input_size') else input_size
resize = int(math.ceil(input_size / crop_ratio))
transform_size = transforms.Compose([
transforms.Resize(resize),
transforms.CenterCrop(input_size),
transforms.ToTensor(),
transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
])
def predict_img(img, ensemble=False):
proba = self.predict_proba(img)
if ensemble:
return proba
else:
ind = mx.nd.argmax(proba, axis=1).astype('int')
idx = mx.nd.stack(mx.nd.arange(proba.shape[0], ctx=proba.context), ind.astype('float32'))
probai = mx.nd.gather_nd(proba, idx)
return ind, probai, proba
def avg_prediction(different_dataset, threshold=0.001):
result = defaultdict(list)
inds, probas, probals_all = [], [], []
for i in range(len(different_dataset)):
for j in range(len(different_dataset[0])):
result[j].append(different_dataset[i][j])
for c in result.keys():
proba_all = sum([*result[c]]) / len(different_dataset)
proba_all = (proba_all >= threshold) * proba_all
ind = mx.nd.argmax(proba_all, axis=1).astype('int')
idx = mx.nd.stack(mx.nd.arange(proba_all.shape[0], ctx=proba_all.context), ind.astype('float32'))
proba = mx.nd.gather_nd(proba_all, idx)
inds.append(ind.asscalar())
probas.append(proba.asnumpy())
probals_all.append(proba_all.asnumpy().flatten())
return inds, probas, probals_all
def predict_imgs(X):
if isinstance(X, list):
different_dataset = []
for i, x in enumerate(X):
proba_all_one_dataset = []
tbar = tqdm(range(len(x.items)))
for j, x_item in enumerate(x):
tbar.update(1)
proba_all = predict_img(x_item[0], ensemble=True)
tbar.set_description('ratio:[%d],The input picture [%d]' % (i, j))
proba_all_one_dataset.append(proba_all)
different_dataset.append(proba_all_one_dataset)
inds, probas, probals_all = avg_prediction(different_dataset, threshold=set_prob_thresh)
else:
inds, probas, probals_all = [], [], []
tbar = tqdm(range(len(X.items)))
for i, x in enumerate(X):
tbar.update(1)
ind, proba, proba_all = predict_img(x[0])
tbar.set_description(
'The input picture [%d] is classified as [%d], with probability %.2f ' %
(i, ind.asscalar(), proba.asscalar())
)
inds.append(ind.asscalar())
probas.append(proba.asnumpy())
probals_all.append(proba_all.asnumpy().flatten())
return inds, probas, probals_all
if isinstance(X, str) and os.path.isfile(X):
img = mx.image.imread(filename=X)
if plot:
plt.imshow(img.asnumpy())
plt.show()
img = transform_size(img)
return predict_img(img)
if isinstance(X, AutoGluonObject):
X = X.init()
return predict_imgs(X)
if isinstance(X, list) and len(X) > 1:
X_group = []
for X_item in X:
X_item = X_item.init()
X_group.append(X_item)
return predict_imgs(X_group)
@staticmethod
def loader(path):
with open(path, 'rb') as f:
img = Image.open(f)
return img.convert('RGB')
def predict_proba(self, X):
"""Produces predicted class probabilities for a given image.
"""
pred = self.model(X.expand_dims(0))
return mx.nd.softmax(pred)
[docs] def evaluate(self, dataset, input_size=224, ctx=[mx.cpu()]):
"""Evaluate predictive performance of trained image classifier using given test data.
Parameters
----------
dataset : :class:`autogluon.vision.ImagePredictor.Dataset`
The dataset containing test images (must be in same format as the training dataset).
input_size : int
Size of the images (pixels).
ctx : List of mxnet.context elements.
Determines whether to use CPU or GPU(s), options include: `[mx.cpu()]` or `[mx.gpu()]`.
Examples
--------
>>> import autogluon.core as ag
>>> from autogluon.vision import ImagePredictor as vision
>>> train_data = ImagePredictor.Dataset(train_path='~/data/train')
>>> classifier = ImagePredictor().fit(train_data,
>>> nets=ag.space.Categorical['resnet18_v1', 'resnet34_v1'],
>>> time_limit=600, ngpus_per_trial=1)
>>> test_data = ImagePredictor.Dataset('~/data/test', train=False)
>>> test_acc = classifier.evaluate(test_data)
"""
args = self.args
net = self.model
batch_size = args.batch_size * max(len(ctx), 1)
metric = get_metric_instance(args.metric)
input_size = net.input_size if hasattr(net, 'input_size') else input_size
test_data, _, batch_fn, _ = get_data_loader(dataset, input_size, batch_size, args.num_workers, True, None)
tbar = tqdm(test_data)
for batch in tbar:
self.eval_func(net, batch, batch_fn, metric, ctx)
_, test_reward = metric.get()
tbar.set_description('{}: {}'.format(args.metric, test_reward))
_, test_reward = metric.get()
return test_reward
def evaluate_predictions(self, y_true, y_pred):
raise NotImplementedError # TODO