使用ImageDataGenerator与VGG16输入视频两帧时的通道错误及解决问询

问题背景 #

我正在构建VGG图像流水线，尝试输入视频中的连续两帧，将两帧堆叠成(224,224,6)的形状作为模型输入，但使用ImageDataGenerator时触发了通道数错误。

原始代码 #

datagen = ImageDataGenerator()
datagen.fit(X_train)

model = Sequential()
model.add(Conv2D(input_shape=(224, 224, 6), filters=64, kernel_size=(3, 3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros'))
model.add(Conv2D(filters=64, kernel_size=(3, 3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros'))
model.add(MaxPool2D(pool_size=(2, 2), strides=(2, 2)))

model.add(Conv2D(filters=128, kernel_size=(3, 3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros'))
model.add(Conv2D(filters=128, kernel_size=(3, 3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros'))
model.add(MaxPool2D(pool_size=(2, 2), strides=(2, 2)))

model.add(Conv2D(filters=256, kernel_size=(3, 3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros'))
model.add(Conv2D(filters=256, kernel_size=(3, 3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros'))
model.add(Conv2D(filters=256, kernel_size=(3, 3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros'))
model.add(MaxPool2D(pool_size=(2, 2), strides=(2, 2)))

model.add(Conv2D(filters=512, kernel_size=(3, 3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros'))
model.add(Conv2D(filters=512, kernel_size=(3, 3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros'))
model.add(Conv2D(filters=512, kernel_size=(3, 3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros'))
model.add(MaxPool2D(pool_size=(2, 2), strides=(2, 2)))

model.add(GlobalAveragePooling2D())
model.add(Dense(units=4096, activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros'))
model.add(Dropout(0.2))
model.add(Dense(units=4096, activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros'))
model.add(Dropout(0.2))
model.add(Dense(units=1, activation='sigmoid'))

# opt = Adam(learning_rate=0.001)
opt = SGD(lr=0.01, momentum=0.3)
checkpoint = ModelCheckpoint(config.CLASH_PATH() + '/models/step_01.h5', monitor='binary_accuracy', verbose=1, save_best_only=True,
                             save_weights_only=False, mode='auto', period=1)
early = EarlyStopping(monitor='binary_accuracy', min_delta=0, patience=40, verbose=1, mode='auto')

model.compile(loss='binary_crossentropy', optimizer=opt, metrics=['binary_accuracy'])
model.summary()

model.fit(datagen.flow(X_train, y_train, batch_size=32,
     subset='training', ignore_class_split=True), validation_data=datagen.flow(X_train, y_train, batch_size=16,
     subset='validation', ignore_class_split=True), steps_per_epoch=len(X_train) / 48,
     epochs=1000, verbose=1,
     callbacks=[checkpoint, early])

触发错误 #

NumpyArrayIterator is set to use the data format convention "channels_last" (channels on axis 3), i.e. expected either 1, 3, or 4 channels on axis 3. However, it was passed an array with shape (6666, 224, 224, 6) (6 channels).

解决方案 #

核心思路是把连续两帧作为两个独立的输入源，在模型内部做拼接，同时自定义生成器保证每对帧的同步性：

1. 重构模型结构（改用函数式API） #

放弃Sequential，改用函数式API分别处理两帧输入，再拼接：

from tensorflow.keras import Input, Model
from tensorflow.keras.layers import Conv2D, MaxPool2D, GlobalAveragePooling2D, Dense, Dropout, concatenate

# 定义单帧输入的特征提取分支（复用VGG结构）
def vgg_branch(input_shape):
    inputs = Input(shape=input_shape)
    x = Conv2D(filters=64, kernel_size=(3,3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros')(inputs)
    x = Conv2D(filters=64, kernel_size=(3,3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros')(x)
    x = MaxPool2D(pool_size=(2,2), strides=(2,2))(x)
    
    x = Conv2D(filters=128, kernel_size=(3,3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros')(x)
    x = Conv2D(filters=128, kernel_size=(3,3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros')(x)
    x = MaxPool2D(pool_size=(2,2), strides=(2,2))(x)
    
    x = Conv2D(filters=256, kernel_size=(3,3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros')(x)
    x = Conv2D(filters=256, kernel_size=(3,3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros')(x)
    x = Conv2D(filters=256, kernel_size=(3,3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros')(x)
    x = MaxPool2D(pool_size=(2,2), strides=(2,2))(x)
    
    x = Conv2D(filters=512, kernel_size=(3,3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros')(x)
    x = Conv2D(filters=512, kernel_size=(3,3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros')(x)
    x = Conv2D(filters=512, kernel_size=(3,3), padding='same', activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros')(x)
    x = MaxPool2D(pool_size=(2,2), strides=(2,2))(x)
    
    x = GlobalAveragePooling2D()(x)
    return inputs, x

# 创建两个单帧输入
input_frame1, feat1 = vgg_branch((224,224,3))
input_frame2, feat2 = vgg_branch((224,224,3))

# 拼接两帧的特征
concat_feat = concatenate([feat1, feat2])

# 后续全连接层
x = Dense(units=4096, activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros')(concat_feat)
x = Dropout(0.2)(x)
x = Dense(units=4096, activation='relu', kernel_initializer='he_uniform', bias_initializer='zeros')(x)
x = Dropout(0.2)(x)
output = Dense(units=1, activation='sigmoid')(x)

# 构建完整模型
model = Model(inputs=[input_frame1, input_frame2], outputs=output)

2. 自定义同步帧生成器 #

确保生成器每次返回连续的帧对，同时支持数据增强（复用ImageDataGenerator的变换逻辑）：

import numpy as np
from tensorflow.keras.preprocessing.image import ImageDataGenerator

def frame_pair_generator(X_frames, y, batch_size=32, datagen=None):
    # X_frames的形状应为(n_samples, 2, 224, 224, 3)，第二个维度存储连续两帧
    n_samples = X_frames.shape[0]
    indices = np.arange(n_samples)
    
    while True:
        np.random.shuffle(indices)
        for start in range(0, n_samples, batch_size):
            end = min(start + batch_size, n_samples)
            batch_indices = indices[start:end]
            
            # 取出当前batch的帧对
            batch_frame1 = X_frames[batch_indices, 0]
            batch_frame2 = X_frames[batch_indices, 1]
            batch_y = y[batch_indices]
            
            # 对两帧应用相同的数据增强变换
            if datagen is not None:
                for i in range(len(batch_frame1)):
                    transform_params = datagen.get_random_transform(batch_frame1[i].shape)
                    batch_frame1[i] = datagen.apply_transform(batch_frame1[i], transform_params)
                    batch_frame2[i] = datagen.apply_transform(batch_frame2[i], transform_params)
            
            yield [batch_frame1, batch_frame2], batch_y

3. 训练代码调整 #

# 初始化数据生成器（定义增强规则即可，无需适配6通道数据）
datagen = ImageDataGenerator(rotation_range=10, width_shift_range=0.1, height_shift_range=0.1, horizontal_flip=True)

# 假设X_train已整理为(n,2,224,224,3)格式，每个样本存储连续两帧
train_generator = frame_pair_generator(X_train, y_train, batch_size=32, datagen=datagen)
val_generator = frame_pair_generator(X_val, y_val, batch_size=16)  # 验证集一般不做增强

# 编译模型
opt = SGD(lr=0.01, momentum=0.3)
checkpoint = ModelCheckpoint(config.CLASH_PATH() + '/models/step_01.h5', monitor='binary_accuracy', verbose=1, save_best_only=True,
                             save_weights_only=False, mode='auto', period=1)
early = EarlyStopping(monitor='binary_accuracy', min_delta=0, patience=40, verbose=1, mode='auto')

model.compile(loss='binary_crossentropy', optimizer=opt, metrics=['binary_accuracy'])
model.summary()

# 启动训练
model.fit(train_generator, 
          validation_data=val_generator,
          steps_per_epoch=len(X_train)//32,
          epochs=1000, 
          verbose=1,
          callbacks=[checkpoint, early])

关键说明 #

• 模型层面：用两个独立的3通道输入替代6通道输入，规避ImageDataGenerator的通道数限制
• 生成器层面：保证每对连续帧使用完全相同的数据增强变换，避免帧间信息错位
• 数据预处理：需提前将数据集整理为(n_samples, 2, 224,224,3)格式，每个样本存储连续两帧

内容的提问来源于stack exchange，提问作者C. Cooney