Delete scripts/smoke_test.py

scripts/smoke_test.py

@@ -1,354 +0,0 @@
-#!/usr/bin/env python3
-"""smoke_test.py — Smoke test do pipeline BiGRU_T_version.
-
-Verifica:
-1. Importação de todos os módulos
-2. Criação do UnifiedModel
-3. Forward pass sem erro
-4. Backward pass sem erro
-5. QuantizedLinear funcionando (W8A8 fake quant)
-6. ModuleSelector produzindo alpha + entropy_reg
-7. apply_gradient_surgery sem erro
-8. MetaConfigurator sem erro
-9. KillSwitch detectando condições de kill
-10. Multimodal encoders importáveis
-
-NÃO treina — apenas valida que o pipeline está íntegro.
-"""
-from __future__ import annotations
-
-import os
-import sys
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
-
-os.environ.setdefault("OMP_NUM_THREADS", "2")
-os.environ.setdefault("MKL_NUM_THREADS", "2")
-
-import torch
-torch.set_num_threads(2)
-
-
-def test_imports():
-    """Testa importação de todos os módulos."""
-    print("=== Test 1: Imports ===")
-    try:
-        from bigru_t import (
-            UnifiedModel, UnifiedModelConfig, create_unified_model,
-            u8cell_T, BiGRU4, TransformerUnit, OrqCell, TrainT, HypT,
-            ModuleSelector,
-            QuantizedLinear, quantize_tensor, apply_w8a8,
-            apply_gradient_surgery, orthogonalize_gradient,
-            MetaConfigurator,
-            KillSwitch, KillSwitchState,
-            BiGRU_T_Trainer, TrainerConfig,
-        )
-        print("  OK: all imports successful")
-        return True
-    except Exception as e:
-        print(f"  FAIL: {e}")
-        import traceback; traceback.print_exc()
-        return False
-
-
-def test_model_forward():
-    """Testa forward pass do UnifiedModel."""
-    print("\n=== Test 2: Model forward ===")
-    try:
-        from bigru_t import create_unified_model, UnifiedModelConfig
-        config = UnifiedModelConfig(
-            vocab_size=1000,
-            d_model=32,
-            max_seq_len=16,
-            pad_token_id=1,
-            max_modules=2,  # pequeno para teste rápido
-            bigru_hidden=8,
-            d_transformer=16,
-            nhead_tu=2,
-            d_ff_tu=32,
-            output_dim_u8cell=16,
-            cache_len=4,
-            d_cache=32,
-            nhead_orq=2,
-            d_ff_orq=64,
-            trainT_dim=32,
-            nhead_train=2,
-            d_ff_train=64,
-            num_layers_train=1,
-            hypT_dim=32,
-            nhead_hyp=2,
-            d_ff_hyp=64,
-            num_layers_hyp=1,
-        )
-        model, _ = create_unified_model(config)
-        params = model.count_parameters()
-        print(f"  params: {params['total']:,} ({params['total_M']:.3f}M)")
-
-        # Forward com token IDs
-        x = torch.randint(0, 1000, (2, 16))  # (batch=2, T=16)
-        y_hat, delta = model(x, temperature=1.0, use_hypothesis=False)
-        assert y_hat.shape == (2, 1000), f"y_hat shape {y_hat.shape} != (2, 1000)"
-        assert delta.shape == (2, 1000), f"delta shape {delta.shape} != (2, 1000)"
-        print(f"  OK: forward y_hat {y_hat.shape}, delta {delta.shape}")
-
-        # Forward com hipótese
-        y_hat, delta = model(x, temperature=1.0, use_hypothesis=True, stop_grad_hyp=True)
-        assert y_hat.shape == (2, 1000)
-        assert delta.shape == (2, 1000)
-        print(f"  OK: forward with hypothesis")
-
-        # Forward com return_aux
-        y_hat, delta, aux = model(x, temperature=1.0, use_hypothesis=False, return_aux=True)
-        assert "entropy_reg" in aux
-        assert "alpha" in aux
-        assert aux["alpha"].shape == (2,)
-        print(f"  OK: return_aux entropy_reg={aux['entropy_reg'].item():.4f}, alpha={aux['alpha'].tolist()}")
-
-        return True
-    except Exception as e:
-        print(f"  FAIL: {e}")
-        import traceback; traceback.print_exc()
-        return False
-
-
-def test_backward():
-    """Testa backward pass."""
-    print("\n=== Test 3: Backward ===")
-    try:
-        from bigru_t import create_unified_model, UnifiedModelConfig
-        import torch.nn.functional as F
-        config = UnifiedModelConfig(
-            vocab_size=100, d_model=16, max_seq_len=8, pad_token_id=1,
-            max_modules=2, bigru_hidden=4, d_transformer=8, nhead_tu=2, d_ff_tu=16,
-            output_dim_u8cell=8, cache_len=4, d_cache=16, nhead_orq=2, d_ff_orq=32,
-            trainT_dim=16, nhead_train=2, d_ff_train=32, num_layers_train=1,
-            hypT_dim=16, nhead_hyp=2, d_ff_hyp=32, num_layers_hyp=1,
-        )
-        model, _ = create_unified_model(config)
-        x = torch.randint(0, 100, (2, 8))
-        target = torch.randint(0, 100, (2,))
-        y_hat, _ = model(x, use_hypothesis=False)
-        loss = F.cross_entropy(y_hat, target)
-        loss.backward()
-        # Verifica que gradientes foram computados
-        n_with_grad = sum(1 for p in model.parameters() if p.grad is not None and p.grad.abs().sum() > 0)
-        n_total = sum(1 for p in model.parameters() if p.requires_grad)
-        print(f"  OK: backward done, loss={loss.item():.4f}, {n_with_grad}/{n_total} params have grad")
-        return True
-    except Exception as e:
-        print(f"  FAIL: {e}")
-        import traceback; traceback.print_exc()
-        return False
-
-
-def test_quantized_linear():
-    """Testa QuantizedLinear (Lema 3)."""
-    print("\n=== Test 4: QuantizedLinear (W8A8) ===")
-    try:
-        from bigru_t.quantization.quantized_linear import QuantizedLinear, quantize_tensor
-        # Test quantize_tensor
-        x = torch.randn(100)
-        x_q = quantize_tensor(x, num_bits=8)
-        err = (x - x_q).abs().max().item()
-        print(f"  quantize_tensor max err: {err:.4f}")
-
-        # Test QuantizedLinear
-        ql = QuantizedLinear(10, 5)
-        x = torch.randn(2, 10)
-        y = ql(x)
-        assert y.shape == (2, 5)
-        print(f"  OK: QuantizedLinear forward {y.shape}")
-
-        # Backward
-        y.sum().backward()
-        assert ql.weight.grad is not None
-        print(f"  OK: QuantizedLinear backward, grad norm {ql.weight.grad.norm().item():.4f}")
-        return True
-    except Exception as e:
-        print(f"  FAIL: {e}")
-        return False
-
-
-def test_gradient_surgery():
-    """Testa apply_gradient_surgery (Lema 2)."""
-    print("\n=== Test 5: Gradient surgery ===")
-    try:
-        from bigru_t.training.gradient_surgery import orthogonalize_gradient, apply_gradient_surgery
-        from bigru_t import create_unified_model, UnifiedModelConfig
-        import torch.nn.functional as F
-
-        # Test orthogonalize_gradient — caso conflitante
-        g_main = torch.tensor([1.0, 0.0])
-        g_hyp = torch.tensor([-1.0, 0.0])  # conflitante (dot = -1 < 0)
-        g_orth = orthogonalize_gradient(g_main, g_hyp)
-        # Projeção: g_hyp - (dot/norm_sq) * g_main = (-1, 0) - (-1/1) * (1, 0) = (0, 0)
-        assert torch.allclose(g_orth, torch.zeros(2), atol=1e-6), f"Expected (0,0), got {g_orth}"
-        print(f"  OK: orthogonalize conflitante → {g_orth.tolist()} (should be [0, 0])")
-
-        # Test orthogonalize_gradient — caso alinhado
-        g_main = torch.tensor([1.0, 0.0])
-        g_hyp = torch.tensor([0.5, 0.0])  # alinhado (dot = 0.5 > 0)
-        g_orth = orthogonalize_gradient(g_main, g_hyp)
-        # Sem projeção: g_hyp permanece
-        assert torch.allclose(g_orth, g_hyp), f"Expected {g_hyp}, got {g_orth}"
-        print(f"  OK: orthogonalize alinhado → {g_orth.tolist()} (should be [0.5, 0.0])")
-
-        # Test apply_gradient_surgery end-to-end
-        config = UnifiedModelConfig(
-            vocab_size=50, d_model=8, max_seq_len=4, pad_token_id=1,
-            max_modules=2, bigru_hidden=2, d_transformer=4, nhead_tu=2, d_ff_tu=8,
-            output_dim_u8cell=4, cache_len=2, d_cache=8, nhead_orq=2, d_ff_orq=16,
-            trainT_dim=8, nhead_train=2, d_ff_train=16, num_layers_train=1,
-            hypT_dim=8, nhead_hyp=2, d_ff_hyp=16, num_layers_hyp=1,
-        )
-        model, _ = create_unified_model(config)
-        x = torch.randint(0, 50, (2, 4))
-        target = torch.randint(0, 50, (2,))
-        y_hat_main, delta = model(x, use_hypothesis=True, stop_grad_hyp=True)
-        loss_main = F.cross_entropy(y_hat_main, target)
-        loss_hyp = F.cross_entropy(y_hat_main + delta, target)
-        apply_gradient_surgery(model, loss_main, loss_hyp)
-        n_with_grad = sum(1 for p in model.parameters() if p.grad is not None)
-        print(f"  OK: apply_gradient_surgery, {n_with_grad} params have .grad")
-        return True
-    except Exception as e:
-        print(f"  FAIL: {e}")
-        import traceback; traceback.print_exc()
-        return False
-
-
-def test_meta_configurator():
-    """Testa MetaConfigurator (Lema 4)."""
-    print("\n=== Test 6: MetaConfigurator ===")
-    try:
-        from bigru_t import create_unified_model, UnifiedModelConfig
-        from bigru_t.training.meta_configurator import MetaConfigurator
-
-        config = UnifiedModelConfig(
-            vocab_size=50, d_model=8, max_seq_len=4, pad_token_id=1,
-            max_modules=2, bigru_hidden=2, d_transformer=4, nhead_tu=2, d_ff_tu=8,
-            output_dim_u8cell=4, cache_len=2, d_cache=8, nhead_orq=2, d_ff_orq=16,
-            trainT_dim=8, nhead_train=2, d_ff_train=16, num_layers_train=1,
-            hypT_dim=8, nhead_hyp=2, d_ff_hyp=16, num_layers_hyp=1,
-        )
-        model, _ = create_unified_model(config)
-        meta = MetaConfigurator(model, meta_lr=0.01, sharpness_lambda=0.01)
-
-        # Initial T and tau
-        print(f"  Initial: T={meta.temperature:.4f}, tau={meta.tau:.4f}")
-
-        # Run one meta step
-        x = torch.randint(0, 50, (2, 4))
-        target = torch.randint(0, 50, (2,))
-        result = meta.forward_with_meta(x, target)
-        print(f"  After 1 step: T={result['T_new']:.4f}, tau={result['tau_new']:.4f}, "
-              f"loss_val={result['loss_val']:.4f}, sharpness={result['sharpness']:.4f}")
-
-        # Check tau was synced to model
-        assert model.tau.item() == result["tau_new"], f"model.tau {model.tau.item()} != {result['tau_new']}"
-        print(f"  OK: MetaConfigurator synced model.tau = {model.tau.item():.4f}")
-        return True
-    except Exception as e:
-        print(f"  FAIL: {e}")
-        import traceback; traceback.print_exc()
-        return False
-
-
-def test_kill_switch():
-    """Testa KillSwitch."""
-    print("\n=== Test 7: KillSwitch ===")
-    try:
-        from bigru_t.training.kill_switch import KillSwitch
-        ks = KillSwitch(loss_patience=3, ram_threshold_pct=99.9, disk_min_free_gb=0.001)
-
-        # Simula 5 steps com loss não decrescente
-        for i in range(5):
-            state = ks.check(i, loss=10.0, active_modules=2)
-        assert state.reason is not None, "Expected kill after patience exhausted"
-        print(f"  OK: kill triggered after {state.step} steps: {state.reason}")
-
-        # Resumo
-        summary = ks.summary()
-        print(f"  OK: summary keys: {list(summary.keys())}")
-        return True
-    except Exception as e:
-        print(f"  FAIL: {e}")
-        return False
-
-
-def test_multimodal_imports():
-    """Testa importação dos encoders multimodais (reaproveitados)."""
-    print("\n=== Test 8: Multimodal imports ===")
-    try:
-        # Não importa os módulos diretamente (podem ter dependências pesadas)
-        # Apenas verifica que os arquivos existem
-        multimodal_dir = Path(__file__).parent.parent / "src" / "bigru_t" / "multimodal"
-        expected = ["text_encoder.py", "image_encoder.py", "audio_encoder.py", "video_encoder.py", "modal_router.py", "fusion_layer.py"]
-        for f in expected:
-            assert (multimodal_dir / f).exists(), f"Missing {f}"
-        print(f"  OK: {len(expected)} multimodal modules present")
-        return True
-    except Exception as e:
-        print(f"  FAIL: {e}")
-        return False
-
-
-def test_utils_reused():
-    """Verifica que módulos utilitários reaproveitados estão presentes."""
-    print("\n=== Test 9: Reused utility modules ===")
-    try:
-        utils_dir = Path(__file__).parent.parent / "src" / "bigru_t" / "utils"
-        expected = ["hardware_detector.py", "xeon_runtime.py", "oom_guard.py", "memory_monitor.py", "tensor_ops.py", "validators.py", "logging_utils.py"]
-        for f in expected:
-            assert (utils_dir / f).exists(), f"Missing {f}"
-        print(f"  OK: {len(expected)} utility modules present")
-
-        # Test xeon_runtime import
-        try:
-            from bigru_t.utils.xeon_runtime import optimize_xeon_environment
-            optimize_xeon_environment()
-            print(f"  OK: optimize_xeon_environment() called")
-        except Exception as e:
-            print(f"  WARN: xeon_runtime optimize failed: {e}")
-        return True
-    except Exception as e:
-        print(f"  FAIL: {e}")
-        return False
-
-
-def main():
-    print("=" * 60)
-    print("BiGRU_T_version — Smoke Test")
-    print("=" * 60)
-
-    tests = [
-        test_imports,
-        test_model_forward,
-        test_backward,
-        test_quantized_linear,
-        test_gradient_surgery,
-        test_meta_configurator,
-        test_kill_switch,
-        test_multimodal_imports,
-        test_utils_reused,
-    ]
-    results = []
-    for t in tests:
-        try:
-            r = t()
-            results.append(r)
-        except Exception as e:
-            print(f"  CRASH: {e}")
-            results.append(False)
-
-    print("\n" + "=" * 60)
-    passed = sum(results)
-    total = len(results)
-    print(f"Smoke test: {passed}/{total} passed")
-    print("=" * 60)
-    sys.exit(0 if passed == total else 1)
-
-
-if __name__ == "__main__":
-    main()