Create routes .py

routes .py

@@ -0,0 +1,148 @@
+"""
+data/routes.py
+Descoberta automática de Rotas Atômicas a partir das chaves estrangeiras.
+
+Uma "Rota Atômica" é um caminho entre tabelas conectadas via FKs.
+NÃO convertemos para grafo — as rotas são inferidas do schema relacional.
+"""
+
+from dataclasses import dataclass, field
+from typing import List, Dict, Tuple, Optional
+import numpy as np
+from collections import deque
+
+
+# ─── SCHEMA HARDCODED TPC-H ──────────────────────────────────────────────────
+
+TPCH_FOREIGN_KEYS: List[Tuple[str, str, str, str]] = [
+    # (tabela_origem, fk_coluna, tabela_destino, pk_coluna)
+    ("orders",    "o_custkey",   "customers", "c_custkey"),
+    ("lineitem",  "l_orderkey",  "orders",    "o_orderkey"),
+    ("lineitem",  "l_suppkey",   "supplier",  "s_suppkey"),
+    ("lineitem",  "l_partkey",   "part",      "p_partkey"),
+    ("customers", "c_nationkey", "nation",    "n_nationkey"),
+    ("supplier",  "s_nationkey", "nation",    "n_nationkey"),
+]
+
+
+@dataclass
+class RouteConfig:
+    max_hops: int = 3
+    target_table: str = "customers"   # entidade alvo da predição
+    min_attention_weight: float = 0.1
+
+
+@dataclass
+class AtomicRoute:
+    path: List[str]            # ex: ["customers", "orders", "lineitem"]
+    fk_edges: List[Tuple]      # [(from, fk_col, to, pk_col), ...]
+    n_hops: int = 0
+    attention_weight: float = 1.0
+    active: bool = True
+
+    def __post_init__(self):
+        self.n_hops = len(self.path) - 1
+
+    def __repr__(self):
+        return f"Route({' → '.join(self.path)}, α={self.attention_weight:.3f})"
+
+
+def _build_adjacency(fk_list: List[Tuple]) -> Dict[str, List[Tuple]]:
+    """Constrói lista de adjacência bidirecional a partir das FKs."""
+    adj = {}
+    for (src, src_col, dst, dst_col) in fk_list:
+        if src not in adj:
+            adj[src] = []
+        if dst not in adj:
+            adj[dst] = []
+        adj[src].append((dst, src_col, dst_col, "forward"))
+        adj[dst].append((src, dst_col, src_col, "backward"))
+    return adj
+
+
+def discover_atomic_routes(
+    tables: Dict,
+    config: RouteConfig,
+    fk_list: Optional[List[Tuple]] = None,
+) -> List[AtomicRoute]:
+    """
+    BFS a partir da tabela alvo para descobrir todas as rotas atômicas
+    até `config.max_hops` saltos.
+
+    NÃO requer conversão para grafo — opera diretamente no schema.
+    """
+    if fk_list is None:
+        fk_list = TPCH_FOREIGN_KEYS
+
+    adj = _build_adjacency(fk_list)
+    start = config.target_table
+    routes: List[AtomicRoute] = []
+
+    # BFS: (caminho_atual, arestas_percorridas, visitados)
+    queue = deque()
+    queue.append(([start], [], {start}))
+
+    while queue:
+        path, edges, visited = queue.popleft()
+        current = path[-1]
+        n_hops  = len(path) - 1
+
+        # Só registra rotas com pelo menos 1 hop
+        if n_hops >= 1:
+            fk_edges = [
+                (fk_list[i] if i < len(fk_list) else edges[i])
+                for i in range(len(edges))
+            ]
+            route = AtomicRoute(
+                path=list(path),
+                fk_edges=list(edges),
+                attention_weight=_initial_attention(n_hops),
+                active=(n_hops <= 2),
+            )
+            routes.append(route)
+
+        if n_hops >= config.max_hops:
+            continue
+
+        # Expande vizinhos
+        for (neighbor, col_a, col_b, direction) in adj.get(current, []):
+            if neighbor not in visited and neighbor in tables:
+                new_path    = path + [neighbor]
+                new_edges   = edges + [(current, col_a, neighbor, col_b)]
+                new_visited = visited | {neighbor}
+                queue.append((new_path, new_edges, new_visited))
+
+    # Ordena por peso inicial (hops menores têm mais peso)
+    routes.sort(key=lambda r: -r.attention_weight)
+
+    # Normaliza pesos com softmax simulado
+    weights = np.array([r.attention_weight for r in routes])
+    weights = np.exp(weights) / np.exp(weights).sum()
+    for r, w in zip(routes, weights):
+        r.attention_weight = float(w)
+
+    return routes
+
+
+def _initial_attention(n_hops: int) -> float:
+    """Peso inicial decrescente por número de hops (heurística)."""
+    return 1.0 / (n_hops ** 1.5)
+
+
+def routes_to_dataframe(routes: List[AtomicRoute]):
+    """Converte lista de rotas para DataFrame para exibição."""
+    import pandas as pd
+    return pd.DataFrame([{
+        "Rota":     " → ".join(r.path),
+        "Hops":     r.n_hops,
+        "Peso α":   round(r.attention_weight, 4),
+        "Ativa":    r.active,
+    } for r in routes])
+
+
+def get_feature_tables_for_route(
+    route: AtomicRoute,
+    tables: Dict,
+) -> List:
+    """Retorna as tabelas (DataFrames) na ordem da rota."""
+    return [tables[t] for t in route.path if t in tables]