Nigel
/
mindspore
forked from mindspore-Ecosystem/mindspore
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

!49074 [MindSpore]概率编程toolbox接口删除 

Merge pull request !49074 from 十六夜/master
This commit is contained in:
i-robot 2023-02-20 08:31:35 +00:00 committed by Gitee
parent 9aeab9e7d4 20fa03cbae
commit fcabc30952
No known key found for this signature in database
GPG Key ID: 173E9B9CA92EEF8F
6 changed files with 0 additions and 705 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
docs/api/api_python/probability/mindspore.nn.probability.toolbox.UncertaintyEvaluation.rst
View File

@ -1,39 +0,0 @@
mindspore.nn.probability.toolbox.UncertaintyEvaluation
======================================================
.. py:class:: mindspore.nn.probability.toolbox.UncertaintyEvaluation(model, train_dataset, task_type, num_classes=None, epochs=1, epi_uncer_model_path=None, ale_uncer_model_path=None, save_model=False)
包含数据不确定性和模型不确定性的评估工具箱。
参数：
- **model** (Cell) - 不确定性评估的模型。
- **train_dataset** (Dataset) - 用于训练模型的数据集迭代器。
- **task_type** (str) - 模型任务类型的选项。
- regression：回归模型。
- classification：分类模型。
- **num_classes** (int) - 分类标签的数量。如果任务类型为分类，则必须设置；否则，它是不需要的。默认值：None。
- **epochs** (int) - 数据的迭代总数。默认值：1。
- **epi_uncer_model_path** (str) - 认知不确定性模型的保存或读取路径。默认值：None。
- **ale_uncer_model_path** (str) - 任意不确定性模型的保存或读取路径。默认值：None。
- **save_model** (bool) - 是否保存不确定性模型，如果为 true，`epi_uncer_model_path` 和 `ale_uncer_model_path` 不能为 None。
如果为 false，则从不确定性模型的路径中加载要评估的模型；如果未给出路径，则不会保存或加载不确定性模型。默认值：false。
.. py:method:: eval_aleatoric_uncertainty(eval_data)
评估推理结果的任意不确定性，也称为数据不确定性。
参数：
- **eval_data** (Tensor) - 要评估的数据样本，shape 必须是 (N,C,H,W)。
返回：
numpy.dtype，数据样本推断结果的任意不确定性。
.. py:method:: eval_epistemic_uncertainty(eval_data)
评估推理结果的认知不确定性，也称为模型不确定性。
参数：
- **eval_data** (Tensor) - 要评估的数据样本，shape 必须是 (N,C,H,W)。
返回：
numpy.dtype，数据样本推断结果的任意不确定性。

46
docs/api/api_python/probability/mindspore.nn.probability.toolbox.VAEAnomalyDetection.rst
View File

@ -1,46 +0,0 @@
mindspore.nn.probability.toolbox.VAEAnomalyDetection
====================================================
.. py:class:: mindspore.nn.probability.toolbox.VAEAnomalyDetection(encoder, decoder, hidden_size=400, latent_size=20)
使用 VAE 进行异常检测的工具箱。
变分自动编码器（VAE）可用于无监督异常检测。异常分数是 sample_x 与重建 sample_x 之间的误差。如果分数高，则 X 大多是异常值。
参数：
- **encoder** (Cell) - 定义为编码器的深度神经网络 (DNN) 模型。
- **decoder** (Cell) - 定义为解码器的深度神经网络 (DNN) 模型。
- **hidden_size** (int) - 编码器输出 Tensor 的大小。默认值：400。
- **latent_size** (int) - 潜在空间的大小。默认值：20。
.. py:method:: predict_outlier(sample_x, threshold=100.0)
预测样本是否为异常值。
参数：
- **sample_x** (Tensor) - 待预测的样本，shape 为 (N, C, H, W)。
- **threshold** (float) - 异常值的阈值。默认值：100.0。
返回：
bool，样本是否为异常值。
.. py:method:: predict_outlier_score(sample_x)
预测异常值分数。
参数：
- **sample_x** (Tensor) - 待预测的样本，shape 为 (N, C, H, W)。
返回：
float，样本的预测异常值分数。
.. py:method:: train(train_dataset, epochs=5)
训练 VAE 模型。
参数：
- **train_dataset** (Dataset) - 用于训练模型的数据集迭代器。
- **epochs** (int) - 数据的迭代总数。默认值：5。
返回：
Cell，训练完的模型。

22
mindspore/python/mindspore/nn/probability/toolbox/__init__.py
View File

@ -1,22 +0,0 @@
# Copyright 2020 Huawei Technologies Co., Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============================================================================
"""
Uncertainty toolbox.
"""
from .uncertainty_evaluation import UncertaintyEvaluation
from .anomaly_detection import VAEAnomalyDetection
__all__ = ['UncertaintyEvaluation', 'VAEAnomalyDetection']

99
mindspore/python/mindspore/nn/probability/toolbox/anomaly_detection.py
View File

@ -1,99 +0,0 @@
# Copyright 2020 Huawei Technologies Co., Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============================================================================
"""Toolbox for anomaly detection by using VAE."""
import numpy as np
from mindspore._checkparam import Validator
from mindspore.common.tensor import Tensor
from ..dpn import VAE
from ..infer import ELBO, SVI
from ...optim import Adam
from ...wrap.cell_wrapper import WithLossCell
class VAEAnomalyDetection:
r"""
Toolbox for anomaly detection by using VAE.
Variational Auto-Encoder(VAE) can be used for Unsupervised Anomaly Detection. The anomaly score is the error
between the X and the reconstruction of X. If the score is high, the X is mostly outlier.
Args:
encoder(Cell): The Deep Neural Network (DNN) model defined as encoder.
decoder(Cell): The DNN model defined as decoder.
hidden_size(int): The size of encoder's output tensor. Default: 400.
latent_size(int): The size of the latent space. Default: 20.
Supported Platforms:
``Ascend`` ``GPU``
"""
def __init__(self, encoder, decoder, hidden_size=400, latent_size=20):
self.vae = VAE(encoder, decoder, hidden_size, latent_size)
def train(self, train_dataset, epochs=5):
"""
Train the VAE model.
Args:
train_dataset (Dataset): A dataset iterator to train model.
epochs (int): Total number of iterations on the data. Default: 5.
Returns:
Cell, the trained model.
"""
net_loss = ELBO()
optimizer = Adam(params=self.vae.trainable_params(), learning_rate=0.001)
net_with_loss = WithLossCell(self.vae, net_loss)
vi = SVI(net_with_loss=net_with_loss, optimizer=optimizer)
self.vae = vi.run(train_dataset, epochs)
return self.vae
def predict_outlier_score(self, sample_x):
"""
Predict the outlier score.
Args:
sample_x (Tensor): The sample to be predicted, the shape is (N, C, H, W).
Returns:
float, the predicted outlier score of the sample.
"""
if not isinstance(sample_x, Tensor):
raise TypeError("The sample_x must be Tensor type.")
reconstructed_sample = self.vae.reconstruct_sample(sample_x)
return self._calculate_euclidean_distance(sample_x.asnumpy(), reconstructed_sample.asnumpy())
def predict_outlier(self, sample_x, threshold=100.0):
"""
Predict whether the sample is an outlier.
Args:
sample_x (Tensor): The sample to be predicted, the shape is (N, C, H, W).
threshold (float): the threshold of the outlier. Default: 100.0.
Returns:
Bool, whether the sample is an outlier.
"""
threshold = Validator.check_positive_float(threshold)
score = self.predict_outlier_score(sample_x)
return score >= threshold
def _calculate_euclidean_distance(self, sample_x, reconstructed_sample):
"""
Calculate the euclidean distance of the sample_x and reconstructed_sample.
"""
return np.sqrt(np.sum(np.square(sample_x - reconstructed_sample)))

364
mindspore/python/mindspore/nn/probability/toolbox/uncertainty_evaluation.py
View File

@ -1,364 +0,0 @@
# Copyright 2020-2021 Huawei Technologies Co., Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============================================================================
"""Toolbox for Uncertainty Evaluation."""
from copy import deepcopy
import numpy as np
from mindspore._checkparam import Validator
from mindspore.ops import composite as C
from mindspore.ops import operations as P
from mindspore.train import Model
from mindspore.train.callback import LossMonitor, ModelCheckpoint, CheckpointConfig
from mindspore.train.serialization import load_checkpoint, load_param_into_net
from mindspore.common import dtype as mstype
from ...cell import Cell
from ...layer.basic import Dense, Flatten, Dropout
from ...layer.container import SequentialCell
from ...layer.conv import Conv2d
from ...loss import SoftmaxCrossEntropyWithLogits, MSELoss
from ...metrics import Accuracy, MSE
from ...optim import Adam
class UncertaintyEvaluation:
r"""
Toolbox for Uncertainty Evaluation.
Args:
model (Cell): The model for uncertainty evaluation.
train_dataset (Dataset): A dataset iterator to train model.
task_type (str): Option for the task types of model
- regression: A regression model.
- classification: A classification model.
num_classes (int): The number of labels of classification.
If the task type is classification, it must be set; otherwise, it is not needed.
Default: None.
epochs (int): Total number of iterations on the data. Default: 1.
epi_uncer_model_path (str): The save or read path of the epistemic uncertainty model. Default: None.
ale_uncer_model_path (str): The save or read path of the aleatoric uncertainty model. Default: None.
save_model (bool): Whether to save the uncertainty model or not, if true, the epi_uncer_model_path
and ale_uncer_model_path must not be None. If false, the model to evaluate will be loaded from
the the path of the uncertainty model; if the path is not given , it will not save or load the
uncertainty model. Default: False.
Supported Platforms:
``Ascend`` ``GPU``
Examples:
>>> network = LeNet()
>>> ds_train = create_dataset('workspace/mnist/train') # handle train data
>>> ds_eval = create_dataset('workspace/mnist/test') # handle test data
>>> evaluation = UncertaintyEvaluation(model=network,
... train_dataset=ds_train,
... task_type='classification',
... num_classes=10,
... epochs=1,
... epi_uncer_model_path=None,
... ale_uncer_model_path=None,
... save_model=False)
>>> for eval_data in ds_eval.create_dict_iterator(output_numpy=True, num_epochs=1):
... eval_data = Tensor(eval_data['image'], mstype.float32)
... epistemic_uncertainty = evaluation.eval_epistemic_uncertainty(eval_data)
... aleatoric_uncertainty = evaluation.eval_aleatoric_uncertainty(eval_data)
>>> output = epistemic_uncertainty.shape
>>> print(output)
(32, 10)
>>> output = aleatoric_uncertainty.shape
>>> print(output)
(32,)
"""
def __init__(self, model, train_dataset, task_type, num_classes=None, epochs=1,
epi_uncer_model_path=None, ale_uncer_model_path=None, save_model=False):
self.epi_model = deepcopy(model)
self.ale_model = deepcopy(model)
self.epi_train_dataset = train_dataset
self.ale_train_dataset = deepcopy(train_dataset)
self.task_type = task_type
self.epochs = Validator.check_positive_int(epochs)
self.epi_uncer_model_path = epi_uncer_model_path
self.ale_uncer_model_path = ale_uncer_model_path
self.save_model = Validator.check_bool(save_model)
self.epi_uncer_model = None
self.ale_uncer_model = None
self.concat = P.Concat(axis=0)
self.sum = P.ReduceSum()
self.pow = P.Pow()
if not isinstance(model, Cell):
raise TypeError('The model must be Cell type.')
if task_type not in ('regression', 'classification'):
raise ValueError(
'The task should be regression or classification.')
if task_type == 'classification':
self.num_classes = Validator.check_positive_int(num_classes)
else:
self.num_classes = num_classes
if save_model:
if epi_uncer_model_path is None or ale_uncer_model_path is None:
raise ValueError("If save_model is True, the epi_uncer_model_path and "
"ale_uncer_model_path should not be None.")
def _get_epistemic_uncertainty_model(self):
"""
Get the model which can obtain the epistemic uncertainty.
"""
if self.epi_uncer_model is None:
self.epi_uncer_model = EpistemicUncertaintyModel(self.epi_model)
if self.epi_uncer_model.drop_count == 0 and self.epi_train_dataset is not None:
if self.task_type == 'classification':
net_loss = SoftmaxCrossEntropyWithLogits(
sparse=True, reduction="mean")
net_opt = Adam(self.epi_uncer_model.trainable_params())
model = Model(self.epi_uncer_model, net_loss,
net_opt, metrics={"Accuracy": Accuracy()})
else:
net_loss = MSELoss()
net_opt = Adam(self.epi_uncer_model.trainable_params())
model = Model(self.epi_uncer_model, net_loss,
net_opt, metrics={"MSE": MSE()})
if self.save_model:
config_ck = CheckpointConfig(
keep_checkpoint_max=self.epochs)
ckpoint_cb = ModelCheckpoint(prefix='checkpoint_epi_uncer_model',
directory=self.epi_uncer_model_path,
config=config_ck)
model.train(self.epochs, self.epi_train_dataset, dataset_sink_mode=False,
callbacks=[ckpoint_cb, LossMonitor()])
elif self.epi_uncer_model_path is None:
model.train(self.epochs, self.epi_train_dataset, dataset_sink_mode=False,
callbacks=[LossMonitor()])
else:
uncer_param_dict = load_checkpoint(
self.epi_uncer_model_path)
load_param_into_net(self.epi_uncer_model, uncer_param_dict)
def _eval_epistemic_uncertainty(self, eval_data, mc=10):
"""
Evaluate the epistemic uncertainty of classification and regression models using MC dropout.
"""
self._get_epistemic_uncertainty_model()
self.epi_uncer_model.set_train(True)
outputs = [None] * mc
for i in range(mc):
pred = self.epi_uncer_model(eval_data)
outputs[i] = pred.asnumpy()
if self.task_type == 'classification':
outputs = np.stack(outputs, axis=2)
epi_uncertainty = outputs.var(axis=2)
else:
outputs = np.stack(outputs, axis=1)
epi_uncertainty = outputs.var(axis=1)
epi_uncertainty = np.array(epi_uncertainty)
return epi_uncertainty
def _get_aleatoric_uncertainty_model(self):
"""
Get the model which can obtain the aleatoric uncertainty.
"""
if self.ale_train_dataset is None:
raise ValueError(
'The train dataset should not be None when evaluating aleatoric uncertainty.')
if self.ale_uncer_model is None:
self.ale_uncer_model = AleatoricUncertaintyModel(
self.ale_model, self.num_classes, self.task_type)
net_loss = AleatoricLoss(self.task_type)
net_opt = Adam(self.ale_uncer_model.trainable_params())
if self.task_type == 'classification':
model = Model(self.ale_uncer_model, net_loss,
net_opt, metrics={"Accuracy": Accuracy()})
else:
model = Model(self.ale_uncer_model, net_loss,
net_opt, metrics={"MSE": MSE()})
if self.save_model:
config_ck = CheckpointConfig(keep_checkpoint_max=self.epochs)
ckpoint_cb = ModelCheckpoint(prefix='checkpoint_ale_uncer_model',
directory=self.ale_uncer_model_path,
config=config_ck)
model.train(self.epochs, self.ale_train_dataset, dataset_sink_mode=False,
callbacks=[ckpoint_cb, LossMonitor()])
elif self.ale_uncer_model_path is None:
model.train(self.epochs, self.ale_train_dataset, dataset_sink_mode=False,
callbacks=[LossMonitor()])
else:
uncer_param_dict = load_checkpoint(self.ale_uncer_model_path)
load_param_into_net(self.ale_uncer_model, uncer_param_dict)
def _eval_aleatoric_uncertainty(self, eval_data):
"""
Evaluate the aleatoric uncertainty of classification and regression models.
"""
self._get_aleatoric_uncertainty_model()
_, var = self.ale_uncer_model(eval_data)
ale_uncertainty = self.sum(self.pow(var, 2), 1)
ale_uncertainty = ale_uncertainty.asnumpy()
return ale_uncertainty
def eval_epistemic_uncertainty(self, eval_data):
"""
Evaluate the epistemic uncertainty of inference results, which also called model uncertainty.
Args:
eval_data (Tensor): The data samples to be evaluated, the shape must be (N,C,H,W).
Returns:
numpy.dtype, the epistemic uncertainty of inference results of data samples.
"""
uncertainty = self._eval_epistemic_uncertainty(eval_data)
return uncertainty
def eval_aleatoric_uncertainty(self, eval_data):
"""
Evaluate the aleatoric uncertainty of inference results, which also called data uncertainty.
Args:
eval_data (Tensor): The data samples to be evaluated, the shape must be (N,C,H,W).
Returns:
numpy.dtype, the aleatoric uncertainty of inference results of data samples.
"""
uncertainty = self._eval_aleatoric_uncertainty(eval_data)
return uncertainty
class EpistemicUncertaintyModel(Cell):
"""
Using dropout during training and eval time which is approximate bayesian inference. In this way,
we can obtain the epistemic uncertainty (also called model uncertainty).
If the original model has Dropout layer, just use dropout when eval time, if not, add dropout layer
after Dense layer or Conv layer, then use dropout during train and eval time.
See more details in `Dropout as a Bayesian Approximation: Representing Model uncertainty in Deep Learning
<https://arxiv.org/abs/1506.02142>`_.
"""
def __init__(self, epi_model):
super(EpistemicUncertaintyModel, self).__init__()
self.drop_count = 0
if not self._make_epistemic(epi_model):
raise ValueError("The model has not Dense Layer or Convolution Layer, "
"it can not evaluate epistemic uncertainty so far.")
self.epi_model = self._make_epistemic(epi_model)
def construct(self, x):
x = self.epi_model(x)
return x
def _make_epistemic(self, epi_model, keep_prob=0.5):
"""
The dropout rate is set to 0.5 by default.
"""
for (name, layer) in epi_model.name_cells().items():
if isinstance(layer, (Conv2d, Dense, Dropout)):
if isinstance(layer, Dropout):
self.drop_count += 1
return epi_model
uncertainty_layer = layer
uncertainty_name = name
drop = Dropout(keep_prob=keep_prob)
bnn_drop = SequentialCell([uncertainty_layer, drop])
setattr(epi_model, uncertainty_name, bnn_drop)
return epi_model
if self._make_epistemic(layer):
return epi_model
return None
class AleatoricUncertaintyModel(Cell):
"""
The aleatoric uncertainty (also called data uncertainty) is caused by input data, to obtain this
uncertainty, the loss function must be modified in order to add variance into loss.
See more details in `What Uncertainties Do We Need in Bayesian Deep Learning for Computer Vision?
<https://arxiv.org/abs/1703.04977>`_.
"""
def __init__(self, ale_model, num_classes, task):
super(AleatoricUncertaintyModel, self).__init__()
self.task = task
if task == 'classification':
self.ale_model = ale_model
self.var_layer = Dense(num_classes, num_classes)
else:
self.ale_model, self.var_layer, self.pred_layer = self._make_aleatoric(
ale_model)
def construct(self, x):
if self.task == 'classification':
pred = self.ale_model(x)
var = self.var_layer(pred)
else:
x = self.ale_model(x)
pred = self.pred_layer(x)
var = self.var_layer(x)
return pred, var
def _make_aleatoric(self, ale_model):
"""
In order to add variance into original loss, add var Layer after the original network.
"""
dense_layer = dense_name = None
for (name, layer) in ale_model.name_cells().items():
if isinstance(layer, Dense):
dense_layer = layer
dense_name = name
if dense_layer is None:
raise ValueError("The model has not Dense Layer, "
"it can not evaluate aleatoric uncertainty so far.")
setattr(ale_model, dense_name, Flatten())
var_layer = Dense(dense_layer.in_channels, dense_layer.out_channels)
return ale_model, var_layer, dense_layer
class AleatoricLoss(Cell):
"""
The loss function of aleatoric model, different modification methods are adopted for
classification and regression.
"""
def __init__(self, task):
super(AleatoricLoss, self).__init__()
self.task = task
if self.task == 'classification':
self.sum = P.ReduceSum()
self.exp = P.Exp()
self.normal = C.normal
self.to_tensor = P.ScalarToTensor()
self.entropy = SoftmaxCrossEntropyWithLogits(
sparse=True, reduction="mean")
else:
self.mean = P.ReduceMean()
self.exp = P.Exp()
self.pow = P.Pow()
def construct(self, data_pred, y):
y_pred, var = data_pred
if self.task == 'classification':
sample_times = 10
epsilon = self.normal((1, sample_times), self.to_tensor(0.0, mstype.float32),
self.to_tensor(1.0, mstype.float32), 0)
total_loss = 0
for i in range(sample_times):
y_pred_i = y_pred + epsilon[0][i] * var
loss = self.entropy(y_pred_i, y)
total_loss += loss
avg_loss = total_loss / sample_times
return avg_loss
loss = self.mean(0.5 * self.exp(-var) *
self.pow(y - y_pred, 2) + 0.5 * var)
return loss

135
tests/st/probability/toolbox/test_uncertainty.py
View File

@ -1,135 +0,0 @@
# Copyright 2020-2022 Huawei Technologies Co., Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============================================================================
""" test uncertainty toolbox """
import mindspore.dataset as ds
import mindspore.dataset.transforms as C
import mindspore.dataset.vision as CV
import mindspore.nn as nn
from mindspore import context, Tensor
from mindspore import dtype as mstype
from mindspore.common.initializer import TruncatedNormal
from mindspore.dataset.vision import Inter
from mindspore.nn.probability.toolbox.uncertainty_evaluation import UncertaintyEvaluation
from mindspore.train import load_checkpoint, load_param_into_net
context.set_context(mode=context.GRAPH_MODE, device_target="GPU")
def conv(in_channels, out_channels, kernel_size, stride=1, padding=0):
"""weight initial for conv layer"""
weight = weight_variable()
return nn.Conv2d(in_channels, out_channels,
kernel_size=kernel_size, stride=stride, padding=padding,
weight_init=weight, has_bias=False, pad_mode="valid")
def fc_with_initialize(input_channels, out_channels):
"""weight initial for fc layer"""
weight = weight_variable()
bias = weight_variable()
return nn.Dense(input_channels, out_channels, weight, bias)
def weight_variable():
"""weight initial"""
return TruncatedNormal(0.02)
class LeNet5(nn.Cell):
def __init__(self, num_class=10, channel=1):
super(LeNet5, self).__init__()
self.num_class = num_class
self.conv1 = conv(channel, 6, 5)
self.conv2 = conv(6, 16, 5)
self.fc1 = fc_with_initialize(16 * 5 * 5, 120)
self.fc2 = fc_with_initialize(120, 84)
self.fc3 = fc_with_initialize(84, self.num_class)
self.relu = nn.ReLU()
self.max_pool2d = nn.MaxPool2d(kernel_size=2, stride=2)
self.flatten = nn.Flatten()
def construct(self, x):
x = self.conv1(x)
x = self.relu(x)
x = self.max_pool2d(x)
x = self.conv2(x)
x = self.relu(x)
x = self.max_pool2d(x)
x = self.flatten(x)
x = self.fc1(x)
x = self.relu(x)
x = self.fc2(x)
x = self.relu(x)
x = self.fc3(x)
return x
def create_dataset(data_path, batch_size=32, repeat_size=1,
num_parallel_workers=1):
"""
create dataset for train or test
"""
# define dataset
mnist_ds = ds.MnistDataset(data_path)
resize_height, resize_width = 32, 32
rescale = 1.0 / 255.0
shift = 0.0
rescale_nml = 1 / 0.3081
shift_nml = -1 * 0.1307 / 0.3081
# define map operations
resize_op = CV.Resize((resize_height, resize_width), interpolation=Inter.LINEAR) # Bilinear mode
rescale_nml_op = CV.Rescale(rescale_nml, shift_nml)
rescale_op = CV.Rescale(rescale, shift)
hwc2chw_op = CV.HWC2CHW()
type_cast_op = C.TypeCast(mstype.int32)
# apply map operations on images
mnist_ds = mnist_ds.map(operations=type_cast_op, input_columns="label", num_parallel_workers=num_parallel_workers)
mnist_ds = mnist_ds.map(operations=resize_op, input_columns="image", num_parallel_workers=num_parallel_workers)
mnist_ds = mnist_ds.map(operations=rescale_op, input_columns="image", num_parallel_workers=num_parallel_workers)
mnist_ds = mnist_ds.map(operations=rescale_nml_op, input_columns="image", num_parallel_workers=num_parallel_workers)
mnist_ds = mnist_ds.map(operations=hwc2chw_op, input_columns="image", num_parallel_workers=num_parallel_workers)
# apply DatasetOps
buffer_size = 10000
mnist_ds = mnist_ds.shuffle(buffer_size=buffer_size) # 10000 as in LeNet train script
mnist_ds = mnist_ds.batch(batch_size, drop_remainder=True)
mnist_ds = mnist_ds.repeat(repeat_size)
return mnist_ds
if __name__ == '__main__':
# get trained model
network = LeNet5()
param_dict = load_checkpoint('checkpoint_lenet.ckpt')
load_param_into_net(network, param_dict)
# get train and eval dataset
ds_train = create_dataset('workspace/mnist/train')
ds_eval = create_dataset('workspace/mnist/test')
evaluation = UncertaintyEvaluation(model=network,
train_dataset=ds_train,
task_type='classification',
num_classes=10,
epochs=1,
epi_uncer_model_path=None,
ale_uncer_model_path=None,
save_model=False)
for eval_data in ds_eval.create_dict_iterator(output_numpy=True, num_epochs=1):
eval_data = Tensor(eval_data['image'], mstype.float32)
epistemic_uncertainty = evaluation.eval_epistemic_uncertainty(eval_data)
aleatoric_uncertainty = evaluation.eval_aleatoric_uncertainty(eval_data)