#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Vérification de Qwen3.8-27B (dépôt Qwen/Qwen3.8-27B, publié le 14 août 2026).

Aucune dépendance : Python 3 pur, pas de numpy, pas de torch.

Ce script part uniquement du `config.json` publié avec les poids et recalcule :

  1. la structure des couches                   (annoncé : hybride 3:1)
  2. le nombre total de paramètres              (annoncé : « 27B », affiché « 28B »)
  3. la taille du dépôt en BF16                 (observé  : 55,6 Go)
  4. le poids du cache clé-valeur               (non annoncé)
  5. l'état récurrent de l'attention linéaire
  6. le contexte réellement tenable par carte

Il conserve enfin le calcul contrefactuel qui servait de borne avant la
publication, pour mesurer l'écart entre ce qui avait été prévu et ce qui est.

Usage :  python3 verif-qwen3-8-27b.py
"""

import sys

# --------------------------------------------------------------------------
# 1. La configuration publiée
# --------------------------------------------------------------------------
# Recopiée depuis
# https://huggingface.co/Qwen/Qwen3.8-27B/raw/main/config.json
# La liste `layer_types` est résumée : 64 entrées, répétition de
# ["linear_attention"] * 3 + ["full_attention"], 16 fois.

TEXTE = {
    "hidden_size": 5120,
    "vocab_size": 248320,
    "num_hidden_layers": 64,
    "full_attention_interval": 4,
    "intermediate_size": 17408,        # réseau dense : le modèle n'est PAS un MoE
    # attention complète (gatée)
    "num_attention_heads": 24,
    "num_key_value_heads": 4,
    "head_dim": 256,
    "attn_output_gate": True,
    "partial_rotary_factor": 0.25,
    # attention linéaire (Gated DeltaNet)
    "linear_num_value_heads": 48,
    "linear_num_key_heads": 16,
    "linear_key_head_dim": 128,
    "linear_value_head_dim": 128,
    "linear_conv_kernel_dim": 4,
    # divers
    "tie_word_embeddings": False,
    "mtp_num_hidden_layers": 1,
    "max_position_embeddings": 262144,
}

# Encodeur visuel — présent, ce qui fait du modèle un multimodal natif.
VISION = {
    "depth": 27,
    "hidden_size": 1152,
    "intermediate_size": 4304,
    "num_heads": 16,
    "in_channels": 3,
    "patch_size": 16,
    "temporal_patch_size": 2,
    "spatial_merge_size": 2,
    "num_position_embeddings": 2304,
    "out_hidden_size": 5120,
}

# Références externes à confronter au calcul.
OBSERVE_DEPOT_GO = 55.6         # taille du dépôt affichée par Hugging Face
CONTEXTE_ETENDU = 1_000_000     # « extensible up to 1,000,000 tokens »

G = 10 ** 9


def titre(txt):
    print()
    print(txt)
    print("-" * len(txt))


def ligne(label, valeur, unite=""):
    print("  {:<48} {:>16} {}".format(label, valeur, unite))


# --------------------------------------------------------------------------
# 2. Structure des couches
# --------------------------------------------------------------------------
def compte_couches(cfg):
    """16 blocs de (3 linéaires + 1 complète) = 64 couches."""
    total = cfg["num_hidden_layers"]
    n_complete = total // cfg["full_attention_interval"]
    return total - n_complete, n_complete


# --------------------------------------------------------------------------
# 3. Paramètres par bloc
# --------------------------------------------------------------------------
def params_attention_complete(cfg):
    """
    Attention complète gatée. `attn_output_gate` étant vrai, la projection des
    requêtes produit aussi la porte de sortie : elle est donc doublée.
    """
    h = cfg["hidden_size"]
    d = cfg["head_dim"]
    dim_q = cfg["num_attention_heads"] * d
    dim_kv = cfg["num_key_value_heads"] * d

    q = h * dim_q * (2 if cfg["attn_output_gate"] else 1)
    k = h * dim_kv
    v = h * dim_kv
    o = dim_q * h
    normes = 2 * d
    return q + k + v + o + normes


def params_attention_lineaire(cfg):
    """Gated DeltaNet : projection unique (q, k, v, z), portes, convolution."""
    h = cfg["hidden_size"]
    dim_qk = cfg["linear_num_key_heads"] * cfg["linear_key_head_dim"]
    dim_v = cfg["linear_num_value_heads"] * cfg["linear_value_head_dim"]

    in_proj_qkvz = h * (2 * dim_qk + 2 * dim_v)
    in_proj_ba = h * (2 * cfg["linear_num_value_heads"])
    conv = (2 * dim_qk + dim_v) * cfg["linear_conv_kernel_dim"]
    out_proj = dim_v * h
    normes = cfg["linear_value_head_dim"]
    return in_proj_qkvz + in_proj_ba + conv + out_proj + normes


def params_ffn(cfg):
    """Réseau dense SwiGLU : gate, up, down. Pas d'experts dans ce modèle."""
    return 3 * cfg["hidden_size"] * cfg["intermediate_size"]


def params_vision(v):
    """
    Encodeur visuel de type ViT, plus le module de fusion vers la dimension
    du modèle de langage.

    La structure exacte n'est pas documentée : ce décompte est une
    reconstruction. Il ne pèse que ~1,7 % du total, et l'accord final avec la
    taille du dépôt confirme qu'il est du bon ordre.
    """
    h = v["hidden_size"]
    i = v["intermediate_size"]

    patch = (v["in_channels"] * v["temporal_patch_size"]
             * v["patch_size"] ** 2 * h) + h
    positions = v["num_position_embeddings"] * h

    qkv = h * 3 * h + 3 * h
    proj = h * h + h
    fc1 = h * i + i
    fc2 = i * h + h
    normes = 4 * h
    par_bloc = qkv + proj + fc1 + fc2 + normes
    blocs = v["depth"] * par_bloc

    fusion_in = h * v["spatial_merge_size"] ** 2
    fusion = (fusion_in * fusion_in + fusion_in
              + fusion_in * v["out_hidden_size"] + v["out_hidden_size"]
              + 2 * h)

    return patch + positions + blocs + fusion


# --------------------------------------------------------------------------
# 4. Total
# --------------------------------------------------------------------------
def decompte(txt, vis):
    n_lin, n_full = compte_couches(txt)
    h = txt["hidden_size"]

    p_full = params_attention_complete(txt)
    p_lin = params_attention_lineaire(txt)
    p_ffn = params_ffn(txt)

    embeddings = txt["vocab_size"] * h
    tete = 0 if txt["tie_word_embeddings"] else txt["vocab_size"] * h
    normes = 2 * h * txt["num_hidden_layers"] + h

    corps = (n_full * p_full + n_lin * p_lin
             + txt["num_hidden_layers"] * p_ffn + normes)
    texte = corps + embeddings + tete
    vision = params_vision(vis)
    mtp = txt["mtp_num_hidden_layers"] * (p_ffn + p_full)

    return {
        "n_lin": n_lin, "n_full": n_full,
        "p_full": p_full, "p_lin": p_lin, "p_ffn": p_ffn,
        "embeddings": embeddings, "corps": corps,
        "texte": texte, "vision": vision, "mtp": mtp,
        "total": texte + vision + mtp,
    }


# --------------------------------------------------------------------------
# 5. Mémoire d'inférence
# --------------------------------------------------------------------------
def kv_par_jeton(cfg, bits=16):
    """Seules les couches d'attention complète alimentent le cache."""
    _, n_full = compte_couches(cfg)
    dim_kv = cfg["num_key_value_heads"] * cfg["head_dim"]
    return n_full * 2 * dim_kv * bits // 8


def etat_lineaire(cfg):
    """État de DeltaNet : (d_k x d_v) par tête de valeur, en float32, constant."""
    n_lin, _ = compte_couches(cfg)
    par_couche = (cfg["linear_num_value_heads"]
                  * cfg["linear_key_head_dim"]
                  * cfg["linear_value_head_dim"] * 4)
    dim_qk = cfg["linear_num_key_heads"] * cfg["linear_key_head_dim"]
    dim_v = cfg["linear_num_value_heads"] * cfg["linear_value_head_dim"]
    conv = (2 * dim_qk + dim_v) * cfg["linear_conv_kernel_dim"] * 4
    return n_lin * (par_couche + conv)


# --------------------------------------------------------------------------
# 6. Exécution
# --------------------------------------------------------------------------
def main():
    d = decompte(TEXTE, VISION)
    echecs = []

    def verifie(nom, calcule, attendu, tolerance):
        ecart = abs(calcule - attendu) / attendu
        ok = ecart <= tolerance
        print("  [{}] {:<42} écart {:>6.2f} %".format(
            "OK " if ok else "ÉCHEC", nom, 100 * ecart))
        if not ok:
            echecs.append(nom)

    print("=" * 78)
    print("  Vérification de Qwen3.8-27B — Qwen/Qwen3.8-27B")
    print("  Source unique : config.json publié le 14 août 2026")
    print("=" * 78)

    titre("1. Structure")
    ligne("couches", d["n_lin"] + d["n_full"])
    ligne("couches à attention linéaire (DeltaNet)", d["n_lin"])
    ligne("couches à attention complète", d["n_full"])
    ligne("ratio", "{}:1".format(d["n_lin"] // d["n_full"]))
    ligne("réseau dense par couche (pas de MoE)", "oui")
    ligne("encodeur visuel", "{} couches".format(VISION["depth"]))

    titre("2. Paramètres par bloc")
    ligne("attention complète gatée", "{:.1f}".format(d["p_full"] / 1e6), "M")
    ligne("Gated DeltaNet", "{:.1f}".format(d["p_lin"] / 1e6), "M")
    ligne("réseau dense SwiGLU", "{:.1f}".format(d["p_ffn"] / 1e6), "M")
    ligne("table d'embeddings", "{:.3f}".format(d["embeddings"] / G), "G")

    titre("3. Total des paramètres")
    ligne("corps du modèle de langage", "{:.3f}".format(d["corps"] / G), "G")
    ligne("+ embeddings et tête de sortie", "{:.3f}".format(
        2 * d["embeddings"] / G), "G")
    ligne("= sous-total texte", "{:.3f}".format(d["texte"] / G), "G")
    ligne("encodeur visuel", "{:.3f}".format(d["vision"] / G), "G")
    ligne("couche MTP", "{:.3f}".format(d["mtp"] / G), "G")
    ligne("TOTAL", "{:.3f}".format(d["total"] / G), "G")
    ligne("étiquette Hugging Face", "28", "G")
    print()
    verifie("total vs étiquette « 28B »", d["total"], 28e9, 0.02)

    titre("4. Taille du dépôt en BF16")
    go = d["total"] * 2 / G
    ligne("2 octets par paramètre", "{:.2f}".format(go), "Go")
    ligne("taille affichée par Hugging Face", "{:.2f}".format(
        OBSERVE_DEPOT_GO), "Go")
    print()
    verifie("taille du dépôt vs 55,6 Go", go, OBSERVE_DEPOT_GO, 0.01)

    titre("5. Mémoire d'inférence")
    kv = kv_par_jeton(TEXTE)
    etat = etat_lineaire(TEXTE)
    ligne("cache KV par jeton", "{:.0f}".format(kv / 1024), "KiO")
    ligne("cache à 32 768 jetons", "{:.2f}".format(kv * 32768 / G), "Go")
    ligne("cache à 262 144 jetons (natif)", "{:.2f}".format(
        kv * TEXTE["max_position_embeddings"] / G), "Go")
    ligne("cache à 1 000 000 jetons (étendu)", "{:.2f}".format(
        kv * CONTEXTE_ETENDU / G), "Go")
    ligne("état récurrent DeltaNet (constant)", "{:.3f}".format(etat / G), "Go")

    titre("6. Ce que l'attention hybride économise")
    kv_tout = kv * TEXTE["num_hidden_layers"] // compte_couches(TEXTE)[1]
    ligne("si les 64 couches étaient complètes, par jeton",
          "{:.0f}".format(kv_tout / 1024), "KiO")
    ligne("    à 262 144 jetons", "{:.2f}".format(
        kv_tout * TEXTE["max_position_embeddings"] / G), "Go")
    ligne("économie apportée par l'hybride", "{:.0f}".format(
        100 * (1 - kv / kv_tout)), "%")

    titre("7. Contexte tenable, par carte")
    cartes = [("RTX 4090 / 5080", 24), ("RTX 5090", 32),
              ("L40S / RTX Pro 6000", 48), ("H100 / H200", 80),
              ("Mac 128 Go unifiés", 110)]
    for nom_q, bits in (("4 bits", 4), ("FP8", 8), ("BF16", 16)):
        poids = d["total"] * bits / 8
        print()
        print("  Poids en {} ({:.1f} Go) :".format(nom_q, poids / G))
        for nom_c, vram in cartes:
            dispo = vram * G * 0.90 - poids - etat
            if dispo <= 0:
                ligne("    " + nom_c, "ne charge pas")
                continue
            n = int(dispo / kv)
            plafond = " (plafonné à 1 000 000)" if n > CONTEXTE_ETENDU else ""
            n = min(n, CONTEXTE_ETENDU)
            ligne("    " + nom_c,
                  "{:>9,}".format(n).replace(",", " ") + plafond, "jetons")

    print()
    print("=" * 78)
    if echecs:
        print("  RÉSULTAT : {} vérification(s) en échec : {}".format(
            len(echecs), ", ".join(echecs)))
        return 1
    print("  RÉSULTAT : toutes les vérifications passent.")
    print("=" * 78)
    return 0


if __name__ == "__main__":
    sys.exit(main())
