# encoding :utf-8 import tensorflow as tf from tensorflow import keras # 導入常見網絡層, sequential容器, 優化器, 損失函數 from tensorflow.keras import layers, Sequential, optimizers, losses, metrics import os os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2' import datetime import io import matplotlib.pyplot as plt # 1 """ x = tf.constant([2., 1., 0.1]) layer = layers.Softmax(axis=-1) # 創建softmax層 out = layer(x) # out = tf.nn.softmax(x) print(out) """ # 2 method one """ network = Sequential([ layers.Dense(3, activation=None), layers.ReLU(), layers.Dense(2, activation=None), layers.ReLU() ]) x = tf.random.normal([4, 3]) out = network(x) print(out) """ # 2 method two """ layers_num = 2 # 堆疊兩次 network = Sequential([]) # 先創建空的網絡容器 for _ in range(layers_num): network.add(layers.Dense(3)) # 添加全連接層 network.add(layers.ReLU()) network.build(input_shape=(4, 4)) # 創建網絡參數 network.summary() for p in network.trainable_variables: print(p.name, p.shape) """ # 3 模型裝配 # input data path = r'G:\2019\python\mnist.npz' (x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data(path) x = tf.convert_to_tensor(x_train, dtype = tf.float32)/255. #0:1 ; -1:1(不適合訓練，准確度不高) x = tf.reshape(x, [-1, 28*28]) y = tf.convert_to_tensor(y_train, dtype=tf.int32) y = tf.one_hot(y, depth=10) train_db = tf.data.Dataset.from_tensor_slices((x, y)) train_db = train_db.shuffle(60000) # 盡量與樣本空間一樣大 train_db = train_db.batch(100) # 128 def preprocess(x, y): x = tf.cast(x, dtype=tf.float32) / 255. #先將類型轉化為float32，再歸一到0-1 x = tf.reshape(x, [-1, 28*28]) #不知道x數量，用-1代替，轉化為一維784個數據 y = tf.cast(y, dtype=tf.int32) #轉化為整型32 y = tf.one_hot(y, depth=10) #訓練數據所需的one-hot編碼 return x, y # 將10000組測試數據預處理 test_db = tf.data.Dataset.from_tensor_slices((x_test, y_test)) test_db = test_db.shuffle(10000) test_db = test_db.batch(100) #128 val_db = test_db.map(preprocess) """ # 用於4中調試用 network = Sequential([ layers.Dense(256, activation='relu'), layers.Dense(128, activation='relu'), layers.Dense(64, activation='relu'), layers.Dense(32, activation='relu'), layers.Dense(10) ]) network.build(input_shape=(4, 28*28)) network.summary() network.compile(optimizer=optimizers.Adam(lr=0.01), # Adam優化器 loss=losses.CategoricalCrossentropy(from_logits=True), # 交叉熵損失函數 metrics=['accuracy']) # 設定指標為准確率 # 3 模型訓練 # 訓練5個epochs，每2個epochs驗證一次 fit()代表網絡的訓練過程 history = network.fit(train_db, epochs=5, validation_data=val_db, validation_freq=2) # history.history # 3 模型測試 x, y = next(iter(val_db)) # 加載一個測試數據 print('predict x:', x.shape) # 打印當前batch的形狀 out = network.predict(x) # 模型預測保存在out中 print(out) # network.evaluate(val_db) # 模型測試，性能表現 # 4 模型的保存 # method 1 # network.save_weights('weight.ckpt') # print('saved weights.') # del network # method 2 # network.save('exam6_model.h5') # print('saved total model.') # del network # method 3 tf.saved_model.save(network, 'exam6_model-savedmodel') print('saving savedmodel.') del network """ # 創建相同的網絡 有網絡源的情況下 method 1 """ network = Sequential([ layers.Dense(256, activation='relu'), layers.Dense(128, activation='relu'), layers.Dense(64, activation='relu'), layers.Dense(32, activation='relu'), layers.Dense(10) ]) network.build(input_shape=(4, 28*28)) network.summary() network.compile(optimizer=optimizers.Adam(lr=0.01), # Adam優化器 loss=losses.CategoricalCrossentropy(from_logits=True), # 交叉熵損失函數 metrics=['accuracy']) # 設定指標為准確率 # 從文件中讀入參數數據到當前網絡 network.load_weights('weights.ckpt') print('loaded weights!') # Failed to find any matching files for weights.ckpt(import os !) """ # 無網絡源的情況 method 2 # network = keras.models.load_model('exam6_model.h5') # network.summary() # SaveModel方式 method 3 """ print('load savedmodel from file.') network = tf.saved_model.load('exam6_model-savedmodel') acc_meter = metrics.CategoricalAccuracy() for x, y in val_db: pred = network(x) acc_meter.update_state(y_true=y, y_pred=pred) print("Test Accuracy:%f" % acc_meter.result()) # Test Accuracy:0.967000 """ # 5 自定義網絡層 """ class MyDense(layers.Layer): def __init__(self, inp_dim, outp_dim): super(MyDense, self).__init__() # 創建權值張量並添加到管理列表中 self.kernel = self.add_variable('W', [inp_dim, outp_dim], trainable=True) # net = MyDense(4, 3) # print(net.variables, net.trainable_variables) def call(self, inputs, training=None): out = inputs@self.kernel out = tf.nn.relu(out) return out network = Sequential([ MyDense(784, 256), MyDense(256, 128), MyDense(128, 64), MyDense(64, 32), MyDense(32, 10) ]) network.build(input_shape=(None, 28*28)) network.summary() class MyModel(keras.Model): def __init__(self): super(MyModel, self).__init__() self.fc1 = MyDense(28*28, 256) self.fc2 = MyDense(256, 128) self.fc3 = MyDense(128, 64) self.fc4 = MyDense(64, 32) self.fc5 = MyDense(32, 10) def call(self, inputs, training=None): x = self.fc1(inputs) x = self.fc2(x) x = self.fc3(x) x = self.fc4(x) x = self.fc5(x) return x """ # 6 模型樂園 """ # 加載 ImageNet預訓練模型，去掉最后一層 resnet = keras.applications.ResNet50(weights='imagenet', include_top=False) # resnet.summary() # x = tf.random.normal([4, 224, 224, 3]) # out = resnet(x) # print(out) # shape=(4, 7, 7, 2048) # 新建池化層 global_average_layer = layers.GlobalAveragePooling2D() # x = tf.random.normal([4, 7, 7, 2048]) # out = global_average_layer(x) # print(out.shape) # (4, 2048) # 新建全連接層 fc = layers.Dense(100) # x = tf.random.normal([4, 2048]) #out = fc(x) #print(out.shape) # (4, 100) # 重新包裹網絡模型 mynet = Sequential([resnet, global_average_layer, fc]) mynet.summary() """ # 7 准確率 network = Sequential([ layers.Dense(256, activation='relu'), layers.Dense(128, activation='relu'), layers.Dense(64, activation='relu'), layers.Dense(32, activation='relu'), layers.Dense(10) ]) network.build(input_shape=(None, 28*28)) network.summary() optimizer=optimizers.Adam(lr=0.01) acc_meter = metrics.Accuracy() # 創建准確率測量器 loss_meter = metrics.Mean() # 新建平均測量器 for step, (x, y) in enumerate(train_db): #遍歷切分好的數據step:0->599 with tf.GradientTape() as tape: out = network(x) loss = tf.reduce_mean(tf.losses.categorical_crossentropy(y, out, from_logits=True)) loss_meter.update_state(float(loss)) # 寫入數據 grads = tape.gradient(loss, network.trainable_variables) optimizer.apply_gradients(zip(grads, network.trainable_variables)) if step % 100 == 0: print(step, 'loss:', loss_meter.result().numpy()) # 讀統計數據 loss_meter.reset_states() # 清零 # 測試 if step % 500 == 0: total, total_correct = 0., 0 acc_meter.reset_states() for step, (x, y) in enumerate(val_db): out = network(x) correct = tf.equal(out, y) total_correct += tf.reduce_sum(tf.cast(correct, dtype=tf.int32)).numpy() total += x.shape[0] acc_meter.update_state(y, out) print(step, 'Evaluate Acc:', total_correct/total, acc_meter.result().numpy())
認識到keras用於神經網絡學習的簡便性！