Logistics (`kadi.market.logistics`)¶

The Logistics class models real logistics friction across Beninese trade corridors: transport costs, checkpoint harassment fees, and merchantable quality degradation of agricultural produce.

Since v1.2.0, it optionally integrates weather forecasts to dynamically adjust the road coefficient (gamma_route) and quality loss based on expected rain probability.

It is used internally by Market and accessible via market.logistics. It can also be instantiated directly.

Direct Importing¶

from kadi.market import Logistics

Initialization¶

from kadi.market import Logistics

# Without weather integration (V1 behavior)
logistics = Logistics()

# With weather integration (dynamic cost adjustment based on rain forecast)
import kadi as kd

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

Parameters:

Name	Type	Default	Description
`cache_file`	`str`	`None`	Path to JSON distance cache file. Default: `~/.kadi/osrm_cache.json`
`weather`	`Weather`	`None`	Weather session for climate cost adjustments. If None, no adjustment (V1 behavior)

Public Attributes:

Attribute	Type	Description
`weather`	`Weather`	Injected weather session (or None)
`cache_file`	`str`	Distance cache file path
`cache`	`dict`	In-memory cache dictionary with keys `'coords'` and `'distances'`

Methods¶

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

Calculates total transfer cost from an origin city to a destination city.

Formula:

C_transfer = C_info
           + Distance * (gamma_effective * P_fuel / 100 + mu_checkpoints)
           + C_quality(crop, distance, rain)

Where: - C_info = fixed information search cost (phone calls, local travel), configurable in config.py (default: 5,000 XOF) - gamma_effective = gamma_route * (1 + alpha_rain * prob_rain): weather surcharge multiplier if a weather session is available - mu_checkpoints = average checkpoint harassment fee per km (default: 15 XOF/km) - C_quality = variable merchantable quality loss per crop type and weather condition

import kadi as kd

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

# Transfer cost with weather integration
cost = market.logistics.transfer_cost(
    origine="Parakou",
    destination="Cotonou",
    crop="maize",
)

print(f"Total cost        : {cost['total_cost_cfa']:,.0f} XOF")
print(f"Rain forecast     : {cost['prob_pluie'] * 100:.0f}%")
print(f"Effective gamma   : {cost['gamma_effectif']:.4f}")

# Breakdown details
d = cost["details"]
print(f"  Distance        : {d['distance_km']:.1f} km")
print(f"  Search cost     : {d['search_costs']:,.0f} XOF")
print(f"  Transport cost  : {d['transport_costs']:,.0f} XOF")
print(f"  Quality loss    : {d['quality_loss']:,.0f} XOF")
print(f"  Fuel price used : {d['fuel_price_used']:.0f} XOF/liter")

Parameters:

Name	Type	Default	Description
`origine`	`str`	required	Departure city (e.g. `'Parakou'`)
`destination`	`str`	required	Arrival city (e.g. `'Cotonou'`)
`prix_carburant`	`float`	`None`	Fuel price in XOF per liter. If None, fetched automatically
`crop`	`str`	`None`	Transported crop code for quality loss calculation. If None, uses default factor

Returns: dict

Key	Type	Description
`total_cost_cfa`	`float`	Total transfer cost in XOF
`prob_pluie`	`float`	Rain probability used (0.0 if no weather session)
`gamma_effectif`	`float`	Effective road coefficient applied
`details`	`dict`	Breakdown dictionary of each cost component

Keys in details:

Key	Description
`distance_km`	Road distance in km
`search_costs`	Fixed information search cost (XOF)
`transport_costs`	Distance-based transport cost (fuel + checkpoints) (XOF)
`quality_loss`	Merchantable quality loss value (XOF)
`fuel_price_used`	Fuel price applied (XOF/liter)
`gamma_route_base`	Base road coefficient before weather adjustment
`gamma_effectif`	Final road coefficient after weather adjustment
`prob_pluie`	Rain probability used
`crop`	Crop code (`'_default'` if unspecified)

Quality Loss Factors by Crop (XOF/km/tonne):

Crop	Factor	Crop	Factor
`maize`, `sorghum`, `millet`	5.0	`cowpea`, `soybean`	7.0
`rice`	6.0	`yam`	12.0
`cassava`	10.0	`tomato`	25.0
`onion`	20.0	(default)	8.0

Under rainy conditions, quality loss increases according to: C_quality = factor * distance_km * (1 + beta_rain * prob_rain), where beta_rain is configurable in config.py (default: 0.5).

`distance(origine, destination)`¶

Retrieves road distance between two Beninese cities.

Cascade Strategy:

Local cache (result from previous call, persisted in JSON file)
Nominatim geocoding (OpenStreetMap) + OSRM routing
Haversine fallback (great-circle distance multiplied by 1.3 tortuosity factor) if OSRM is unavailable
Default fallback of 100 km if geocoding fails

d = market.logistics.distance("Parakou", "Cotonou")
print(f"Road distance : {d:.1f} km")

# Result automatically cached
d2 = market.logistics.distance("Cotonou", "Parakou")   # Returned from cache

Parameters:

Name	Type	Description
`origine`	`str`	Departure city name
`destination`	`str`	Arrival city name

Returns: float - Estimated distance in kilometers.

The result is automatically cached in memory for the current session and saved to cache_file for future sessions.

Weather Integration¶

When a weather instance is provided during initialization, transfer cost calculation is automatically adjusted:

gamma_effective = gamma_route * (1 + alpha_rain * prob_rain)

gamma_route: base coefficient (configurable in config.py, default: 1.2)
alpha_rain: weather surcharge intensity (default: 0.25)
prob_rain: tomorrow's rain probability, fetched via weather.rain_prob(days=1)

Rain probability is fetched once per session (cached) to prevent redundant weather API calls.

Example:

With prob_rain=0.8 and alpha=0.25, the road coefficient increases from 1.2 to 1.2 * (1 + 0.25 * 0.8) = 1.44, representing a 20% surge in transport costs.

Fuel Price Retrieval Strategy¶

The transfer_cost() method automatically retrieves fuel price using the following priority order:

Priority	Source
1	`BENIN_FUEL_PRICE` environment variable
2	In-memory session cache (one query per session)
3	`config/fuel_prices.json` file hosted on GitHub
4	Fallback value from `config.py` (default: 680 XOF/liter)

Deprecated Methods¶

Legacy Method	Replaced By	Since
`calculate_transfer_cost(origine, destination)`	`transfer_cost(origine, destination)`	v1.2.0
`get_distance(origine, destination)`	`distance(origine, destination)`	v1.2.0

Deprecated Class¶

Legacy Class	Replaced By	Since
`MarketLogistics`	`Logistics`	v1.2.0

Complete Example¶

import kadi as kd

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

# Compare transfer costs for two crops (tomato being highly perishable)
crops = ["maize", "tomato"]
for crop in crops:
    cost = market.logistics.transfer_cost("Parakou", "Cotonou", crop=crop)
    print(
        f"{crop:8s} -> {cost['total_cost_cfa']:,.0f} XOF "
        f"(quality loss: {cost['details']['quality_loss']:,.0f} XOF)"
    )

# Distance only
d = market.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

Logistics (kadi.market.logistics)¶

Direct Importing¶

Initialization¶

Methods¶

transfer_cost(origine, destination, prix_carburant, crop)¶

distance(origine, destination)¶

Weather Integration¶

Fuel Price Retrieval Strategy¶

Deprecated Methods¶

Deprecated Class¶

Complete Example¶

Logistics ¶

__init__ ¶

__getattr__ ¶

distance ¶

transfer_cost ¶

Logistics (`kadi.market.logistics`)¶

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

`distance(origine, destination)`¶

init ¶

getattr ¶