Deep Learning

import numpy as np

class Neurone:
    """Un seul neurone artificiel."""
    def __init__(self, n_inputs, activation='relu'):
        # He initialization (recommandé pour ReLU)
        self.W = np.random.randn(n_inputs) * np.sqrt(2 / n_inputs)
        self.b = 0.0
        self.act = activation

    def _activation(self, z):
        if self.act == 'sigmoid': return 1 / (1 + np.exp(-z))
        if self.act == 'relu':    return np.maximum(0, z)
        if self.act == 'tanh':    return np.tanh(z)
        return z  # linéaire

    def forward(self, x):
        z = np.dot(self.W, x) + self.b   # combinaison linéaire
        a = self._activation(z)           # activation
        return a, z

# Exemple : prédiction cacao (Côte d'Ivoire)
# x = [pluviométrie, température, humidité, type_sol]
x = np.array([0.72, 0.65, 0.80, 0.40])
neurone = Neurone(n_inputs=4, activation='sigmoid')
a, z = neurone.forward(x)
print(f"z = {z:.4f} | a = {a:.4f}")
print("Bonne récolte probable" if a > 0.5 else "Risque faible")

class CoucheDense:
    def __init__(self, n_in, n_out, activation='relu'):
        # He initialization
        self.W = np.random.randn(n_out, n_in) * np.sqrt(2/n_in)
        self.b = np.zeros((n_out, 1))
        self.act = activation

    def forward(self, A_prev):
        self.A_prev = A_prev                  # cache pour backward
        self.Z = self.W @ A_prev + self.b     # combinaison linéaire
        self.A = self._activate(self.Z)       # activation
        return self.A

    def _activate(self, Z):
        if self.act == 'relu':    return np.maximum(0, Z)
        if self.act == 'sigmoid': return 1 / (1 + np.exp(-Z))
        if self.act == 'softmax':
            e = np.exp(Z - Z.max(axis=0, keepdims=True))
            return e / e.sum(axis=0, keepdims=True)
        return Z

    def _activate_deriv(self, Z):
        if self.act == 'relu':    return (Z > 0).astype(float)
        if self.act == 'sigmoid':
            s = 1 / (1 + np.exp(-Z)); return s * (1 - s)
        return np.ones_like(Z)  # linéaire

    def backward(self, dA, lr=1e-3):
        m   = self.A_prev.shape[1]
        dZ  = dA * self._activate_deriv(self.Z)  # ∂L/∂Z
        dW  = (dZ @ self.A_prev.T) / m            # ∂L/∂W
        db  = dZ.sum(axis=1, keepdims=True) / m   # ∂L/∂b
        dA_prev = self.W.T @ dZ                   # gradient propagé
        self.W -= lr * dW                          # mise à jour
        self.b -= lr * db
        return dA_prev

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers, regularizers

# ── Architecture ─────────────────────────────────────
def construire_modele(n_features, lr=1e-3):
    inputs = keras.Input(shape=(n_features,), name='features')
    x = layers.Dense(256, activation='relu',
                     kernel_regularizer=regularizers.l2(1e-4))(inputs)
    x = layers.BatchNormalization()(x)
    x = layers.Dropout(0.4)(x)
    x = layers.Dense(128, activation='relu')(x)
    x = layers.BatchNormalization()(x)
    x = layers.Dropout(0.3)(x)
    x = layers.Dense(64, activation='relu')(x)
    output = layers.Dense(1, name='prix_predit')(x)

    model = keras.Model(inputs, output)
    model.compile(
        optimizer=keras.optimizers.Adam(lr),
        loss='mse',
        metrics=['mae', 'mape']
    )
    return model

model = construire_modele(n_features=12)
model.summary()

# ── Callbacks professionnels ──────────────────────────
callbacks = [
    keras.callbacks.EarlyStopping(
        monitor='val_loss', patience=15,
        restore_best_weights=True, verbose=1
    ),
    keras.callbacks.ReduceLROnPlateau(
        monitor='val_loss', factor=0.3,
        patience=7, min_lr=1e-6, verbose=1
    ),
    keras.callbacks.ModelCheckpoint(
        'meilleur_modele.keras',
        monitor='val_loss', save_best_only=True
    ),
    keras.callbacks.TensorBoard(
        log_dir='./logs', histogram_freq=1
    ),
]

history = model.fit(
    X_train, y_train,
    validation_data=(X_val, y_val),
    epochs=200, batch_size=32,
    callbacks=callbacks
)

# ── Sauvegarde TFLite (mobile) ────────────────────────
converter = tf.lite.TFLiteConverter.from_keras_model(model)
tflite_model = converter.convert()
with open('prix_mil.tflite', 'wb') as f:
    f.write(tflite_model)

import torch

# Tenseur avec suivi du gradient
x = torch.tensor([2.0, 3.0], requires_grad=True)
y = x**2 + 3*x            # f(x) = x² + 3x
z = y.sum()               # scalaire pour backward
z.backward()              # calcule les gradients
print(x.grad)             # tensor([7., 9.]) = 2x+3

# Arrêter le suivi (inférence)
with torch.no_grad():
    pred = model(X_test)

import torch, torch.nn as nn
from torch.utils.data import Dataset, DataLoader
import numpy as np

# ── Dataset personnalisé ─────────────────────────────
class EauDataset(Dataset):
    def __init__(self, X, y):
        self.X = torch.FloatTensor(X)
        self.y = torch.LongTensor(y)
    def __len__(self):  return len(self.y)
    def __getitem__(self, i): return self.X[i], self.y[i]

train_ds = EauDataset(X_train, y_train)
val_ds   = EauDataset(X_val,   y_val)
train_dl = DataLoader(train_ds, batch_size=64, shuffle=True,
                      num_workers=2, pin_memory=True)
val_dl   = DataLoader(val_ds,  batch_size=128)

# ── Modèle nn.Module ─────────────────────────────────
class EauNet(nn.Module):
    def __init__(self, n_in, couches=[128,64,32], n_out=2):
        super().__init__()
        blocs, n = [], n_in
        for h in couches:
            blocs += [nn.Linear(n, h), nn.BatchNorm1d(h),
                      nn.ReLU(inplace=True), nn.Dropout(0.3)]
            n = h
        blocs.append(nn.Linear(n, n_out))
        self.net = nn.Sequential(*blocs)
    def forward(self, x): return self.net(x)

device = 'cuda' if torch.cuda.is_available() else 'cpu'
model  = EauNet(n_in=8).to(device)

# ── Boucle d'entraînement complète ───────────────────
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3,
                               weight_decay=1e-4)
scheduler = torch.optim.lr_scheduler.OneCycleLR(
    optimizer, max_lr=1e-2,
    steps_per_epoch=len(train_dl), epochs=50)
criterion = nn.CrossEntropyLoss()

best_val_acc, best_state = 0, None
for epoch in range(50):
    # ─ train ─
    model.train()
    for X_b, y_b in train_dl:
        X_b, y_b = X_b.to(device), y_b.to(device)
        optimizer.zero_grad()
        loss = criterion(model(X_b), y_b)
        loss.backward()
        nn.utils.clip_grad_norm_(model.parameters(), 1.0)
        optimizer.step()
        scheduler.step()
    # ─ validation ─
    model.eval()
    correct = 0
    with torch.no_grad():
        for X_b, y_b in val_dl:
            X_b, y_b = X_b.to(device), y_b.to(device)
            correct += (model(X_b).argmax(1)==y_b).sum().item()
    val_acc = correct / len(val_ds)
    if val_acc > best_val_acc:
        best_val_acc = val_acc
        best_state   = {k:v.clone() for k,v in model.state_dict().items()}
    if (epoch+1) % 10 == 0:
        print(f"Epoch {epoch+1:3d} | Val Acc: {val_acc:.3f}")

model.load_state_dict(best_state)  # restaurer le meilleur modèle
print(f"Meilleure val acc : {best_val_acc:.3f}")

import torch.nn as nn

class PlanteCNN(nn.Module):
    """CNN pour classification de 5 maladies de plantes."""
    def __init__(self, n_classes=5, img_size=64):
        super().__init__()

        # ── Blocs convolutifs (feature extraction) ────────
        self.features = nn.Sequential(
            # Bloc 1 : 3 → 32 filtres | 64×64 → 32×32
            nn.Conv2d(3, 32, 3, padding=1), nn.BatchNorm2d(32),
            nn.ReLU(inplace=True),
            nn.Conv2d(32, 32, 3, padding=1), nn.BatchNorm2d(32),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2, 2), nn.Dropout2d(0.25),

            # Bloc 2 : 32 → 64 filtres | 32×32 → 16×16
            nn.Conv2d(32, 64, 3, padding=1), nn.BatchNorm2d(64),
            nn.ReLU(inplace=True),
            nn.Conv2d(64, 64, 3, padding=1), nn.BatchNorm2d(64),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2, 2), nn.Dropout2d(0.25),

            # Bloc 3 : 64 → 128 filtres | 16×16 → 8×8
            nn.Conv2d(64, 128, 3, padding=1), nn.BatchNorm2d(128),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2, 2), nn.Dropout2d(0.25),
        )

        # ── Tête de classification ────────────────────────
        self.classifier = nn.Sequential(
            nn.AdaptiveAvgPool2d(1),  # Global Average Pooling
            nn.Flatten(),
            nn.Linear(128, 256), nn.ReLU(), nn.Dropout(0.5),
            nn.Linear(256, n_classes)
        )

    def forward(self, x):
        return self.classifier(self.features(x))

# Vérification des dimensions
model  = PlanteCNN(n_classes=5)
sample = torch.randn(8, 3, 64, 64)   # batch de 8 images
output = model(sample)
print(f"Sortie : {output.shape}")     # [8, 5]

import torch, torch.nn as nn
from torchvision import models
from torchvision.models import ResNet50_Weights

CLASSES    = ['manioc','maïs','coton','sorgho','arachide']
N_CLASSES  = len(CLASSES)

# ── 1. Charger le backbone pré-entraîné ──────────────
backbone = models.resnet50(weights=ResNet50_Weights.IMAGENET1K_V2)

# ── 2. Geler toutes les couches ───────────────────────
for p in backbone.parameters():
    p.requires_grad = False

# ── 3. Remplacer la tête (fc) ─────────────────────────
n_features    = backbone.fc.in_features  # 2048 pour ResNet50
backbone.fc   = nn.Sequential(
    nn.Linear(n_features, 512),
    nn.BatchNorm1d(512),
    nn.ReLU(inplace=True),
    nn.Dropout(0.5),
    nn.Linear(512, 256),
    nn.ReLU(inplace=True),
    nn.Dropout(0.3),
    nn.Linear(256, N_CLASSES)
)

# ── 4. Phase 1 : Feature Extraction (5 epochs) ────────
optimizer_ft = torch.optim.Adam(
    backbone.fc.parameters(), lr=1e-3)

# ── 5. Phase 2 : Dégeler les 2 derniers blocs ─────────
for name, p in backbone.named_parameters():
    if 'layer4' in name or 'layer3' in name:
        p.requires_grad = True

optimizer_full = torch.optim.Adam([
    {'params': backbone.fc.parameters(),    'lr': 1e-3},
    {'params': backbone.layer4.parameters(),'lr': 1e-5},
    {'params': backbone.layer3.parameters(),'lr': 1e-6},
])
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
    optimizer_full, T_max=20)

# ── 6. Grad-CAM — Visualiser ce que regarde le modèle ─
def grad_cam(model, img_tensor, target_layer):
    """Carte de chaleur sur la feature map de target_layer."""
    gradients, activations = [], []
    target_layer.register_forward_hook(
        lambda m,i,o: activations.append(o))
    target_layer.register_full_backward_hook(
        lambda m,gi,go: gradients.append(go[0]))
    out  = model(img_tensor.unsqueeze(0))
    pred = out.argmax(1)
    model.zero_grad()
    out[0, pred].backward()
    grad = gradients[0].mean(dim=[2,3], keepdim=True)
    cam  = (grad * activations[0]).sum(1).squeeze()
    cam  = torch.relu(cam)
    return (cam / cam.max()).detach().numpy()

import mlflow, mlflow.pytorch
import torch

mlflow.set_experiment("Teranga-AI-PlantesClassif")

with mlflow.start_run(run_name="ResNet50-finetune-v2"):
    # Log hyperparamètres
    mlflow.log_params({
        "model":          "resnet50",
        "lr":             1e-4,
        "batch_size":     32,
        "n_epochs":       30,
        "augmentation":   True,
        "optimizer":      "AdamW",
    })
    # Entraînement ...
    best_acc, best_f1 = entraîner(model, ...)

    # Log métriques
    mlflow.log_metrics({
        "val_accuracy": best_acc,
        "val_f1_macro": best_f1,
    })
    # Log du modèle
    mlflow.pytorch.log_model(
        model, "model",
        registered_model_name="TenaragaPlanteCNN"
    )
    print(f"Acc={best_acc:.3f} | F1={best_f1:.3f}")

from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import torch
from PIL import Image
import io, time

app = FastAPI(
    title="Teranga AI — API Agricole",
    description="Détection de maladies + analyse de sentiment",
    version="1.0.0"
)
app.add_middleware(CORSMiddleware, allow_origins=["*"])

# ── Chargement au démarrage ───────────────────────────
@app.on_event("startup")
async def charger_modeles():
    app.state.cnn = torch.jit.load("plantes_scripted.pt")
    app.state.cnn.eval()
    app.state.nlp = pipeline("text-classification",
                             model="./camembert_alerte")

# ── Endpoint : classification de plantes ──────────────
@app.post("/predict/plante")
async def predire_maladie(file: UploadFile = File(...)):
    if file.content_type not in ["image/jpeg","image/png"]:
        raise HTTPException(400, "Format non supporté")
    t0     = time.time()
    img    = Image.open(io.BytesIO(await file.read())).convert("RGB")
    tensor = val_transform(img).unsqueeze(0)
    with torch.no_grad():
        probs = torch.softmax(app.state.cnn(tensor), 1)[0]
    pred_idx = probs.argmax().item()
    return {
        "maladie":    CLASSES[pred_idx],
        "confiance":  round(float(probs[pred_idx])*100, 1),
        "traitement": TRAITEMENTS[CLASSES[pred_idx]],
        "duree_ms":   round((time.time()-t0)*1000, 1)
    }

# ── Endpoint : analyse d'alerte agricole ──────────────
class AlerteRequest(BaseModel):
    message: str

@app.post("/predict/alerte")
async def analyser_alerte(req: AlerteRequest):
    result = app.state.nlp(req.message)[0]
    return {"type": result["label"], "score": round(result["score"],3)}

# Démarrage : uvicorn api_teranga:app --host 0.0.0.0 --port 8000

import streamlit as st
import torch
from PIL import Image
from transformers import pipeline
import plotly.graph_objects as go

st.set_page_config(page_title="🌿 Teranga AI", layout="wide",
                   page_icon="🌿")

# ── Chargement des modèles ────────────────────────────
@st.cache_resource
def charger_modeles():
    cnn = torch.jit.load("plantes_scripted.pt")
    cnn.eval()
    nlp = pipeline("text-classification",
                   model="./camembert_alerte")
    return cnn, nlp

cnn_model, nlp_model = charger_modeles()

# ── Sidebar navigation ────────────────────────────────
st.sidebar.image("logo_teranga.png", width=120)
module = st.sidebar.radio("🗂️ Module", [
    "🌿 Diagnostic Plantes",
    "📢 Analyse d'Alertes",
    "📊 Prix des Céréales"
])

CLASSES = ['Saine','Mildiou manioc','Rouille maïs',
           'Anthracnose manguier','Pourriture cacao']
TRAITEMENTS = {
    'Saine':                "Aucun traitement nécessaire ✅",
    'Mildiou manioc':       "Fongicide à base de cuivre — 2500 XOF/ha",
    'Rouille maïs':         "Propiconazole 250 EC — 3200 XOF/ha",
    'Anthracnose manguier': "Thirame + Manèbe — 4100 XOF/ha",
    'Pourriture cacao':     "Cuivre oxychloride — 5500 XOF/ha",
}

# ── Module 1 : Diagnostic ─────────────────────────────
if module == "🌿 Diagnostic Plantes":
    st.title("🔬 Diagnostic de Maladies des Plantes")
    col1, col2 = st.columns([1, 1.2])
    with col1:
        photo = st.file_uploader("📷 Photo de la feuille",
                                  type=["jpg","jpeg","png"])
        if photo:
            img    = Image.open(photo).convert("RGB")
            tensor = val_transform(img).unsqueeze(0)
            with torch.no_grad():
                probs = torch.softmax(cnn_model(tensor), 1)[0]
            pred_cls  = CLASSES[probs.argmax().item()]
            confiance = float(probs.max()) * 100
    with col2:
        if photo:
            st.image(img, caption="Image analysée", width=280)
            st.metric("🔬 Diagnostic", pred_cls,
                      f"{confiance:.1f}% de confiance")
            st.info(f"💊 {TRAITEMENTS[pred_cls]}")
            # Graphique de probabilités
            fig = go.Figure(go.Bar(
                x=CLASSES, y=probs.tolist(),
                marker_color=['#2f7d5c' if c==pred_cls
                              else '#bcd3c4' for c in CLASSES]))
            fig.update_layout(title="Probabilités par classe",
                             height=300)
            st.plotly_chart(fig, use_container_width=True)