1# Copyright 2020-2021 Huawei Technologies Co., Ltd 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14# ============================================================================ 15"""test SummaryCollector.""" 16import os 17import re 18import shutil 19import tempfile 20from collections import Counter 21 22import pytest 23 24from mindspore import nn, Tensor, context 25from mindspore.common.initializer import Normal 26from mindspore.nn.metrics import Loss 27from mindspore.nn.optim import Momentum 28from mindspore.ops import operations as P 29from mindspore.train import Model 30from mindspore.train.callback import SummaryCollector 31from tests.st.summary.dataset import create_mnist_dataset 32from tests.summary_utils import SummaryReader 33from tests.security_utils import security_off_wrap 34 35 36class LeNet5(nn.Cell): 37 """ 38 Lenet network 39 40 Args: 41 num_class (int): Number of classes. Default: 10. 42 num_channel (int): Number of channels. Default: 1. 43 44 Returns: 45 Tensor, output tensor 46 Examples: 47 >>> LeNet(num_class=10) 48 49 """ 50 51 def __init__(self, num_class=10, num_channel=1, include_top=True): 52 super(LeNet5, self).__init__() 53 self.conv1 = nn.Conv2d(num_channel, 6, 5, pad_mode='valid') 54 self.conv2 = nn.Conv2d(6, 16, 5, pad_mode='valid') 55 self.relu = nn.ReLU() 56 self.max_pool2d = nn.MaxPool2d(kernel_size=2, stride=2) 57 self.include_top = include_top 58 if self.include_top: 59 self.flatten = nn.Flatten() 60 self.fc1 = nn.Dense(16 * 5 * 5, 120, weight_init=Normal(0.02)) 61 self.fc2 = nn.Dense(120, 84, weight_init=Normal(0.02)) 62 self.fc3 = nn.Dense(84, num_class, weight_init=Normal(0.02)) 63 64 self.scalar_summary = P.ScalarSummary() 65 self.image_summary = P.ImageSummary() 66 self.histogram_summary = P.HistogramSummary() 67 self.tensor_summary = P.TensorSummary() 68 self.channel = Tensor(num_channel) 69 70 def construct(self, x): 71 """construct.""" 72 self.image_summary('image', x) 73 x = self.conv1(x) 74 self.histogram_summary('histogram', x) 75 x = self.relu(x) 76 self.tensor_summary('tensor', x) 77 x = self.relu(x) 78 x = self.max_pool2d(x) 79 self.scalar_summary('scalar', self.channel) 80 x = self.conv2(x) 81 x = self.relu(x) 82 x = self.max_pool2d(x) 83 if not self.include_top: 84 return x 85 x = self.flatten(x) 86 x = self.relu(self.fc1(x)) 87 x = self.relu(self.fc2(x)) 88 x = self.fc3(x) 89 return x 90 91 92class TestSummary: 93 """Test summary collector the basic function.""" 94 base_summary_dir = '' 95 96 @classmethod 97 def setup_class(cls): 98 """Run before test this class.""" 99 device_id = int(os.getenv('DEVICE_ID')) if os.getenv('DEVICE_ID') else 0 100 context.set_context(mode=context.GRAPH_MODE, device_id=device_id) 101 cls.base_summary_dir = tempfile.mkdtemp(suffix='summary') 102 103 @classmethod 104 def teardown_class(cls): 105 """Run after test this class.""" 106 if os.path.exists(cls.base_summary_dir): 107 shutil.rmtree(cls.base_summary_dir) 108 109 def _run_network(self, dataset_sink_mode=False, num_samples=2, **kwargs): 110 """run network.""" 111 lenet = LeNet5() 112 loss = nn.SoftmaxCrossEntropyWithLogits(sparse=True, reduction="mean") 113 optim = Momentum(lenet.trainable_params(), learning_rate=0.1, momentum=0.9) 114 model = Model(lenet, loss_fn=loss, optimizer=optim, metrics={'loss': Loss()}) 115 summary_dir = tempfile.mkdtemp(dir=self.base_summary_dir) 116 summary_collector = SummaryCollector(summary_dir=summary_dir, collect_freq=2, **kwargs) 117 118 ds_train = create_mnist_dataset("train", num_samples=num_samples) 119 model.train(1, ds_train, callbacks=[summary_collector], dataset_sink_mode=dataset_sink_mode) 120 121 ds_eval = create_mnist_dataset("test") 122 model.eval(ds_eval, dataset_sink_mode=dataset_sink_mode, callbacks=[summary_collector]) 123 return summary_dir 124 125 @pytest.mark.level0 126 @pytest.mark.platform_x86_ascend_training 127 @pytest.mark.platform_arm_ascend_training 128 @pytest.mark.platform_x86_gpu_training 129 @pytest.mark.env_onecard 130 @security_off_wrap 131 def test_summary_with_sink_mode_false(self): 132 """Test summary with sink mode false, and num samples is 64.""" 133 summary_dir = self._run_network(num_samples=10) 134 135 tag_list = self._list_summary_tags(summary_dir) 136 137 expected_tag_set = {'conv1.weight/auto', 'conv2.weight/auto', 'fc1.weight/auto', 'fc1.bias/auto', 138 'fc2.weight/auto', 'input_data/auto', 'loss/auto', 139 'histogram', 'image', 'scalar', 'tensor'} 140 assert set(expected_tag_set) == set(tag_list) 141 142 # num samples is 10, batch size is 2, so step is 5, collect freq is 2, 143 # SummaryCollector will collect the first step and 2th, 4th step 144 tag_count = 3 145 for value in Counter(tag_list).values(): 146 assert value == tag_count 147 148 @pytest.mark.level0 149 @pytest.mark.platform_x86_ascend_training 150 @pytest.mark.platform_arm_ascend_training 151 @pytest.mark.platform_x86_gpu_training 152 @pytest.mark.env_onecard 153 @security_off_wrap 154 def test_summary_with_sink_mode_true(self): 155 """Test summary with sink mode true, and num samples is 64.""" 156 summary_dir = self._run_network(dataset_sink_mode=True, num_samples=10) 157 158 tag_list = self._list_summary_tags(summary_dir) 159 160 # There will not record input data when dataset sink mode is True 161 expected_tags = {'conv1.weight/auto', 'conv2.weight/auto', 'fc1.weight/auto', 'fc1.bias/auto', 162 'fc2.weight/auto', 'loss/auto', 'histogram', 'image', 'scalar', 'tensor'} 163 assert set(expected_tags) == set(tag_list) 164 165 tag_count = 1 166 for value in Counter(tag_list).values(): 167 assert value == tag_count 168 169 @pytest.mark.level0 170 @pytest.mark.platform_x86_ascend_training 171 @pytest.mark.platform_arm_ascend_training 172 @pytest.mark.env_onecard 173 @security_off_wrap 174 def test_summarycollector_user_defind(self): 175 """Test SummaryCollector with user-defined.""" 176 summary_dir = self._run_network(dataset_sink_mode=True, num_samples=2, 177 custom_lineage_data={'test': 'self test'}, 178 export_options={'tensor_format': 'npy'}) 179 180 tag_list = self._list_summary_tags(summary_dir) 181 file_list = self._list_tensor_files(summary_dir) 182 # There will not record input data when dataset sink mode is True 183 expected_tags = {'conv1.weight/auto', 'conv2.weight/auto', 'fc1.weight/auto', 'fc1.bias/auto', 184 'fc2.weight/auto', 'loss/auto', 'histogram', 'image', 'scalar', 'tensor'} 185 assert set(expected_tags) == set(tag_list) 186 expected_files = {'tensor_1.npy'} 187 assert set(expected_files) == set(file_list) 188 189 @staticmethod 190 def _list_summary_tags(summary_dir): 191 """list summary tags.""" 192 summary_file_path = '' 193 for file in os.listdir(summary_dir): 194 if re.search("_MS", file): 195 summary_file_path = os.path.join(summary_dir, file) 196 break 197 assert summary_file_path 198 199 tags = list() 200 with SummaryReader(summary_file_path) as summary_reader: 201 202 while True: 203 summary_event = summary_reader.read_event() 204 if not summary_event: 205 break 206 for value in summary_event.summary.value: 207 tags.append(value.tag) 208 return tags 209 210 @staticmethod 211 def _list_tensor_files(summary_dir): 212 """list tensor tags.""" 213 export_file_path = '' 214 for file in os.listdir(summary_dir): 215 if re.search("export_", file): 216 export_file_path = os.path.join(summary_dir, file) 217 break 218 assert export_file_path 219 tensor_file_path = os.path.join(export_file_path, "tensor") 220 assert tensor_file_path 221 222 tensors = list() 223 for file in os.listdir(tensor_file_path): 224 tensors.append(file) 225 226 return tensors 227