Logistique (`kadi.market.logistics`)¶

La classe Logistics modélise les frictions logistiques réelles sur les corridors commerciaux béninois : coût de transport, tracasseries aux postes de contrôle et dégradation de la qualité marchande des produits.

Depuis la v1.2.0, elle intègre optionnellement la météo pour ajuster dynamiquement le coefficient de route (gamma_route) et la perte de qualité selon la probabilité de pluie prévue.

Elle est utilisée en interne par Market et accessible via market.logistics. Elle peut aussi être instanciée directement.

Importation directe¶

from kadi.market import Logistics

Initialisation¶

from kadi.market import Logistics

# Sans intégration météo (comportement V1)
logistics = Logistics()

# Avec intégration météo (ajustement dynamique du coût selon la pluie)
import kadi as kd

weather = kd.Weather(lat=9.3, lon=2.3, name="Parakou")
logistics = Logistics(weather=weather)

Paramètres :

Nom	Type	Défaut	Description
`cache_file`	`str`	`None`	Chemin vers le fichier de cache JSON des distances. Par défaut : `~/.kadi/osrm_cache.json`
`weather`	`Weather`	`None`	Session météo pour l'ajustement climatique des coûts. Si None, aucun ajustement (comportement V1)

Attributs publics :

Attribut	Type	Description
`weather`	`Weather`	Session météo injectée (ou None)
`cache_file`	`str`	Chemin du fichier de cache des distances
`cache`	`dict`	Cache en mémoire : clés `'coords'` et `'distances'`

Méthodes¶

`transfer_cost(origine, destination, prix_carburant, crop)`¶

Calcule le coût total de transfert d'une ville à une autre.

Formule :

C_transfer = C_info
           + Distance * (gamma_effectif * P_carburant / 100 + mu_checkpoints)
           + C_qualite(culture, distance, pluie)

Où : - C_info = coût fixe de recherche d'information (appels, déplacements), configurable dans config.py (défaut : 5 000 XOF) - gamma_effectif = gamma_route * (1 + alpha_pluie * prob_pluie) : majoration météo si une session weather est disponible - mu_checkpoints = coût moyen des tracasseries par km (défaut : 15 XOF/km) - C_qualite = perte de valeur marchande variable par culture et météo

import kadi as kd

weather = kd.Weather(lat=9.3, lon=2.3, name="Parakou")
marche = kd.Market(lat=9.3, lon=2.3, location="Parakou", weather=weather)

# Coût de transfert avec intégration météo
cout = marche.logistics.transfer_cost(
    origine="Parakou",
    destination="Cotonou",
    crop="maize",
)

print(f"Coût total       : {cout['total_cost_cfa']:,.0f} XOF")
print(f"Pluie prévue     : {cout['prob_pluie'] * 100:.0f} %")
print(f"Gamma effectif   : {cout['gamma_effectif']:.4f}")

# Détail par poste
d = cout["details"]
print(f"  Distance       : {d['distance_km']:.1f} km")
print(f"  Coût info      : {d['search_costs']:,.0f} XOF")
print(f"  Transport      : {d['transport_costs']:,.0f} XOF")
print(f"  Perte qualité  : {d['quality_loss']:,.0f} XOF")
print(f"  Carburant      : {d['fuel_price_used']:.0f} XOF/litre")

Paramètres :

Nom	Type	Défaut	Description
`origine`	`str`	requis	Ville de départ (ex: `'Parakou'`)
`destination`	`str`	requis	Ville d'arrivée (ex: `'Cotonou'`)
`prix_carburant`	`float`	`None`	Prix du litre d'essence en XOF. Si None, récupéré automatiquement
`crop`	`str`	`None`	Culture transportée, pour la perte de qualité. Si None, utilise le facteur par défaut

Retour : dict

Clé	Type	Description
`total_cost_cfa`	`float`	Cout total du transfert en XOF
`prob_pluie`	`float`	Probabilite de pluie utilisee (0.0 si pas de météo)
`gamma_effectif`	`float`	Coefficient de route effectivement applique
`details`	`dict`	Sous-dictionnaire avec chaque composante du cout

Clés de details :

Clé	Description
`distance_km`	Distance routiere en km
`search_costs`	Cout fixe de recherche d'information (XOF)
`transport_costs`	Cout lié à la distance (carburant + tracasseries) (XOF)
`quality_loss`	Perte de valeur marchande (XOF)
`fuel_price_used`	Prix du carburant utilise (XOF/litre)
`gamma_route_base`	Coefficient de base avant ajustement météo
`gamma_effectif`	Coefficient apres ajustement météo
`prob_pluie`	Probabilite de pluie utilisee
`crop`	Culture concernee (`'_default'` si non spécifiée)

Perte de qualité par culture (XOF/km/tonne) :

Culture	Facteur	Culture	Facteur
`maize`, `sorghum`, `millet`	5.0	`cowpea`, `soybean`	7.0
`rice`	6.0	`yam`	12.0
`cassava`	10.0	`tomato`	25.0
`onion`	20.0	(défaut)	8.0

Sous la pluie, la perte de qualité est majorée : C_qualite = facteur * distance_km * (1 + beta_pluie * prob_pluie), avec beta_pluie configurable dans config.py (défaut : 0.5).

`distance(origine, destination)`¶

Récupère la distance routière entre deux villes béninoises.

Stratégie en cascade :

Cache local (résultat d'un appel précédent, persisté dans le fichier JSON)
Géocodage Nominatim (OpenStreetMap) + routage OSRM
Fallback Haversine (vol d'oiseau x 1.3) si OSRM est indisponible
Valeur de repli 100 km si le géocodage échoue

d = marche.logistics.distance("Parakou", "Cotonou")
print(f"Distance routière : {d:.1f} km")

# Résultat mis en cache automatiquement
d2 = marche.logistics.distance("Cotonou", "Parakou")   # Retourné depuis le cache

Paramètres :

Nom	Type	Description
`origine`	`str`	Nom de la ville de départ
`destination`	`str`	Nom de la ville d'arrivée

Retour : float - Distance estimée en kilomètres.

Le résultat est automatiquement mis en cache pour la session en cours et sauvegardé dans cache_file pour les sessions ultérieures.

Intégration météo¶

Quand une instance weather est fournie à l'initialisation, le calcul du coût de transfert est ajusté automatiquement :

gamma_effectif = gamma_route * (1 + alpha_pluie * prob_pluie)

gamma_route : coefficient de base (configurable dans config.py, défaut : 1.2)
alpha_pluie : intensité de la majoration météo (défaut : 0.25)
prob_pluie : probabilité de pluie demain, obtenue depuis weather.rain_prob(days=1)

La probabilité de pluie est calculée une seule fois par session (mise en cache) pour éviter des appels répétés au module météo.

Exemple :

Avec prob_pluie=0.8 et alpha=0.25, le coefficient de route passe de 1.2 à 1.2 * (1 + 0.25 * 0.8) = 1.44, soit une majoration de 20% sur le coût de transport.

Prix du carburant¶

La méthode transfer_cost() récupère automatiquement le prix du carburant selon la stratégie suivante :

Priorité	Source
1	Variable d'environnement `BENIN_FUEL_PRICE`
2	Cache en mémoire (une seule requête par session)
3	Fichier `config/fuel_prices.json` sur GitHub
4	Valeur de repli depuis `config.py` (défaut : 680 XOF/litre)

Méthodes dépréciées¶

Ancienne méthode	Remplacée par	Depuis
`calculate_transfer_cost(origine, destination)`	`transfer_cost(origine, destination)`	v1.2.0
`get_distance(origine, destination)`	`distance(origine, destination)`	v1.2.0

Classe dépréciée¶

Ancien nom	Remplacé par	Depuis
`MarketLogistics`	`Logistics`	v1.2.0

Exemple complet¶

import kadi as kd

weather = kd.Weather(lat=9.3, lon=2.3, name="Parakou")
marche = kd.Market(lat=9.3, lon=2.3, location="Parakou", weather=weather)

# Comparer le coût pour deux cultures (tomate plus sensible)
cultures = ["maize", "tomato"]
for crop in cultures:
    cout = marche.logistics.transfer_cost("Parakou", "Cotonou", crop=crop)
    print(
        f"{crop:8s} -> {cout['total_cost_cfa']:,.0f} XOF "
        f"(perte qualité : {cout['details']['quality_loss']:,.0f} XOF)"
    )

# Distance seule
d = marche.logistics.distance("Abomey", "Cotonou")
print(f"Abomey - Cotonou : {d:.1f} km")

Logistics ¶

Classe modélisant les frictions logistiques réelles au Bénin : coûts de transport, tracasseries aux postes de contrôle, et dégradation de la qualité des marchandises.

Phase 4 - Intégration météo

Si une session weather est fourni à l'initialisation, le calcul des coûts de transfert utilise la probabilité de pluie du lendemain pour ajuster gamma_route (état des routes) et la perte de qualité. En l'absence de la session weather, le comportement est identique à la V1.

Source code in kadi/market/logistics.py

class Logistics:
    """
    Classe modélisant les frictions logistiques réelles au Bénin :
    coûts de transport, tracasseries aux postes de contrôle,
    et dégradation de la qualité des marchandises.

    Phase 4 - Intégration météo :
        Si une session weather est fourni à l'initialisation, le calcul
        des coûts de transfert utilise la probabilité de pluie du lendemain
        pour ajuster gamma_route (état des routes) et la perte de qualité.
        En l'absence de la session weather, le comportement est identique à la V1.
    """

    def __init__(self, cache_file: str = None, weather=None):
        """
        Initialise le module logistique.

        Charge le cache persistant des coordonnées GPS et des distances
        routières pour éviter des appels réseau redondants.

        Args:
            cache_file (str, optional): Chemin vers le fichier de cache JSON.
                Si None, utilise ``~/.kadi/osrm_cache.json``.
            weather (Weather, optional): Session météo
                (kadi.weather.Weather) pour ajuster les coûts selon
                la météo prévue. Si None, pas d'ajustement climatique.
        """
        # Session météo optionnelle (intégration Phase 4)
        self.weather = weather

        # Initialisation du chemin du fichier de cache
        if cache_file is None:
            cache_dir = os.path.expanduser("~/.kadi")
            try:
                os.makedirs(cache_dir, exist_ok=True)
                self.cache_file = os.path.join(cache_dir, "osrm_cache.json")
            except OSError:
                import tempfile
                cache_dir = tempfile.gettempdir()
                self.cache_file = os.path.join(cache_dir, "osrm_cache.json")
        else:
            self.cache_file = cache_file


        # Structure du cache en mémoire (coordonnées et distances routières)
        self.cache = {"coords": {}, "distances": {}}

        # Cache en mémoire pour le prix du carburant (évite les appels répétés)
        self._cached_fuel_price: Optional[float] = None

        # Cache en mémoire pour la probabilité de pluie (une seule requête par session)
        self._cached_prob_pluie: Optional[float] = None

        # Chargement du cache depuis le disque
        self._load_cache()

    # ------------------------------------------------------------------
    # Rétrocompatibilité : méthodes publiques renommées
    # ------------------------------------------------------------------

    def __getattr__(self, name: str):
        """Intercepte les accès aux anciens noms de méthodes publiques.

        Délègue vers le nouveau nom et émet un DeprecationWarning.

        Args:
            name (str): Nom de l'attribut ou méthode demandé.

        Returns:
            callable: La méthode correspondante sous son nouveau nom.

        Raises:
            AttributeError: Si le nom n'est ni nouveau ni ancien.
        """
        # Vérification dans la table de rétrocompatibilité
        if name in _DEPRECATED_METHODS:
            new_name = _DEPRECATED_METHODS[name]
            warnings.warn(
                f"Logistics.{name}() est obsolète et sera supprimé dans "
                f"KadiPy v2.0. Utilisez Logistics.{new_name}() à la place.",
                DeprecationWarning,
                stacklevel=2,
            )
            return getattr(self, new_name)
        raise AttributeError(f"'Logistics' n'a pas d'attribut '{name}'.")

    def _load_cache(self):
        """Charge le cache persistant depuis le fichier JSON s'il existe."""
        if os.path.isfile(self.cache_file):
            try:
                with open(self.cache_file, "r", encoding="utf-8") as f:
                    data = json.load(f)
                    self.cache["coords"] = data.get("coords", {})
                    self.cache["distances"] = data.get("distances", {})
            except Exception as e:
                logger.warning(f"Impossible de lire le cache {self.cache_file}: {e}")

    def _save_cache(self):
        """Sauvegarde le cache en mémoire dans le fichier JSON persistant."""
        try:
            with open(self.cache_file, "w", encoding="utf-8") as f:
                json.dump(self.cache, f, indent=4)
        except Exception as e:
            logger.warning(f"Impossible de sauvegarder le cache {self.cache_file}: {e}")

    def _get_prob_pluie(self) -> float:
        """
        Retourne la probabilité de pluie, mise en cache pour la durée de la session.

        La probabilité est calculée une seule fois par session logistique pour
        éviter des appels répétés au module météo.

        Returns:
            float: Probabilité de pluie entre 0.0 et 1.0.
        """
        if self._cached_prob_pluie is None:
            self._cached_prob_pluie = _rain_prob(self.weather)
        return self._cached_prob_pluie

    def _haversine_distance(
        self, lon1: float, lat1: float, lon2: float, lat2: float
    ) -> float:
        """
        Calcule la distance orthodromique entre deux points GPS
        en utilisant la formule de Haversine.

        Utilisée comme fallback si OSRM est indisponible. La distance
        vol d'oiseau est multipliée par un facteur de tortuosité (1.3)
        pour approximer la distance réelle sur les routes béninoises.

        Args:
            lon1 (float): Longitude du point de départ.
            lat1 (float): Latitude du point de départ.
            lon2 (float): Longitude du point d'arrivée.
            lat2 (float): Latitude du point d'arrivée.

        Returns:
            float: Distance routière approximée en kilomètres.
        """
        # Rayon de la Terre en kilomètres
        R = 6371.0

        # Conversion des coordonnées en radians
        lat1_r = math.radians(lat1)
        lon1_r = math.radians(lon1)
        lat2_r = math.radians(lat2)
        lon2_r = math.radians(lon2)

        dlon = lon2_r - lon1_r
        dlat = lat2_r - lat1_r

        # Formule de Haversine
        a = (
            math.sin(dlat / 2) ** 2
            + math.cos(lat1_r) * math.cos(lat2_r) * math.sin(dlon / 2) ** 2
        )
        c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))

        distance_vol_oiseau = R * c

        # Facteur de tortuosité : les routes béninoises font environ 30%
        # de plus que la distance à vol d'oiseau en moyenne
        return distance_vol_oiseau * 1.3

    def _geocode_city(self, city_name: str) -> Optional[Tuple[float, float]]:
        """
        Interroge Nominatim (OpenStreetMap) pour obtenir les coordonnées GPS
        d'une ville. Respecte la limite de taux de 1 requête/seconde.

        Le résultat est mis en cache pour éviter des appels répétés.

        Args:
            city_name (str): Nom de la ville à géocoder (ex: 'Cotonou').

        Returns:
            tuple: (longitude, latitude) ou None si la ville est introuvable.
        """
        city_key = city_name.strip().lower()

        # Retour depuis le cache si déjà résolu
        if city_key in self.cache["coords"]:
            return tuple(self.cache["coords"][city_key])

        # Respect du rate-limit Nominatim avant toute requête
        _respect_nominatim_rate_limit()

        url = "https://nominatim.openstreetmap.org/search"
        params = {
            "q": f"{city_name}, Benin",
            "format": "json",
            "limit": 1,
        }
        # Identification de l'application (requis par les conditions d'utilisation Nominatim)
        headers = {"User-Agent": "KadiPy/0.1.0 (Agritech Research, Benin)"}

        try:
            response = requests.get(url, params=params, headers=headers, timeout=5)
            response.raise_for_status()
            data = response.json()

            if data and len(data) > 0:
                lon = float(data[0]["lon"])
                lat = float(data[0]["lat"])
                # Mise en cache du résultat
                self.cache["coords"][city_key] = [lon, lat]
                self._save_cache()
                return (lon, lat)
            else:
                logger.warning(f"Ville '{city_name}' introuvable via Nominatim.")

        except Exception as e:
            logger.warning(f"Erreur lors du géocodage de '{city_name}': {e}")

        return None

    def _fetch_osrm_distance(
        self, lon1: float, lat1: float, lon2: float, lat2: float
    ) -> Optional[float]:
        """
        Interroge le serveur public OSRM pour calculer la distance routière
        réelle entre deux points GPS.

        Args:
            lon1 (float): Longitude du point de départ.
            lat1 (float): Latitude du point de départ.
            lon2 (float): Longitude du point d'arrivée.
            lat2 (float): Latitude du point d'arrivée.

        Returns:
            float: Distance en kilomètres, ou None si la requête a échoué.
        """
        url = (
            f"http://router.project-osrm.org/route/v1/driving/"
            f"{lon1},{lat1};{lon2},{lat2}?overview=false"
        )
        try:
            response = requests.get(url, timeout=5)
            response.raise_for_status()
            data = response.json()

            if data.get("code") == "Ok" and data.get("routes"):
                distance_metres = data["routes"][0]["distance"]
                return distance_metres / 1000.0

        except Exception as e:
            logger.warning(f"Erreur lors de la requête OSRM : {e}")

        return None

    def distance(self, origine: str, destination: str) -> float:
        """
        Récupère la distance routière réelle entre deux villes béninoises.

        Stratégie en cascade :
        1. Cache local (résultat d'un appel précédent)
        2. Géocodage Nominatim + routage OSRM
        3. Fallback Haversine (vol d'oiseau * 1.3) si OSRM échoue
        4. Valeur par défaut 100 km si le géocodage échoue

        Args:
            origine (str): Nom de la ville de départ (ex: 'Parakou').
            destination (str): Nom de la ville d'arrivée (ex: 'Cotonou').

        Returns:
            float: Distance estimée en kilomètres.
        """
        orig_key = origine.strip().lower()
        dest_key = destination.strip().lower()
        route_key = f"{orig_key}_{dest_key}"
        rev_route_key = f"{dest_key}_{orig_key}"

        # 1. Vérification du cache (aller et retour)
        if route_key in self.cache["distances"]:
            return self.cache["distances"][route_key]
        if rev_route_key in self.cache["distances"]:
            return self.cache["distances"][rev_route_key]

        # 2. Géocodage des deux villes
        coords_orig = self._geocode_city(origine)
        coords_dest = self._geocode_city(destination)

        if not coords_orig or not coords_dest:
            logger.warning(
                f"Géocodage impossible pour '{origine}' -> '{destination}'. "
                "Valeur de repli : 100 km."
            )
            return 100.0

        # 3. Requête OSRM pour la vraie distance routière
        dist = self._fetch_osrm_distance(
            coords_orig[0], coords_orig[1],
            coords_dest[0], coords_dest[1],
        )

        # 4. Fallback Haversine si OSRM est indisponible
        if dist is None:
            logger.warning(
                f"OSRM indisponible pour '{origine}' -> '{destination}'. "
                "Utilisation de la formule Haversine (distance approximée)."
            )
            dist = self._haversine_distance(
                coords_orig[0], coords_orig[1],
                coords_dest[0], coords_dest[1],
            )

        # Mise en cache du résultat
        self.cache["distances"][route_key] = dist
        self._save_cache()

        return dist

    def _fetch_fuel_price(self) -> float:
        """
        Récupère le prix actuel du carburant (essence) au Bénin.

        Stratégie en cascade :
        1. Variable d'environnement BENIN_FUEL_PRICE (priorité absolue)
        2. Cache en mémoire (évite les appels répétés)
        3. Fichier de configuration sur GitHub (config/fuel_prices.json)
        4. Valeur de repli depuis la configuration (680 XOF/litre par défaut)

        Returns:
            float: Prix de l'essence en XOF par litre.
        """
        # Priorité 1 : variable d'environnement
        env_price = os.getenv("BENIN_FUEL_PRICE")
        if env_price is not None:
            try:
                return float(env_price)
            except ValueError:
                logger.warning(
                    f"Valeur BENIN_FUEL_PRICE invalide dans .env : '{env_price}'. "
                    "Passage à la source suivante."
                )

        # Priorité 2 : cache en mémoire
        if self._cached_fuel_price is not None:
            return self._cached_fuel_price

        # Priorité 3 : configuration en ligne (GitHub)
        url = "https://raw.githubusercontent.com/delsDin/kadipy/main/config/fuel_prices.json"
        try:
            response = requests.get(url, timeout=5)
            response.raise_for_status()
            data = response.json()
            price = float(data.get("benin", {}).get("essence", None))
            self._cached_fuel_price = price
            return price
        except Exception as e:
            logger.warning(
                f"Impossible de récupérer le prix du carburant en ligne ({e}). "
                "Utilisation de la valeur de repli."
            )

        # Priorité 4 : valeur de repli depuis la configuration
        prix_repli = _LOGISTICS_CONFIG.get("prix_carburant_fallback_xof", 680.0)
        self._cached_fuel_price = prix_repli
        return prix_repli

    def transfer_cost(
        self,
        origine: str,
        destination: str,
        prix_carburant: float = None,
        crop: str = None,
    ) -> dict:
        """
        Calcule le coût total de transfert d'un point A vers un point B.

        Phase 4 - Intégration météo :
            Si une session weather a été fourni à l'initialisation, le calcul
            ajuste automatiquement :
            - gamma_route selon la probabilité de pluie prévue demain.
            - La perte de qualité selon la culture et la météo.

        Formule :
            C_transfer = C_info
                       + Distance * (gamma_effectif * P_carburant / 100 + mu_checkpoints)
                       + C_qualite(culture, distance, pluie)

        Ou :
        - C_info = coût fixe de recherche d'informations (appels, déplacements)
        - gamma_effectif = gamma_route * (1 + alpha_pluie * prob_pluie)
        - P_carburant = prix du litre d'essence en XOF
        - mu_checkpoints = coût moyen des tracasseries par km
        - C_qualite = perte de valeur marchande (variable par culture et météo)

        Args:
            origine (str): Ville de départ (ex: 'Parakou').
            destination (str): Ville de destination (ex: 'Cotonou').
            prix_carburant (float, optional): Prix du litre d'essence en XOF.
                Si None, récupéré automatiquement depuis les sources configurées.
            crop (str, optional): Code de la culture transportée (ex: 'maize').
                Influence la perte de qualité. Si None, utilise le facteur par défaut.

        Returns:
            dict: Dictionnaire contenant le coût total et le détail par poste :
                - 'total_cost_cfa' : coût total du transfert en XOF
                - 'details'        : sous-dictionnaire avec chaque composante
                - 'prob_pluie'     : probabilité de pluie utilisée (0 si pas de météo)
                - 'gamma_effectif' : gamma_route réellement appliqué
        """
        # Récupération du prix du carburant si non fourni
        if prix_carburant is None:
            prix_carburant = self._fetch_fuel_price()

        # Récupération des coefficients depuis la configuration centrale
        c_info = _LOGISTICS_CONFIG.get("c_info_xof", 5000.0)
        gamma_route_base = _LOGISTICS_CONFIG.get("gamma_route", 1.2)
        mu_checkpoints = _LOGISTICS_CONFIG.get("mu_checkpoints_xof_per_km", 15.0)

        # Probabilité de pluie (0.0 si pas de module météo disponible)
        prob_pluie = self._get_prob_pluie()

        # Coefficient de route effectif, ajusté selon la météo
        gamma_effectif = _gamma_effective(gamma_route_base, prob_pluie)

        # Calcul de la distance routière entre les deux villes
        d_ab = self.distance(origine, destination)

        # Calcul du coût lié à la distance (carburant + tracasseries)
        cout_distance = d_ab * ((gamma_effectif * prix_carburant / 100.0) + mu_checkpoints)

        # Perte de qualité dynamique (par culture, distance et météo)
        c_qualite = _quality_loss(crop, d_ab, prob_pluie)

        # Coût de transfert total
        c_transfer = c_info + cout_distance + c_qualite

        # Construction du résultat détaillé
        resultat = {
            "total_cost_cfa": round(c_transfer, 2),
            "prob_pluie": round(prob_pluie, 3),
            "gamma_effectif": round(gamma_effectif, 4),
            "details": {
                "distance_km": round(d_ab, 2),
                "search_costs": c_info,
                "transport_costs": round(cout_distance, 2),
                "quality_loss": c_qualite,
                "fuel_price_used": prix_carburant,
                "gamma_route_base": gamma_route_base,
                "gamma_effectif": round(gamma_effectif, 4),
                "prob_pluie": round(prob_pluie, 3),
                "crop": crop or "_default",
            },
        }

        return resultat

init ¶

__init__(cache_file: str = None, weather=None)

Initialise le module logistique.

Charge le cache persistant des coordonnées GPS et des distances routières pour éviter des appels réseau redondants.

Parameters:

Name	Type	Description	Default
`cache_file`	`str`	Chemin vers le fichier de cache JSON. Si None, utilise `~/.kadi/osrm_cache.json`.	`None`
`weather`	`Weather`	Session météo (kadi.weather.Weather) pour ajuster les coûts selon la météo prévue. Si None, pas d'ajustement climatique.	`None`

Source code in kadi/market/logistics.py

def __init__(self, cache_file: str = None, weather=None):
    """
    Initialise le module logistique.

    Charge le cache persistant des coordonnées GPS et des distances
    routières pour éviter des appels réseau redondants.

    Args:
        cache_file (str, optional): Chemin vers le fichier de cache JSON.
            Si None, utilise ``~/.kadi/osrm_cache.json``.
        weather (Weather, optional): Session météo
            (kadi.weather.Weather) pour ajuster les coûts selon
            la météo prévue. Si None, pas d'ajustement climatique.
    """
    # Session météo optionnelle (intégration Phase 4)
    self.weather = weather

    # Initialisation du chemin du fichier de cache
    if cache_file is None:
        cache_dir = os.path.expanduser("~/.kadi")
        try:
            os.makedirs(cache_dir, exist_ok=True)
            self.cache_file = os.path.join(cache_dir, "osrm_cache.json")
        except OSError:
            import tempfile
            cache_dir = tempfile.gettempdir()
            self.cache_file = os.path.join(cache_dir, "osrm_cache.json")
    else:
        self.cache_file = cache_file


    # Structure du cache en mémoire (coordonnées et distances routières)
    self.cache = {"coords": {}, "distances": {}}

    # Cache en mémoire pour le prix du carburant (évite les appels répétés)
    self._cached_fuel_price: Optional[float] = None

    # Cache en mémoire pour la probabilité de pluie (une seule requête par session)
    self._cached_prob_pluie: Optional[float] = None

    # Chargement du cache depuis le disque
    self._load_cache()

getattr ¶

__getattr__(name: str)

Intercepte les accès aux anciens noms de méthodes publiques.

Délègue vers le nouveau nom et émet un DeprecationWarning.

Parameters:

Name	Type	Description	Default
`name`	`str`	Nom de l'attribut ou méthode demandé.	required

Returns:

Name	Type	Description
`callable`		La méthode correspondante sous son nouveau nom.

Raises:

Type	Description
`AttributeError`	Si le nom n'est ni nouveau ni ancien.

Source code in kadi/market/logistics.py

def __getattr__(self, name: str):
    """Intercepte les accès aux anciens noms de méthodes publiques.

    Délègue vers le nouveau nom et émet un DeprecationWarning.

    Args:
        name (str): Nom de l'attribut ou méthode demandé.

    Returns:
        callable: La méthode correspondante sous son nouveau nom.

    Raises:
        AttributeError: Si le nom n'est ni nouveau ni ancien.
    """
    # Vérification dans la table de rétrocompatibilité
    if name in _DEPRECATED_METHODS:
        new_name = _DEPRECATED_METHODS[name]
        warnings.warn(
            f"Logistics.{name}() est obsolète et sera supprimé dans "
            f"KadiPy v2.0. Utilisez Logistics.{new_name}() à la place.",
            DeprecationWarning,
            stacklevel=2,
        )
        return getattr(self, new_name)
    raise AttributeError(f"'Logistics' n'a pas d'attribut '{name}'.")

distance ¶

distance(origine: str, destination: str) -> float

Récupère la distance routière réelle entre deux villes béninoises.

Stratégie en cascade : 1. Cache local (résultat d'un appel précédent) 2. Géocodage Nominatim + routage OSRM 3. Fallback Haversine (vol d'oiseau * 1.3) si OSRM échoue 4. Valeur par défaut 100 km si le géocodage échoue

Parameters:

Name	Type	Description	Default
`origine`	`str`	Nom de la ville de départ (ex: 'Parakou').	required
`destination`	`str`	Nom de la ville d'arrivée (ex: 'Cotonou').	required

Returns:

Name	Type	Description
`float`	`float`	Distance estimée en kilomètres.

Source code in kadi/market/logistics.py

def distance(self, origine: str, destination: str) -> float:
    """
    Récupère la distance routière réelle entre deux villes béninoises.

    Stratégie en cascade :
    1. Cache local (résultat d'un appel précédent)
    2. Géocodage Nominatim + routage OSRM
    3. Fallback Haversine (vol d'oiseau * 1.3) si OSRM échoue
    4. Valeur par défaut 100 km si le géocodage échoue

    Args:
        origine (str): Nom de la ville de départ (ex: 'Parakou').
        destination (str): Nom de la ville d'arrivée (ex: 'Cotonou').

    Returns:
        float: Distance estimée en kilomètres.
    """
    orig_key = origine.strip().lower()
    dest_key = destination.strip().lower()
    route_key = f"{orig_key}_{dest_key}"
    rev_route_key = f"{dest_key}_{orig_key}"

    # 1. Vérification du cache (aller et retour)
    if route_key in self.cache["distances"]:
        return self.cache["distances"][route_key]
    if rev_route_key in self.cache["distances"]:
        return self.cache["distances"][rev_route_key]

    # 2. Géocodage des deux villes
    coords_orig = self._geocode_city(origine)
    coords_dest = self._geocode_city(destination)

    if not coords_orig or not coords_dest:
        logger.warning(
            f"Géocodage impossible pour '{origine}' -> '{destination}'. "
            "Valeur de repli : 100 km."
        )
        return 100.0

    # 3. Requête OSRM pour la vraie distance routière
    dist = self._fetch_osrm_distance(
        coords_orig[0], coords_orig[1],
        coords_dest[0], coords_dest[1],
    )

    # 4. Fallback Haversine si OSRM est indisponible
    if dist is None:
        logger.warning(
            f"OSRM indisponible pour '{origine}' -> '{destination}'. "
            "Utilisation de la formule Haversine (distance approximée)."
        )
        dist = self._haversine_distance(
            coords_orig[0], coords_orig[1],
            coords_dest[0], coords_dest[1],
        )

    # Mise en cache du résultat
    self.cache["distances"][route_key] = dist
    self._save_cache()

    return dist

transfer_cost ¶

transfer_cost(origine: str, destination: str, prix_carburant: float = None, crop: str = None) -> dict

Calcule le coût total de transfert d'un point A vers un point B.

Phase 4 - Intégration météo

Si une session weather a été fourni à l'initialisation, le calcul ajuste automatiquement : - gamma_route selon la probabilité de pluie prévue demain. - La perte de qualité selon la culture et la météo.

Formule

C_transfer = C_info + Distance * (gamma_effectif * P_carburant / 100 + mu_checkpoints) + C_qualite(culture, distance, pluie)

Ou : - C_info = coût fixe de recherche d'informations (appels, déplacements) - gamma_effectif = gamma_route * (1 + alpha_pluie * prob_pluie) - P_carburant = prix du litre d'essence en XOF - mu_checkpoints = coût moyen des tracasseries par km - C_qualite = perte de valeur marchande (variable par culture et météo)

Parameters:

Name	Type	Description	Default
`origine`	`str`	Ville de départ (ex: 'Parakou').	required
`destination`	`str`	Ville de destination (ex: 'Cotonou').	required
`prix_carburant`	`float`	Prix du litre d'essence en XOF. Si None, récupéré automatiquement depuis les sources configurées.	`None`
`crop`	`str`	Code de la culture transportée (ex: 'maize'). Influence la perte de qualité. Si None, utilise le facteur par défaut.	`None`

Returns:

Name	Type	Description
`dict`	`dict`	Dictionnaire contenant le coût total et le détail par poste : - 'total_cost_cfa' : coût total du transfert en XOF - 'details' : sous-dictionnaire avec chaque composante - 'prob_pluie' : probabilité de pluie utilisée (0 si pas de météo) - 'gamma_effectif' : gamma_route réellement appliqué

Source code in kadi/market/logistics.py

def transfer_cost(
    self,
    origine: str,
    destination: str,
    prix_carburant: float = None,
    crop: str = None,
) -> dict:
    """
    Calcule le coût total de transfert d'un point A vers un point B.

    Phase 4 - Intégration météo :
        Si une session weather a été fourni à l'initialisation, le calcul
        ajuste automatiquement :
        - gamma_route selon la probabilité de pluie prévue demain.
        - La perte de qualité selon la culture et la météo.

    Formule :
        C_transfer = C_info
                   + Distance * (gamma_effectif * P_carburant / 100 + mu_checkpoints)
                   + C_qualite(culture, distance, pluie)

    Ou :
    - C_info = coût fixe de recherche d'informations (appels, déplacements)
    - gamma_effectif = gamma_route * (1 + alpha_pluie * prob_pluie)
    - P_carburant = prix du litre d'essence en XOF
    - mu_checkpoints = coût moyen des tracasseries par km
    - C_qualite = perte de valeur marchande (variable par culture et météo)

    Args:
        origine (str): Ville de départ (ex: 'Parakou').
        destination (str): Ville de destination (ex: 'Cotonou').
        prix_carburant (float, optional): Prix du litre d'essence en XOF.
            Si None, récupéré automatiquement depuis les sources configurées.
        crop (str, optional): Code de la culture transportée (ex: 'maize').
            Influence la perte de qualité. Si None, utilise le facteur par défaut.

    Returns:
        dict: Dictionnaire contenant le coût total et le détail par poste :
            - 'total_cost_cfa' : coût total du transfert en XOF
            - 'details'        : sous-dictionnaire avec chaque composante
            - 'prob_pluie'     : probabilité de pluie utilisée (0 si pas de météo)
            - 'gamma_effectif' : gamma_route réellement appliqué
    """
    # Récupération du prix du carburant si non fourni
    if prix_carburant is None:
        prix_carburant = self._fetch_fuel_price()

    # Récupération des coefficients depuis la configuration centrale
    c_info = _LOGISTICS_CONFIG.get("c_info_xof", 5000.0)
    gamma_route_base = _LOGISTICS_CONFIG.get("gamma_route", 1.2)
    mu_checkpoints = _LOGISTICS_CONFIG.get("mu_checkpoints_xof_per_km", 15.0)

    # Probabilité de pluie (0.0 si pas de module météo disponible)
    prob_pluie = self._get_prob_pluie()

    # Coefficient de route effectif, ajusté selon la météo
    gamma_effectif = _gamma_effective(gamma_route_base, prob_pluie)

    # Calcul de la distance routière entre les deux villes
    d_ab = self.distance(origine, destination)

    # Calcul du coût lié à la distance (carburant + tracasseries)
    cout_distance = d_ab * ((gamma_effectif * prix_carburant / 100.0) + mu_checkpoints)

    # Perte de qualité dynamique (par culture, distance et météo)
    c_qualite = _quality_loss(crop, d_ab, prob_pluie)

    # Coût de transfert total
    c_transfer = c_info + cout_distance + c_qualite

    # Construction du résultat détaillé
    resultat = {
        "total_cost_cfa": round(c_transfer, 2),
        "prob_pluie": round(prob_pluie, 3),
        "gamma_effectif": round(gamma_effectif, 4),
        "details": {
            "distance_km": round(d_ab, 2),
            "search_costs": c_info,
            "transport_costs": round(cout_distance, 2),
            "quality_loss": c_qualite,
            "fuel_price_used": prix_carburant,
            "gamma_route_base": gamma_route_base,
            "gamma_effectif": round(gamma_effectif, 4),
            "prob_pluie": round(prob_pluie, 3),
            "crop": crop or "_default",
        },
    }

    return resultat

Logistique (kadi.market.logistics)¶

Importation directe¶

Initialisation¶

Méthodes¶

transfer_cost(origine, destination, prix_carburant, crop)¶

distance(origine, destination)¶

Intégration météo¶

Prix du carburant¶

Méthodes dépréciées¶

Classe dépréciée¶

Exemple complet¶

Logistics ¶

__init__ ¶

__getattr__ ¶

distance ¶

transfer_cost ¶

Logistique (`kadi.market.logistics`)¶

`transfer_cost(origine, destination, prix_carburant, crop)`¶

`distance(origine, destination)`¶

init ¶

getattr ¶