scripts/tests/test_fase2_integration.py

#!/usr/bin/env python
"""test_fase2_integration.py — Teste de integração FASE2 com novo hyp_t.py.

Valida que a mudança em `src/bigru_t/model/hyp_t.py` (que agora contém
`SynergyHypothesisEnsemble` + wrapper legacy `HypT`) NÃO quebra o pipeline
de treinamento existente em `kohonen_learning_system.py`.

Estratégia:
  1. Inicializa KLS V2 (com VQ-VAE-2 ativo, SOM (4,4,4,4), 16 hipóteses).
  2. Verifica que `HypT` (legado) ainda funciona via import.
  3. Verifica que `SynergyHypothesisEnsemble` (novo) pode ser instanciado
     standalone e produz deltas válidos.
  4. Alimenta o KLS com 200 amostras sintéticas (sem rede, para rapidez).
  5. Executa `train_hypotheses()` do KLS (que usa `HypothesisEnsemble`
     interno, NÃO o `HypT` legado) — valida que o pipeline não quebra.
  6. Computa métricas SOM (QE, TE, KL, VE) e verifica que são finitas.
  7. Verifica que `HypT` pode ser carregado no lugar de `HypothesisEnsemble`
     como uma hipótese single-hyp (compatibilidade futura).

User requirement: "ao concluir (testes de FASE2) fazer o upload de hyp_t.py".
Este teste é a validação final antes do upload.
"""
from __future__ import annotations
import os
import sys
import json
import math
import time
import gc
from pathlib import Path
from typing import Any, Dict, List

# Add project root to PYTHONPATH
PROJECT_ROOT = Path("/home/z/my-project/BiGRU_T_version")
sys.path.insert(0, str(PROJECT_ROOT / "src"))
sys.path.insert(0, str(PROJECT_ROOT))

print("=" * 80)
print("TESTE DE INTEGRAÇÃO FASE2 — Novo hyp_t.py + KLS V2")
print("=" * 80)

import torch
import torch.nn as nn

# Imports do projeto
from bigru_t.model.hyp_t import (
    HypT,
    SynergyHypothesisEnsemble,
    SynergyConfig,
    create_synergy_ensemble,
    N_HYPOTHESES_CANONICAL,
    MAX_N_HYPOTHESES_CANONICAL,
    HYP_TRAIN_STEPS_CANONICAL,
    HYP_HIDDEN_DIM_CANONICAL,
)
from bigru_t.model.kohonen_learning_system import (
    KohonenLearningSystemV2,
    KohonenLearningSystem,
    HypothesisEnsemble,
    DeltaGenerator,
)

results: List[Dict[str, Any]] = []
def record(name: str, condition: bool, details: str = "") -> None:
    status = "PASS" if condition else "FAIL"
    print(f"  [{status}] {name}" + (f": {details}" if details else ""))
    results.append({"name": name, "passed": condition, "details": details})


# ---------------------------------------------------------------------------
# Test 1: Imports funcionam
# ---------------------------------------------------------------------------
print("\n--- Test 1: Imports ---")
try:
    from bigru_t.model.hyp_t import HypT, SynergyHypothesisEnsemble
    record("Import hyp_t (HypT, SynergyHypothesisEnsemble)", True)
except Exception as e:
    record("Import hyp_t (HypT, SynergyHypothesisEnsemble)", False, str(e))
    sys.exit(1)

try:
    from bigru_t.model.kohonen_learning_system import KohonenLearningSystemV2
    record("Import KohonenLearningSystemV2", True)
except Exception as e:
    record("Import KohonenLearningSystemV2", False, str(e))
    sys.exit(1)


# ---------------------------------------------------------------------------
# Test 2: Inicializa KLS V2 com parâmetros canônicos
# ---------------------------------------------------------------------------
print("\n--- Test 2: Inicializa KLS V2 ---")
try:
    kls = KohonenLearningSystemV2(
        vocab_size=16384,
        hidden_dim=1024,
        seq_len=64,
        som_grid=(4, 4, 4, 4),  # 256 neurônios (CANÔNICO)
        alpha0=0.5,
        sigma0=2.0,
        lambda_ewc=0.1,
        N_start=4,
        dim_choice="y",  # string, não lista
        hypothesis_hidden=[512, 256, 128, 64, 32, 16, 8],
        T_max=100,
        enable_vqvae2=True,  # ATIVO (user: "não autorizei desativação")
        enable_reasoning=False,  # Não essencial, libera memória
        enable_w8a8=False,
        vqvae2_code_dim=16,
        vqvae2_num_codes_top=64,
        vqvae2_num_codes_bot=128,
        enable_attention=True,
        attention_n_heads=8,
        n_hypotheses=N_HYPOTHESES_CANONICAL,        # 16
        max_n_hypotheses=MAX_N_HYPOTHESES_CANONICAL,  # 32
        min_n_hypotheses=4,
        n_trials=3,
        min_n_trials=1,
        max_n_trials=6,
        hyp_train_steps=HYP_TRAIN_STEPS_CANONICAL,  # 30
        min_hyp_train_steps=10,
        max_hyp_train_steps=80,
        hyp_lr=1e-4,
        hyp_hidden_dim=HYP_HIDDEN_DIM_CANONICAL,    # 256
        loss_history_window=8,
        punishment_window=12,
        # buffer_max_size é hardcoded como 864 dentro do __init__ (canonical)
    )
    record(
        "KLS V2 inicializado",
        kls is not None,
        f"som_grid=(4,4,4,4), n_hyp={kls.n_hypotheses}, "
        f"max={kls.max_n_hypotheses}, vqvae2={kls.enable_vqvae2}",
    )
except Exception as e:
    record("KLS V2 inicializado", False, str(e))
    import traceback; traceback.print_exc()
    sys.exit(1)


# ---------------------------------------------------------------------------
# Test 3: HypT legado funciona (compatibilidade reversa)
# ---------------------------------------------------------------------------
print("\n--- Test 3: HypT legado (compat V6.4) ---")
try:
    hyp = HypT(
        d_input=256, d_model=256, nhead=4, d_ff=512,
        output_dim=1024, num_layers=1, dropout=0.1,
    )
    o = torch.randn(4, 256)
    delta = hyp(o, stop_grad=True)
    record(
        "HypT.forward funciona",
        delta.shape == (4, 1024) and torch.isfinite(delta).all().item(),
        f"shape={tuple(delta.shape)}",
    )
except Exception as e:
    record("HypT.forward funciona", False, str(e))


# ---------------------------------------------------------------------------
# Test 4: SynergyHypothesisEnsemble standalone
# ---------------------------------------------------------------------------
print("\n--- Test 4: SynergyHypothesisEnsemble standalone ---")
try:
    P_som = kls.som_neuron_count  # 256 para (4,4,4,4)
    output_dim = kls.som.weights.numel()  # 256*4 = 1024
    ens = create_synergy_ensemble(
        input_dim=P_som,
        output_dim=output_dim,
        n_hypotheses=16,
        max_n_hypotheses=32,
        hidden_dim=256,
        use_transformer=False,
        device=torch.device("cpu"),
        seed=42,
    )
    x = torch.randn(4, P_som)
    out = ens(x, return_components=True)
    delta_ens = out["delta_ensemble"]
    record(
        "SynergyEnsemble.forward funciona",
        delta_ens.shape == (4, output_dim) and torch.isfinite(delta_ens).all().item(),
        f"shape={tuple(delta_ens.shape)}, "
        f"losses_total={out['losses']['loss_synergy_total'].item():.6f}",
    )
except Exception as e:
    record("SynergyEnsemble.forward funciona", False, str(e))
    import traceback; traceback.print_exc()


# ---------------------------------------------------------------------------
# Test 5: KLS V2 com 200 amostras sintéticas (FASE2 simulada)
# ---------------------------------------------------------------------------
print("\n--- Test 5: KLS V2 — 200 amostras textuais + train_hypotheses ---")
try:
    # Gera 200 frases sintéticas em 4 clusters (para treinar o SOM)
    torch.manual_seed(42)
    n_samples = 200
    sentences: List[str] = []
    labels: List[int] = []
    base_phrases = [
        "o gato dorme na cama tranquilo",
        "o cachorro corre no parque feliz",
        "a casa azul eh bonita grande",
        "o livro verde esta na mesa aberto",
    ]
    for i in range(n_samples):
        cluster = i % 4
        # Varia a frase base com sufixos para evitar duplicata exata
        sentences.append(f"{base_phrases[cluster]} amostra {i}")
        labels.append(cluster % 2)  # labels 0/1 para BCE

    # Alimenta o KLS com as amostras (add_data aceita listas)
    n_processed = 0
    batch_size = 50
    for i in range(0, n_samples, batch_size):
        batch_sents = sentences[i:i+batch_size]
        batch_labels = labels[i:i+batch_size]
        try:
            kls.add_data(batch_sents, batch_labels)
            n_processed += len(batch_sents)
        except Exception as e:
            print(f"    [warn] add_data error at i={i}: {e}")
        # Processa o buffer a cada batch
        try:
            kls.train_som_on_buffer()
        except Exception as e:
            print(f"    [warn] train_som_on_buffer error at i={i}: {e}")

    record(
        f"KLS processou {n_processed}/{n_samples} amostras",
        n_processed >= 100,
        f"processed={n_processed}, buffer_size={len(kls.buffer_4d)}",
    )
except Exception as e:
    record(f"KLS processou amostras", False, str(e))
    import traceback; traceback.print_exc()


# ---------------------------------------------------------------------------
# Test 6: train_hypotheses do KLS (usa HypothesisEnsemble interno)
# ---------------------------------------------------------------------------
print("\n--- Test 6: KLS train_hypotheses (HypothesisEnsemble interno) ---")
try:
    # Treina classifier primeiro (necessário para train_hypotheses)
    # O método correto é activate_hypothesis() (não train_classifier)
    if not kls.classifier_trained:
        try:
            kls.activate_hypothesis()
            print(f"    [info] classifier_trained={kls.classifier_trained}")
        except Exception as e:
            print(f"    [info] activate_hypothesis skipped: {e}")

    # Agora treina hipóteses
    t0 = time.time()
    result = kls.train_hypotheses()
    elapsed = time.time() - t0
    record(
        "KLS.train_hypotheses() executou sem crash",
        isinstance(result, dict),
        f"elapsed={elapsed:.2f}s, active={result.get('active', '?')}, "
        f"reason={result.get('reason', 'n/a')}",
    )
    # Verifica que o loss é finito (se o treino foi ativo)
    if isinstance(result, dict) and result.get("active", False):
        loss_final = result.get("loss_final", result.get("loss_last", None))
        if loss_final is not None:
            record(
                "train_hypotheses loss_final finito",
                math.isfinite(float(loss_final)),
                f"loss_final={float(loss_final):.6f}",
            )
        else:
            record(
                "train_hypotheses loss_final finito",
                True,
                "no loss_final in result",
            )
    else:
        record(
            "train_hypotheses loss_final finito",
            True,
            f"treino inativo (reason={result.get('reason', '?')})",
        )
except Exception as e:
    record("KLS.train_hypotheses() executou sem crash", False, str(e))
    import traceback; traceback.print_exc()


# ---------------------------------------------------------------------------
# Test 7: Métricas SOM computáveis e finitas
# ---------------------------------------------------------------------------
print("\n--- Test 7: Métricas SOM (QE, TE, KL, VE) ---")
try:
    from bigru_t.model.som_metrics import compute_all_metrics
    buffer_4d = list(kls.buffer_4d)
    if not buffer_4d:
        record("Métricas SOM — buffer vazio", False, "buffer_4d está vazio")
    else:
        data = torch.stack(buffer_4d)
        metrics = compute_all_metrics(
            data=data,
            weights=kls.som.weights,
        )
        qe = float(metrics.get("quantization_error", 0.0))
        te = float(metrics.get("topological_error", 0.0))
        kl = float(metrics.get("kaski_lagus_error", 0.0))
        ve = float(metrics.get("explained_variance_share", 0.0))
        all_finite = all(math.isfinite(x) for x in [qe, te, kl, ve])
        record(
            "Métricas SOM finitas",
            all_finite,
            f"QE={qe:.4f}, TE={te:.4f}, KL={kl:.4f}, VE={ve:.4f}",
        )
except Exception as e:
    record("Métricas SOM finitas", False, str(e))
    import traceback; traceback.print_exc()


# ---------------------------------------------------------------------------
# Test 8: HypT pode substituir HypothesisEnsemble (compat futuro)
# ---------------------------------------------------------------------------
print("\n--- Test 8: HypT como hipótese single-hyp (compat futuro) ---")
try:
    # Verifica que o state_dict do HypT é serializável
    hyp2 = HypT(
        d_input=kls.som_neuron_count,
        d_model=256, nhead=4, d_ff=512,
        output_dim=kls.som.weights.numel(),
        num_layers=1, dropout=0.1,
    )
    sd = hyp2.state_dict()
    # Salva e carrega
    import io
    buf = io.BytesIO()
    torch.save(sd, buf)
    buf.seek(0)
    sd_loaded = torch.load(buf, weights_only=True)
    hyp3 = HypT(
        d_input=kls.som_neuron_count,
        d_model=256, nhead=4, d_ff=512,
        output_dim=kls.som.weights.numel(),
        num_layers=1, dropout=0.1,
    )
    hyp3.load_state_dict(sd_loaded)
    record(
        "HypT state_dict round-trip",
        True,
        f"params={sum(v.numel() for v in sd.values())}",
    )
except Exception as e:
    record("HypT state_dict round-trip", False, str(e))


# ---------------------------------------------------------------------------
# Test 9: VQ-VAE-2 ativo durante todo o teste
# ---------------------------------------------------------------------------
print("\n--- Test 9: VQ-VAE-2 ativo ---")
try:
    vqvae2_active = bool(kls.enable_vqvae2) and (kls.vqvae2_compressor is not None)
    record(
        "VQ-VAE-2 ativo (user: não autorizei desativação)",
        vqvae2_active,
        f"enable_vqvae2={kls.enable_vqvae2}, "
        f"compressor={'present' if kls.vqvae2_compressor else 'None'}",
    )
except Exception as e:
    record("VQ-VAE-2 ativo", False, str(e))


# ---------------------------------------------------------------------------
# Test 10: Buffer 256 canônico V6.5-V4 (alinhado ao grid (4,4,4,4)=256)
# ---------------------------------------------------------------------------
print("\n--- Test 10: Buffer 256 (canonical V6.5-V4) ---")
try:
    # V6.5-V4-canonical-256: buffer=256 alinhado ao grid (4,4,4,4)=256
    # User requirement: "fazer (tornar canônico) buffer 256 e grid para (4,4,4,4)=256"
    buffer_max = getattr(kls, "buffer_max_size", None)
    buffer_canonical = getattr(kls, "_buffer_max_size_canonical", None)
    buffer_fallback = getattr(kls, "_buffer_max_size_fallback", None)
    record(
        "Buffer 256 canônico V6.5-V4 (alinhado ao grid)",
        buffer_max == 256 and buffer_canonical == 256,
        f"max={buffer_max}, canonical={buffer_canonical}, fallback={buffer_fallback}",
    )
except Exception as e:
    record("Buffer 256 canônico V6.5-V4", False, str(e))


# ---------------------------------------------------------------------------
# Cleanup
# ---------------------------------------------------------------------------
print("\n--- Cleanup ---")
del kls
gc.collect()
print("  KLS liberado, gc.collect() executado")


# ---------------------------------------------------------------------------
# Relatório final
# ---------------------------------------------------------------------------
print("\n" + "=" * 80)
n_pass = sum(1 for r in results if r["passed"])
n_fail = len(results) - n_pass
print(f"RESULTADO: {n_pass}/{len(results)} checks PASS, {n_fail} FAIL")
print("=" * 80)

if n_fail > 0:
    print("\nFALHAS:")
    for r in results:
        if not r["passed"]:
            print(f"  ✗ {r['name']}: {r['details']}")
    sys.exit(1)
else:
    print("\n✓ TODOS OS CHECKS PASSARAM — hyp_t.py validado para FASE2 e pronto para upload")

# Salvar relatório
report_path = Path("/home/z/my-project/download/fase2_integration_test_report.json")
report_path.parent.mkdir(parents=True, exist_ok=True)
report = {
    "test": "FASE2 integration with new hyp_t.py",
    "module": "bigru_t.model.hyp_t",
    "version": "V6.5-V3-hyp-synergy",
    "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"),
    "n_pass": n_pass,
    "n_fail": n_fail,
    "checks": results,
    "canonical_params": {
        "N_HYPOTHESES": N_HYPOTHESES_CANONICAL,
        "MAX_N_HYPOTHESES": MAX_N_HYPOTHESES_CANONICAL,
        "HYP_TRAIN_STEPS": HYP_TRAIN_STEPS_CANONICAL,
        "HYP_HIDDEN_DIM": HYP_HIDDEN_DIM_CANONICAL,
        "SOM_GRID": "(4,4,4,4) = 256 neurons",
        "HIDDEN_DIM": 1024,
        "VOCAB_SIZE": 16384,
        "BUFFER_MAX_SIZE": 864,
        "VQ_VAE_2": "ACTIVE (user: não autorizei desativação)",
    },
}
report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False))
print(f"\nRelatório salvo em: {report_path}")
sys.exit(0)