Upload cnn_bigru/data/streaming_dataset.py with huggingface_hub

cnn_bigru/data/streaming_dataset.py

@@ -0,0 +1,308 @@
+"""streaming_dataset.py — Carregador de dataset streaming para o CNN-BiGRU.
+
+Adaptado de xavante_work/flexnet/streaming_datasets_v13_9.py, com:
+    - Modo streaming (IterableDataset, sem materialização completa)
+    - Fallback sintético quando datasets externos estão indisponíveis
+    - Suporte multimodal: texto + imagem (placeholder) + áudio (placeholder)
+    - Garantia de produzir N amostras para o teste de 50 samples
+
+Para o projeto CNN-BiGRU, usamos um gerador sintético multi-modal padrão
+quando a rede ou HF não está acessível, garantindo que o teste de 50 amostras
+sempre execute.
+"""
+from __future__ import annotations
+
+import logging
+import os
+import random
+from dataclasses import dataclass, field
+from typing import Any, Dict, Iterator, List, Optional, Sequence
+
+import numpy as np
+import torch
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class MultimodalSample:
+    """Amostra multimodal: texto + imagem (HxWxC float) + áudio (spec TxF)."""
+    sample_id: int
+    text_a: str            # Stream A (e.g. pergunta/título)
+    text_b: str            # Stream B (e.g. contexto/corpo)
+    image: Optional[np.ndarray] = None     # [H, W, C] float32 in [0,1]
+    audio: Optional[np.ndarray] = None     # [T, F] spectrogram float32
+    label: int = 0
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+# Dataset padrão V13.9.1 (mesmos do reference streaming_datasets_v13_9.py)
+DEFAULT_DATASETS = [
+    "CEIA-POSITIVO/ultrachat_br_clustred_balanced_v1",
+    "Madras1/corpus-ptbr-v2",
+    "rhaymison/multmodal_175k_portuguese",
+    "TucanoBR/GigaVerbo",
+    "nvidia/OpenMathReasoning",
+    "MathLLMs/MathVision",
+    "nvidia/OpenMathInstruct-2",
+    "dominguesm/restore-punctuation-ptbr-dataset",
+    "carolina-c4ai/corpus-carolina",
+]
+
+
+def _extract_field(sample: Dict[str, Any], candidates: List[str]) -> Optional[str]:
+    for f in candidates:
+        if f in sample:
+            v = sample[f]
+            if isinstance(v, str) and v.strip():
+                return v
+            if isinstance(v, list):
+                parts = []
+                for m in v:
+                    if isinstance(m, dict):
+                        c = m.get("content", "")
+                        if isinstance(c, str) and c.strip():
+                            parts.append(c)
+                    elif isinstance(m, str) and m.strip():
+                        parts.append(m)
+                if parts:
+                    return "\n".join(parts)
+    return None
+
+
+def _try_load_hf_streaming(
+    dataset_name: str,
+    split: str = "train",
+    hf_token: Optional[str] = None,
+):
+    """Tenta carregar dataset HF em streaming. Retorna None se falhar."""
+    try:
+        from datasets import load_dataset
+        ds = load_dataset(dataset_name, split=split, streaming=True, token=hf_token)
+        logger.info("Streaming OK: %s[%s]", dataset_name, split)
+        return ds
+    except Exception as e:
+        logger.info("Streaming falhou para %s: %s", dataset_name, str(e)[:120])
+        return None
+
+
+def synthetic_multimodal_stream(
+    n_samples: int,
+    seed: int = 42,
+    image_size: tuple = (28, 28, 1),
+    audio_shape: tuple = (64, 40),
+    vocab_texts: Optional[Sequence[str]] = None,
+) -> Iterator[MultimodalSample]:
+    """Gera amostras sintéticas multimodais determinísticas.
+
+    Usado quando o dataset HF está indisponível (offline, rate-limit, etc.).
+    Garante que o teste de 50 amostras sempre execute.
+    """
+    rng = random.Random(seed)
+    np_rng = np.random.RandomState(seed)
+
+    if vocab_texts is None:
+        vocab_texts = [
+            "o modelo aprende padrões locais com convoluções",
+            "a ponte de cooperação troca informações entre fluxos",
+            "grus capturam dependências temporais bidirecionais",
+            "atenção cruzada reduz perplexidade em textos longos",
+            "a fusão multimodal combina texto imagem e áudio",
+            "penalidades evitam repetições viciosas na geração",
+            "a camada anti-alucinação usa lógica fuzzy de lukasiewicz",
+            "o verificador classifica passos com sigmoid binária",
+            "ajustes dinâmicos de lr controlam explosão de gradiente",
+            "normalização espectral estabiliza o treinamento",
+        ]
+
+    n_vocab = len(vocab_texts)
+    for i in range(n_samples):
+        text_a = vocab_texts[rng.randrange(n_vocab)]
+        text_b = vocab_texts[rng.randrange(n_vocab)]
+        # Imagem sintética: padrões estruturados simples
+        img = np_rng.rand(*image_size).astype(np.float32)
+        # Adiciona um padrão que depende do índice (para discriminação)
+        img[i % image_size[0], :, :] = 1.0
+        # Áudio sintético: espectrograma
+        aud = np_rng.rand(*audio_shape).astype(np.float32) * 0.5
+        # Espectro com picos determinísticos
+        aud[i % audio_shape[0], :] += 0.5
+        yield MultimodalSample(
+            sample_id=i,
+            text_a=text_a,
+            text_b=text_b,
+            image=img,
+            audio=aud,
+            label=i % 3,
+            metadata={"source": "synthetic", "idx": i},
+        )
+
+
+class MultimodalStreamingDataset(torch.utils.data.IterableDataset):
+    """Dataset streaming multimodal. Suporta HF + fallback sintético.
+
+    Args:
+        n_samples: número total de amostras a produzir.
+        hf_datasets: lista de datasets HF para tentar (em ordem).
+        hf_token: token HF (será limpo após uso).
+        use_synthetic_fallback: se True, usa sintético quando HF falha.
+        seed: seed para reprodutibilidade.
+    """
+
+    def __init__(
+        self,
+        n_samples: int = 50,
+        hf_datasets: Optional[List[str]] = None,
+        hf_token: Optional[str] = None,
+        use_synthetic_fallback: bool = True,
+        seed: int = 42,
+        image_size: tuple = (28, 28, 1),
+        audio_shape: tuple = (64, 40),
+    ):
+        super().__init__()
+        self.n_samples = n_samples
+        self.hf_datasets = hf_datasets or DEFAULT_DATASETS
+        self.hf_token = hf_token
+        self.use_synthetic_fallback = use_synthetic_fallback
+        self.seed = seed
+        self.image_size = image_size
+        self.audio_shape = audio_shape
+
+    def _try_hf(self) -> Iterator[Dict[str, Any]]:
+        """Tenta carregar amostras HF. Retorna iterator vazio se falhar."""
+        for ds_name in self.hf_datasets:
+            ds = _try_load_hf_streaming(ds_name, split="train", hf_token=self.hf_token)
+            if ds is None:
+                continue
+            count = 0
+            text_candidates = ["text", "content", "question", "problem", "input",
+                               "conversa", "description", "prompt", "instruction"]
+            label_candidates = ["answer", "response", "output", "solution",
+                                 "punctuated", "restored"]
+            for raw in ds:
+                if count >= self.n_samples:
+                    break
+                try:
+                    text_a = _extract_field(raw, text_candidates) or ""
+                    text_b = _extract_field(raw, label_candidates) or ""
+                    if len(text_a) < 5:
+                        text_a = "pergunta de exemplo sobre o tema"
+                    if len(text_b) < 5:
+                        text_b = "resposta de exemplo para contexto"
+                    # Image/audio placeholder: geramos sintéticos para manter multimodal
+                    img = np.random.rand(*self.image_size).astype(np.float32) * 0.5
+                    aud = np.random.rand(*self.audio_shape).astype(np.float32) * 0.5
+                    yield {
+                        "sample_id": count,
+                        "text_a": text_a[:200],  # truncado para teste
+                        "text_b": text_b[:200],
+                        "image": img,
+                        "audio": aud,
+                        "label": count % 3,
+                        "metadata": {"source": ds_name, "idx": count},
+                    }
+                    count += 1
+                except Exception as e:
+                    logger.debug("skip sample %d from %s: %s", count, ds_name, e)
+                    continue
+            if count > 0:
+                logger.info("HF yield %d samples de %s", count, ds_name)
+                return
+
+    def __iter__(self) -> Iterator[Dict[str, Any]]:
+        # Tentar HF primeiro
+        if self.hf_datasets:
+            try:
+                yield from self._try_hf()
+                return
+            except Exception as e:
+                logger.warning("HF streaming falhou (%s) — usando fallback sintético", e)
+
+        # Fallback sintético
+        if self.use_synthetic_fallback:
+            for s in synthetic_multimodal_stream(
+                n_samples=self.n_samples,
+                seed=self.seed,
+                image_size=self.image_size,
+                audio_shape=self.audio_shape,
+            ):
+                yield {
+                    "sample_id": s.sample_id,
+                    "text_a": s.text_a,
+                    "text_b": s.text_b,
+                    "image": s.image,
+                    "audio": s.audio,
+                    "label": s.label,
+                    "metadata": s.metadata,
+                }
+
+
+def collate_multimodal(
+    batch: List[Dict[str, Any]],
+    tokenizer,
+    max_len: int = 64,
+) -> Dict[str, torch.Tensor]:
+    """Cola um batch de amostras multimodais em tensores.
+
+    Returns dict com:
+        input_ids_a: [B, T]   (stream A)
+        input_ids_b: [B, T]   (stream B)
+        attn_mask_a: [B, T]
+        attn_mask_b: [B, T]
+        images:      [B, C, H, W]
+        audios:      [B, 1, T, F]
+        labels:      [B]
+    """
+    texts_a = [b["text_a"] for b in batch]
+    texts_b = [b["text_b"] for b in batch]
+    ids_a = tokenizer.encode_batch(texts_a, add_special=True)
+    ids_b = tokenizer.encode_batch(texts_b, add_special=True)
+
+    pad_id = tokenizer.pad_id
+
+    def _pad(seqs, max_len):
+        out = []
+        masks = []
+        for s in seqs:
+            s = s[:max_len]
+            n = len(s)
+            padded = s + [pad_id] * (max_len - n)
+            mask = [1] * n + [0] * (max_len - n)
+            out.append(padded)
+            masks.append(mask)
+        return out, masks
+
+    padded_a, masks_a = _pad(ids_a, max_len)
+    padded_b, masks_b = _pad(ids_b, max_len)
+
+    images = np.stack([b["image"] for b in batch])  # [B, H, W, C]
+    # converte para [B, C, H, W]
+    if images.ndim == 4:
+        images = np.transpose(images, (0, 3, 1, 2))
+    else:
+        images = images[:, None, :, :]  # add channel dim
+
+    audios = np.stack([b["audio"] for b in batch])  # [B, T, F]
+    audios = audios[:, None, :, :]  # [B, 1, T, F]
+
+    labels = np.array([b["label"] for b in batch], dtype=np.int64)
+
+    return {
+        "input_ids_a": torch.tensor(padded_a, dtype=torch.long),
+        "input_ids_b": torch.tensor(padded_b, dtype=torch.long),
+        "attn_mask_a": torch.tensor(masks_a, dtype=torch.float),
+        "attn_mask_b": torch.tensor(masks_b, dtype=torch.float),
+        "images": torch.tensor(images, dtype=torch.float32),
+        "audios": torch.tensor(audios, dtype=torch.float32),
+        "labels": torch.tensor(labels, dtype=torch.long),
+        "sample_ids": [b["sample_id"] for b in batch],
+    }
+
+
+__all__ = [
+    "MultimodalSample",
+    "MultimodalStreamingDataset",
+    "collate_multimodal",
+    "DEFAULT_DATASETS",
+    "synthetic_multimodal_stream",
+]