

### 悩んでること

PythonとTensorFlowで点群データを分類するコードを下記ページを参考に作成しました。  
10個ほどの異なる点群をインプットして実行すると、すべてが同じ分類を返されてしまいます。  
コードにどんな改良を加えれば良いでしょうか？

### 参考URL

<https://tensorflow.classcat.com/2021/12/06/keras-2-examples-vision-pointnet/>

#### 環境

Windows10  
Python 3.9.15  
TensorFlow 2.11.0(GPUは使用せずCPUで実行)  
Keras 2.11.0

### コード

**下記コードの実行上の注意点:**  
learn関数の第2引数をFalseでデータを訓練し、predict関数で点群を与え分類を判定してます。

[code]

    import os
    import glob
    import trimesh
    import numpy as np
    import tensorflow as tf
    from tensorflow import keras
    from keras import layers
    import json
    import random
    
    class pointNet_tf2:
        BATCH_SIZE = 32
        NUM_CLASSES = 3
        NUM_POINTS = 2048
        NUM_X = 16
        NUM_Y = 16
        NUM_Z = 8
        #NUM_X * NUM_Y * NUM_Z = NUM_POINTSであること!!
    
        DATASET_DIR = 'core\\pointNet\\Datasets'
        # MODEL_FILE = 'ModelNet40_model.json'
        # WEIGHT_FILE = 'ModelNet40_weight.h5'
        MODEL_FILE = 'ModelNet40_model.keras'
        CLASS_FILE = 'ModelNet40_class.json'
    
        def parse_dataset(self, data_set_dir, num_points):
    
            train_points = []
            train_labels = []
            test_points = []
            test_labels = []
            class_map = {}
            folders = glob.glob(os.path.join(data_set_dir, "[!README]*"))
    
            for i, folder in enumerate(folders):
                print("processing class: {}".format(os.path.basename(folder)))
                # store folder name with ID so we can retrieve later
                class_map[i] = os.path.basename(folder)
                # gather all files
                train_files = glob.glob(os.path.join(folder, "train/*"))
                test_files = glob.glob(os.path.join(folder, "test/*"))
    
                for f in train_files:
                    train_points.append(trimesh.load(f).sample(num_points))
                    train_labels.append(i)
    
                for f in test_files:
                    test_points.append(trimesh.load(f).sample(num_points))
                    test_labels.append(i)
    
            return (
                np.array(train_points),
                np.array(test_points),
                np.array(train_labels),
                np.array(test_labels),
                class_map,
            )
    
        def augment(self, points, label):
    
            # jitter points
            points += tf.random.uniform(points.shape, -0.005, 0.005, dtype=tf.float64)
            # shuffle points
            points = tf.random.shuffle(points)
    
            return points, label
    
        def conv_bn(self, x, filters):
    
            x = layers.Conv1D(filters, kernel_size=1, padding="valid")(x)
            x = layers.BatchNormalization(momentum=0.0)(x)
            return layers.Activation("relu")(x)
    
        def dense_bn(self, x, filters):
    
            x = layers.Dense(filters)(x)
            x = layers.BatchNormalization(momentum=0.0)(x)
            return layers.Activation("relu")(x)
    
        class OrthogonalRegularizer(keras.regularizers.Regularizer):
    
            def __init__(self, num_rows, num_cols, l2reg=0.001, **kwargs):
    
                self.num_features = num_rows
                self.l2reg = l2reg
                self.num_rows = num_rows
                self.num_cols = num_cols
                self.eye = tf.eye(num_rows, num_cols)
    
            def __call__(self, x):
                x = tf.reshape(x, (-1, self.num_features, self.num_features))
                xxt = tf.tensordot(x, x, axes=(2, 2))
                xxt = tf.reshape(xxt, (-1, self.num_features, self.num_features))
                return tf.reduce_sum(self.l2reg * tf.square(xxt - self.eye))
    
            def get_config(self):
                config = {
                    'num_rows': self.num_rows,
                    'num_cols': self.num_cols
                }
    
                return dict(list(config.items()))
    
        def tnet(self, inputs, num_features):
    
            # Initalise bias as the indentity matrix
            bias = keras.initializers.Constant(np.eye(num_features).flatten())
            reg = self.OrthogonalRegularizer(num_features, num_features)
    
            x = self.conv_bn(inputs, 32)
            x = self.conv_bn(x, 64)
            x = self.conv_bn(x, 512)
            x = layers.GlobalMaxPooling1D()(x)
            x = self.dense_bn(x, 256)
            x = self.dense_bn(x, 128)
            x = layers.Dense(
                num_features * num_features,
                kernel_initializer="zeros",
                bias_initializer=bias,
                activity_regularizer=reg,
            )(x)
            feat_T = layers.Reshape((num_features, num_features))(x)
    
            # Apply affine transformation to input features
            return layers.Dot(axes=(2, 1))([inputs, feat_T])
    
        # コンストラクタ
        def __init__(self):
    
            tf.random.set_seed(1234)
    
            self.classes = None
    
            inputs = keras.Input(shape=(self.NUM_POINTS, 3))
    
            x = self.tnet(inputs, 3)
            x = self.conv_bn(x, 32)
            x = self.conv_bn(x, 32)
            x = self.tnet(x, 32)
            x = self.conv_bn(x, 32)
            x = self.conv_bn(x, 64)
            x = self.conv_bn(x, 512)
            x = layers.GlobalMaxPooling1D()(x)
            x = self.dense_bn(x, 256)
            x = layers.Dropout(0.3)(x)
            x = self.dense_bn(x, 128)
            x = layers.Dropout(0.3)(x)
    
            outputs = layers.Dense(self.NUM_CLASSES, activation="softmax")(x)
    
            self.model = keras.Model(inputs=inputs, outputs=outputs, name="pointnet")
    
            self.model.summary()
    
            self.model.compile(
                loss="sparse_categorical_crossentropy",
                optimizer=keras.optimizers.Adam(learning_rate=0.001),
                metrics=["sparse_categorical_accuracy"],
            )
    
        # 学習、訓練
        def learn(self, root_dir, save):
    
            # データセットのロード
            # data_set_dir = tf.keras.utils.get_file(
            #     "modelnet.zip",
            #     "http://3dvision.princeton.edu/projects/2014/3DShapeNets/ModelNet40.zip",
            #     extract=True,
            # )
            # data_set_dir = os.path.join(os.path.dirname(data_set_dir), "ModelNet40")
            data_set_dir = os.path.join(root_dir, self.DATASET_DIR)
    
            # mesh = trimesh.load(os.path.join(data_set_dir, "chair/train/chair_0001.off"))
            # mesh.show()
            # points = mesh.sample(NUM_POINTS)
    
            # fig = plt.figure(figsize=(5, 5))
            # ax = fig.add_subplot(111, projection="3d")
            # ax.scatter(points[:, 0], points[:, 1], points[:, 2])
            # ax.set_axis_off()
            # plt.show()
    
            # データセットのパース
            train_points, test_points, train_labels, test_labels, class_map = \
                self.parse_dataset(data_set_dir + '\\ModelNet40', self.NUM_POINTS)
    
            self.classes = class_map
    
            # データセットの増強
            train_dataset = tf.data.Dataset.from_tensor_slices((train_points, train_labels))
            test_dataset = tf.data.Dataset.from_tensor_slices((test_points, test_labels))
    
            train_dataset = train_dataset.shuffle(len(train_points)).map(self.augment).batch(self.BATCH_SIZE)
            test_dataset = test_dataset.shuffle(len(test_points)).batch(self.BATCH_SIZE)
    
            # モデルの訓練
            self.model.fit(train_dataset, epochs=20, validation_data=test_dataset)
    
            # テスト
            num_test = 10
            test_data = test_dataset.take(1)
            points, labels = list(test_data)[0]
            points = points[: num_test, ...]
            labels = labels[: num_test, ...]
    
            for t in range(num_test):
                print("pred test[" + str(t) + "]:answer=" + self.classes[labels.numpy()[t]] + ",pred=" + self.predict(points[t]))
    
            if (save):
                # 保存
                keras.models.save_model(self.model, data_set_dir + '\\' + self.MODEL_FILE)
    
                # self.model.save_weights(data_set_dir + '\\' + self.WEIGHT_FILE)
    
                with open(data_set_dir + '\\' + self.CLASS_FILE, 'wt') as f:
                    json.dump(class_map, f, ensure_ascii=False, indent=4, sort_keys=True, separators=(',', ': '))
    
        # 学習済みデータのロード
        def load(self, root_dir):
    
            data_set_dir = os.path.join(root_dir, self.DATASET_DIR)
    
            # self.model = keras.models.load_model(filepath=data_set_dir + '\\' + self.MODEL_FILE,
            #     custom_objects={"OrthogonalRegularizer": pointNet.OrthogonalRegularizer})
            self.model = keras.models.load_model(filepath=data_set_dir + '\\' + self.MODEL_FILE)
    
            # self.model.load_weights(data_set_dir + '\\' + self.WEIGHT_FILE)
    
            self.classes = {}
            with open(data_set_dir + '\\' + self.CLASS_FILE) as f:
                work = json.load(f)
                for k,v in work.items():
                    self.classes[int(k)] = v
    
        #アップサンプリング
        def upSampling(self, points):
    
            ret = []
    
            if len(points) > 0:
                while(len(ret) < self.NUM_POINTS):
                    ret.append(points[0])
    
            return ret
    
        #ダウンサンプリング
        def downSampling(self, points):
    
            ret = []
    
            # 各次元の最小値と最大値を取得する
            x_min = None
            x_max = None
            y_min = None
            y_max = None
            z_min = None
            z_max = None
    
            for p in points:
                if ((x_min is None) or (x_min > p[0])):
                    x_min = p[0]
                if ((x_max is None) or (x_max < p[0])):
                    x_max = p[0]
                if ((y_min is None) or (y_min > p[1])):
                    y_min = p[1]
                if ((y_max is None) or (y_max < p[1])):
                    y_max = p[1]
                if ((z_min is None) or (z_min > p[2])):
                    z_min = p[2]
                if ((z_max is None) or (z_max < p[2])):
                    z_max = p[2]
                
            # パラメータをワーク領域にコピーする
            x_size = (x_max - x_min) / self.NUM_X
            y_size = (y_max - y_min) / self.NUM_Y
            z_size = (z_max - z_min) / self.NUM_Z
            dim3 = [[[[] for z in range(self.NUM_Z)] for y in range(self.NUM_Y)] for x in range(self.NUM_X)]
    
            for p in points:
                x = int((p[0] - x_min) / x_size)
                if x == self.NUM_X:
                    x = self.NUM_X - 1
                y = int((p[1] - y_min) / y_size)
                if y == self.NUM_Y:
                    y = self.NUM_Y - 1
                z = int((p[2] - z_min) / z_size)
                if z == self.NUM_Z:
                    z = self.NUM_Z - 1
    
                dim3[x][y][z].append(p)
    
            # ワーク領域からランダムに点を抽出する
            while len(ret) != self.NUM_POINTS:
                for x in range(self.NUM_X):
                    for y in range(self.NUM_Y):
                        for z in range(self.NUM_Z):
                            count = len(dim3[x][y][z])
                            if (count > 0):
                                idx = random.randint(0, count - 1)
                                ret.append(dim3[x][y][z].pop(idx))
    
                                if (len(ret) == self.NUM_POINTS):
                                    break
                        if (len(ret) == self.NUM_POINTS):
                            break
                    if (len(ret) == self.NUM_POINTS):
                        break
    
            return ret
    
        # 予測、推論
        def predict(self, points_input):
    
            if (self.classes is None):
                return 'error(モデルが不十分)'
    
            sample = points_input
    
            if (len(sample) == self.NUM_POINTS):
                pass
            elif (len(sample) < self.NUM_POINTS):
                # アップサンプリング
                sample = self.upSampling(sample)
            else:
                # ダウンサンプリング
                sample = self.downSampling(sample)
    
            if (len(sample) != self.NUM_POINTS):
                return "error(点数が不正)"
    
            points_output = tf.reshape(tf.convert_to_tensor(sample, dtype=np.float64), shape=(1, self.NUM_POINTS, 3))
    
            preds = self.model.predict(points_output)
    
            indexs = tf.math.argmax(preds, axis=1).numpy()
    
            class_name = ''
            for i in indexs:
                class_name += self.classes[i] + ','
    
            if (class_name == ''):
                class_name = 'error(分類不明)'
            else:
                class_name = class_name[: len(class_name) - 1]
    
            return class_name
    
[/code]


CC BY-SA 4.0
2023-01-02T08:07:53.477
93145
2023-01-02T09:21:41.997
2023-01-02T09:21:41.997
3060
35720
question
python python3 tensorflow
tensorflow 点群 predictの予測が全て同じ分類となってしまう