Enedis Data Connect :
intégrer les courbes de charge Linky
dans vos études PV

import requests
import urllib.parse
import secrets
import base64

# ── Configuration ─────────────────────────────────────────────────────────────
ENEDIS_AUTH_URL   = "https://mon-compte-particulier.enedis.fr/dataconnect/v1/oauth2/authorize"
ENEDIS_TOKEN_URL  = "https://mon-compte-particulier.enedis.fr/dataconnect/v1/oauth2/token"
ENEDIS_API_BASE   = "https://ext.api.lincsesdc.cdc.enedis.fr/sg"

CLIENT_ID     = "votre_client_id"      # À remplacer par votre identifiant
CLIENT_SECRET = "votre_client_secret"  # À stocker en variable d'environnement !
REDIRECT_URI  = "https://votre-app.fr/callback"


def construire_url_autorisation(usage_point_id: str = None) -> tuple[str, str]:
    """
    Génère l'URL de redirection vers Enedis pour le consentement.
    Retourne (url, state) – conserver le state pour validation CSRF.
    """
    state = secrets.token_urlsafe(32)

    params = {
        "client_id":     CLIENT_ID,
        "response_type": "code",
        "redirect_uri":  REDIRECT_URI,
        "state":         state,
        "duration":      "P3Y",  # Durée du consentement : 3 ans (ISO 8601)
    }
    if usage_point_id:
        # Pré-remplir le numéro de compteur si connu
        params["usage_point_id"] = usage_point_id

    url = ENEDIS_AUTH_URL + "?" + urllib.parse.urlencode(params)
    return url, state


def echanger_code_contre_token(code: str) -> dict:
    """
    Échange le code d'autorisation (callback) contre un access_token.
    Utilise l'authentification Basic (client_id:client_secret en Base64).
    """
    credentials = f"{CLIENT_ID}:{CLIENT_SECRET}"
    b64 = base64.b64encode(credentials.encode()).decode()

    headers = {
        "Authorization": f"Basic {b64}",
        "Content-Type":  "application/x-www-form-urlencoded",
    }
    body = {
        "grant_type":   "authorization_code",
        "code":         code,
        "redirect_uri": REDIRECT_URI,
    }

    resp = requests.post(ENEDIS_TOKEN_URL, headers=headers, data=body, timeout=15)
    resp.raise_for_status()
    return resp.json()
    # Retourne : {'access_token': ..., 'refresh_token': ..., 'expires_in': 3600, ...}

from datetime import date, timedelta
import pandas as pd

def recuperer_courbe_charge(
    access_token: str,
    usage_point_id: str,
    date_debut: date,
    date_fin: date,
) -> pd.DataFrame:
    """
    Récupère la courbe de charge de consommation (intervalles 30 min).
    
    Returns:
        DataFrame avec colonnes : timestamp (UTC), wh (énergie Wh par pas),
                                  w_moyen (puissance moyenne W par pas)
    """
    url = f"{ENEDIS_API_BASE}/metering_data/consumption_load_curve/v5"
    params = {
        "usage_point_id": usage_point_id,
        "start": date_debut.isoformat(),
        "end":   date_fin.isoformat(),
    }
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Accept": "application/json",
    }

    resp = requests.get(url, params=params, headers=headers, timeout=30)
    resp.raise_for_status()
    data = resp.json()

    # Parsing de la réponse Enedis
    intervals = (data
        .get("meter_reading", {})
        .get("interval_reading"span>, [])
    )

    rows = []
    for iv in intervals:
        rows.append({
            "timestamp": pd.to_datetime(iv["date"], utc=True),
            "wh":        float(iv["value"]),
        })

    df = pd.DataFrame(rows).set_index("timestamp").sort_index()
    df["w_moyen"] = df["wh"] * 2     # Wh/30min → W moyen (×2 car ½ heure)
    return df

import numpy as np
import pandas as pd


def calculer_autoconsommation(
    df_consommation: pd.DataFrame,  # courbe Enedis (colonne 'wh', index UTC 30 min)
    df_production: pd.DataFrame,    # production PVGIS ou simulée (colonne 'wh', index UTC 30 min)
) -> dict:
    """
    Croise production PV et consommation réelle pour calculer :
    - Taux d'autoconsommation (% production autoconsommée)
    - Taux d'autosuffisance (% consommation couverte par PV)
    - Surplus injecté (kWh/an)
    - Économies annuelles estimées (€)
    """
    # Aligner les deux séries sur le même index temporel
    df = pd.DataFrame({
        "conso_wh": df_consommation["wh"],
        "prod_wh":  df_production["wh"],
    }).dropna()

    # Pour chaque pas de temps, autoconsommé = min(prod, conso)
    df["autoconso_wh"] = np.minimum(df["prod_wh"], df["conso_wh"])
    df["surplus_wh"]   = (df["prod_wh"] - df["conso_wh"]).clip(lower=0)

    total_prod   = df["prod_wh"].sum()      # Wh
    total_conso  = df["conso_wh"].sum()
    total_auto   = df["autoconso_wh"].sum()
    total_surplus = df["surplus_wh"].sum()

    taux_autoconsommation = (total_auto / total_prod * 100) if total_prod > 0 else 0
    taux_autosuffisance   = (total_auto / total_conso * 100) if total_conso > 0 else 0

    return {
        "production_kwh_an":          round(total_prod / 1000, 0),
        "consommation_kwh_an":        round(total_conso / 1000, 0),
        "autoconsommation_kwh_an":    round(total_auto / 1000, 0),
        "surplus_injecte_kwh_an":     round(total_surplus / 1000, 0),
        "taux_autoconsommation_pct":  round(taux_autoconsommation, 1),
        "taux_autosuffisance_pct":    round(taux_autosuffisance, 1),
    }