import tensorflow as tf from tensorflow.keras.models import Model from tensorflow.keras.layers import Dense, Input, Dropout, Conv2D, Flatten, Reshape, MaxPooling2D, BatchNormalization, Activation, Concatenate from tensorflow.keras.optimizers import Adam from tensorflow.keras.callbacks import ReduceLROnPlateau, EarlyStopping from sklearn.metrics import classification_report, confusion_matrix import numpy as np import sys # --- 1. DEFINIÇÃO DA GEOMETRIA (AUTO-AJUSTÁVEL) --- # O erro diz: input 1658, target image 41x40 (1640) # Isso significa que temos 18 features de sobra (Wavelet) MFE_COLUMNS = 40 # Certifique-se que seu bloco MFE tem 40 filtros MFE_ROWS = 40 # 1640 / 40 = 41 janelas de tempo MFE_SIZE = MFE_ROWS * MFE_COLUMNS # 1640 WAVELET_SIZE = input_length - MFE_SIZE # O resto (18) print(f"Geometria: Input Total={input_length} | Imagem MFE={MFE_SIZE} ({MFE_ROWS}x{MFE_COLUMNS}) | Wavelet={WAVELET_SIZE}") # Se por acaso você mudou o MFE e a conta não bater, o script avisa if WAVELET_SIZE < 0: print("ERRO CRÍTICO: O tamanho da imagem MFE é maior que o input total. Verifique MFE_COLUMNS.") sys.exit(1) # --- 2. CONFIGURAÇÕES --- EPOCHS = args.epochs or 60 LEARNING_RATE = 0.005 BATCH_SIZE = args.batch_size or 32 ENSURE_DETERMINISM = args.ensure_determinism if not ENSURE_DETERMINISM: train_dataset = train_dataset.shuffle(buffer_size=BATCH_SIZE*4) train_dataset = train_dataset.batch(BATCH_SIZE, drop_remainder=False) validation_dataset = validation_dataset.batch(BATCH_SIZE, drop_remainder=False) # --- 3. ARQUITETURA SENSOR FUSION (MFE + WAVELET) --- # Entrada Única (Pacote misturado) input_layer = Input(shape=(input_length,), name='raw_input') # --- RAMO 1: VISÃO (SPECTROGRAMA/MFE) --- # Pega os primeiros 1640 valores x_mfe = tf.keras.layers.Lambda(lambda x: x[:, :MFE_SIZE], name='slice_mfe')(input_layer) x_mfe = Reshape((MFE_ROWS, MFE_COLUMNS, 1), name='reshape_mfe')(x_mfe) # CNN 2D x_mfe = Conv2D(32, (3, 3), padding='same', kernel_initializer='he_normal')(x_mfe) x_mfe = BatchNormalization()(x_mfe) x_mfe = Activation('relu')(x_mfe) x_mfe = MaxPooling2D((2, 2))(x_mfe) x_mfe = Dropout(0.25)(x_mfe) x_mfe = Conv2D(64, (3, 3), padding='same', kernel_initializer='he_normal')(x_mfe) x_mfe = BatchNormalization()(x_mfe) x_mfe = Activation('relu')(x_mfe) x_mfe = MaxPooling2D((2, 2))(x_mfe) x_mfe = Dropout(0.25)(x_mfe) x_mfe = Flatten()(x_mfe) # --- RAMO 2: ESTATÍSTICA (WAVELET) --- # Pega o que sobrar (os últimos 18 valores) x_wav = tf.keras.layers.Lambda(lambda x: x[:, MFE_SIZE:], name='slice_wavelet')(input_layer) # Rede Densa para os dados numéricos x_wav = Dense(16, kernel_initializer='he_normal')(x_wav) x_wav = BatchNormalization()(x_wav) x_wav = Activation('relu')(x_wav) # --- 4. FUSÃO --- combined = Concatenate()([x_mfe, x_wav]) # Decisão Final z = Dense(64, kernel_initializer='he_normal')(combined) z = BatchNormalization()(z) z = Activation('relu')(z) z = Dropout(0.5)(z) output_layer = Dense(classes, activation='softmax', name='y_pred')(z) model = Model(inputs=input_layer, outputs=output_layer) # --- 5. TREINO E CALLBACKS --- opt = Adam(learning_rate=LEARNING_RATE, beta_1=0.9, beta_2=0.999) callbacks.append(BatchLoggerCallback(BATCH_SIZE, train_sample_count, epochs=EPOCHS, ensure_determinism=ENSURE_DETERMINISM)) callbacks.append(ReduceLROnPlateau(monitor='val_accuracy', factor=0.5, patience=5, min_lr=0.00001, verbose=1)) callbacks.append(EarlyStopping(monitor='val_loss', patience=15, restore_best_weights=True, verbose=1)) model.compile(loss='categorical_crossentropy', optimizer=opt, metrics=['accuracy']) model.fit(train_dataset, epochs=EPOCHS, validation_data=validation_dataset, verbose=2, callbacks=callbacks, class_weight=ei_tensorflow.training.get_class_weights(Y_train)) # --- RELATÓRIO FINAL --- print("--- RELATÓRIO DE FUSÃO ---") X_val = [] Y_val = [] for images, labels in validation_dataset: X_val.append(images.numpy()) Y_val.append(labels.numpy()) if len(X_val) > 0: X_val = np.concatenate(X_val) Y_val = np.concatenate(Y_val) y_true = np.argmax(Y_val, axis=1) y_pred = np.argmax(model.predict(X_val, verbose=0), axis=1) print(classification_report(y_true, y_pred)) print(confusion_matrix(y_true, y_pred))