SafeRouteEngine

Fait par Gillesto

saferoute.engine.SafeRouteEngine

Moteur principal SafeRoute. Orchestre le cache, le chargement OSMnx, le scoring KDE et les appels au core Rust.

Avertissement : le score de risque est une densité statistique de crimes passés ; il ne garantit en aucun cas la sécurité d'un trajet.

Source code in src/saferoute/engine.py

class SafeRouteEngine:
    """
    Moteur principal SafeRoute.
    Orchestre le cache, le chargement OSMnx, le scoring KDE et les appels au core Rust.

    Avertissement : le score de risque est une densité statistique de crimes passés ; il ne
    garantit en aucun cas la sécurité d'un trajet.
    """

    def __init__(
        self,
        eps: float = 0.1,
        cache_dir: Path | str | None = None,
        bandwidth_m: float | str | None = None,
        simulate_familiarity: bool = False,
        familiarity_trips: int = 50,
        max_labels: int | None = None,
    ):
        """
        Args:
            eps: approximation de la recherche. 0.0 = frontière de Pareto exacte ;
                                   0.1 = chaque solution exacte est couverte à 10 % près
            cache_dir: répertoire de cache
            bandwidth_m: écart-type du noyau KDE en mètres. None → Silverman (borné à [200 ; 800] m) ;
                "auto" → choisi par validation sur des crimes mis de côté (voir kde_scorer.select_bandwidth)
            simulate_familiarity: si True, simule des trajectoires au chargement
            familiarity_trips: nombre de trajets simulés
            max_labels: budget de la recherche (None → 5 000 000 labels, ~200 Mo au plus).
                                   Dépassé → SearchBudgetExceededError, jamais un crash du processus.
        """
        self.eps = eps
        self.max_labels = max_labels
        self.bandwidth_m = bandwidth_m
        self.simulate_familiarity = simulate_familiarity
        self.familiarity_trips = familiarity_trips
        self._cache = GraphCache(cache_dir) if cache_dir else GraphCache()
        self._familiarity = FamiliarityEngine()

        self._py_graph = None
        self._nx_graph = None
        self._arrays: GraphArrays | None = None
        self._edge_risk: np.ndarray | None = None
        self._current_city: str | None = None
        self._kde_result: KDEResult | None = None
        # Les appels natifs sont sérialisés : une mise à jour de familiarité (écriture) ne doit pas
        # chevaucher un calcul d'itinéraire (lecture, GIL relâché).
        self._native_lock = threading.Lock()

        self._PyGraph, self._compute, self._BudgetError, self._native_error = _import_native()
        if self._native_error is None:
            logger.info("Module natif saferoute.saferoute_core chargé")
        else:
            logger.warning(
                "Module natif introuvable (%s) : le calcul d'itinéraire sera indisponible. "
                "Réinstallez le paquet ou lancez `maturin develop --release`.",
                self._native_error,
            )

    @property
    def native_available(self) -> bool:
        """True si le noyau Rust est chargé et le calcul d'itinéraire disponible."""
        return self._PyGraph is not None and self._compute is not None

    # ── API publique ───────────────────────────────────────────────────────────

    def load_city(
        self,
        city_key: str,
        crime_points: list[dict] | None = None,
        force_download: bool = False,
    ) -> dict:
        """
        Charge une ville : graphe OSMnx + scores KDE.
        Utilise le cache si disponible, télécharge sinon.

        Args:
            city_key: "london" ou "cape_town" (autre ville : voir load_graph)
            crime_points: liste de {"lat","lon","weight"}. Si None → charge depuis cache
            force_download: ignore le cache et re-télécharge

        Returns:
            dict de statistiques de chargement
        """
        if city_key not in SUPPORTED_CITIES:
            raise UnsupportedCityError(city_key, list(SUPPORTED_CITIES.keys()))

        t0 = time.perf_counter()
        G = None
        if not force_download:
            G = self._cache.load_graph(city_key)

        if G is None:
            G = self._download_graph(city_key)
            report = validate_graph(G, city_key)
            if not report.is_valid:
                raise RuntimeError(
                    f"Graphe '{city_key}' invalide après téléchargement :\n{report.summary()}"
                )
            G = extract_largest_scc(G)
            self._cache.save_graph(city_key, G, stats={"bbox": report.bbox})
        t_graph = time.perf_counter() - t0

        if crime_points is None:
            crime_points = self._cache.load_crimes(city_key)
        if not crime_points:
            logger.warning(
                "Aucun crime disponible pour '%s'. Scores de risque tous à 0. "
                "Lancez scripts/download_cities.py pour pré-charger les données.",
                city_key,
            )
            crime_points = []

        stats = self._ingest(G, crime_points, city_key)
        stats["timings_s"]["load_graph"] = round(t_graph, 3)
        return stats

    def load_graph(
        self,
        nx_graph: Any,
        crime_points: list[dict] | None = None,
        name: str = "custom",
        keep_largest_component: bool = True,
    ) -> dict:
        """
        Charge VOTRE graphe et VOS données de risque (n'importe quelle ville, sans toucher au code).

        Args:
            nx_graph: ``networkx.MultiDiGraph`` de type OSMnx : nœuds à identifiant
                                     entier avec attributs ``y`` (lat) et ``x`` (lon) ; arcs avec
                                     ``length`` en mètres (``osmnx.distance.add_edge_lengths``).
            crime_points: liste de ``{"lat", "lon", "weight"}`` (poids ∈ ]0 ; 10]).
                                     ``None`` ou vide → scores de risque nuls.
            name: nom affiché (``_current_city``).
            keep_largest_component: ne garde que la plus grande composante fortement connexe
                                     (évite les nœuds inatteignables).

        Returns:
            dict de statistiques de chargement (mêmes clés que load_city).

        Raises:
            ValueError: graphe invalide (vide, nœud sans coordonnées, trop d'arcs sans longueur).
        """
        report = validate_graph(nx_graph, name)
        if not report.is_valid:
            raise ValueError(f"Graphe '{name}' invalide :\n{report.summary()}")
        G = extract_largest_scc(nx_graph) if keep_largest_component else nx_graph
        return self._ingest(G, list(crime_points or []), name)

    def record_trip(self, path: list[int]) -> None:
        """
        Enregistre un trajet réel pour mettre à jour la familiarité.
        Appeler après chaque trajet effectué par l'utilisateur.

        Args:
            path: liste de NodeId OSM du trajet effectué
        """
        if self._nx_graph is None:
            raise GraphNotLoadedError("record_trip")
        # La familiarité retient les arcs modifiés ; ils seront envoyés à Rust au prochain
        # compute_routes() (mise à jour partielle, pas de reconstruction du graphe).
        self._familiarity.update_from_path(path, self._nx_graph)
        logger.info(
            "Trajet enregistré. Total trajets : %d", self._familiarity.get_familiarity_map().total_trips
        )

    def compute_routes(self, source_node: int, target_node: int) -> ParetoSet:
        """
        Calcule les itinéraires Pareto-optimaux (shortest, safest, balanced) entre deux nœuds OSM.

        Args:
            source_node: NodeId OSM de départ
            target_node: NodeId OSM d'arrivée

        Returns:
            ParetoSet avec shortest, safest, balanced

        Raises:
            GraphNotLoadedError       : aucun graphe chargé
            NativeModuleNotFoundError : module Rust absent
            RouteNotFoundError        : nœud inconnu ou aucun chemin
            SearchBudgetExceededError : requête trop complexe (voir max_labels, eps)
        """
        if self._nx_graph is None:
            raise GraphNotLoadedError("compute_routes")
        if not self.native_available:
            raise NativeModuleNotFoundError(self._native_error or "")
        if self._py_graph is None:
            raise GraphNotLoadedError("compute_routes")

        try:
            with self._native_lock:
                self._sync_familiarity()
                results = self._compute(
                    self._py_graph, int(source_node), int(target_node), float(self.eps), self.max_labels
                )
        except self._BudgetError as exc:  # sous-classe de RuntimeError : à traiter en premier
            raise SearchBudgetExceededError(source_node, target_node, str(exc)) from exc
        except (ValueError, RuntimeError) as exc:
            raise RouteNotFoundError(source_node, target_node, str(exc)) from exc

        return ParetoSet.from_results(results)

    def nearest_node(self, lat: float, lon: float) -> int:
        """Retourne le NodeId OSM le plus proche d'un point GPS."""
        if self._nx_graph is None:
            raise GraphNotLoadedError("nearest_node")
        return int(ox.nearest_nodes(self._nx_graph, lon, lat))

    def get_node_coords(self, node_id: int) -> tuple[float, float]:
        """Retourne (lat, lon) d'un nœud OSM."""
        if self._nx_graph is None:
            raise GraphNotLoadedError("get_node_coords")
        data = self._nx_graph.nodes[node_id]
        return float(data["y"]), float(data["x"])

    def get_risk_map_geojson(
        self,
        bbox: tuple[float, float, float, float] | None = None,
        max_features: int = 5_000,
    ) -> dict:
        """Retourne la heatmap de risque au format GeoJSON, filtrée par bbox.

        Args:
            bbox: (min_lat, max_lat, min_lon, max_lon). Si None, retourne tout.
            max_features: Nombre maximum d'arcs retournés (défaut 5 000).

        Returns:
            GeoJSON FeatureCollection.
        """
        if self._nx_graph is None or self._kde_result is None or self._arrays is None:
            return {"type": "FeatureCollection", "features": []}

        arrays = self._arrays
        mid_lat, mid_lon = arrays.midpoints()
        if bbox is not None:
            min_lat, max_lat, min_lon, max_lon = bbox
            mask = (mid_lat >= min_lat) & (mid_lat <= max_lat) & (mid_lon >= min_lon) & (mid_lon <= max_lon)
            idx = np.flatnonzero(mask)[:max_features]
        else:
            idx = np.arange(min(arrays.n_edges, max_features))
        if idx.size == 0:
            return {"type": "FeatureCollection", "features": []}

        lat_u, lon_u, lat_v, lon_v = arrays.endpoint_coords(idx)
        risk = self._edge_risk[idx] if self._edge_risk is not None else np.zeros(idx.size)
        features = [
            {
                "type": "Feature",
                "geometry": {"type": "LineString", "coordinates": [[lo_u, la_u], [lo_v, la_v]]},
                "properties": {
                    "risk_score": round(r, 4),
                    "from": int(u),
                    "to": int(v),
                    "length_m": ln,
                },
            }
            for la_u, lo_u, la_v, lo_v, r, u, v, ln in zip(
                lat_u.tolist(), lon_u.tolist(), lat_v.tolist(), lon_v.tolist(),
                risk.tolist(), arrays.edge_u[idx].tolist(), arrays.edge_v[idx].tolist(),
                arrays.edge_length[idx].tolist(),
                strict=True,
            )
        ]
        return {"type": "FeatureCollection", "features": features}

    # ── Méthodes privées ───────────────────────────────────────────────────────

    def _ingest(self, G, crime_points: list[dict], name: str) -> dict:
        """Chaîne commune à load_city et load_graph : colonnes → KDE → familiarité → Rust."""
        timings: dict[str, float] = {}
        self._nx_graph = G
        self._current_city = name

        t = time.perf_counter()
        self._arrays = GraphArrays.from_networkx(G)
        timings["arrays"] = round(time.perf_counter() - t, 3)

        logger.info("Calcul KDE sur %d crimes...", len(crime_points))
        self._kde_result = compute_kde_scores(G, crime_points, bandwidth_m=self.bandwidth_m, arrays=self._arrays)
        timings["kde"] = round(self._kde_result.elapsed_s, 3)
        self._edge_risk = (
            self._kde_result.edge_risk
            if self._kde_result.edge_risk is not None
            else np.zeros(self._arrays.n_edges, dtype=np.float64)
        )

        if self.simulate_familiarity:
            logger.info("Simulation de %d trajets de familiarité...", self.familiarity_trips)
            self._familiarity.simulate_trajectories(G, n_trips=self.familiarity_trips)

        self._py_graph = None
        if self._PyGraph is not None:
            t = time.perf_counter()
            self._build_native_graph()
            timings["native_build"] = round(time.perf_counter() - t, 3)

        fmap_stats = self._familiarity.get_familiarity_map().stats()
        return {
            "city": name,
            "nodes": G.number_of_nodes(),
            "edges": G.number_of_edges(),
            "crimes": len(crime_points),
            "kde_bandwidth_m": self._kde_result.bandwidth_m,
            "kde_method": self._kde_result.method,
            "kde_elapsed_s": self._kde_result.elapsed_s,
            "kde_stats": self._kde_result.stats,
            "familiarity_stats": fmap_stats,
            "native_graph": self._py_graph is not None,
            "timings_s": timings,
        }

    def _familiarity_vector(self) -> np.ndarray:
        """Familiarité de chaque arc (zéro partout, puis les arcs connus de la carte)."""
        assert self._arrays is not None
        vec = np.zeros(self._arrays.n_edges, dtype=np.float64)
        scores = self._familiarity.get_familiarity_map().scores
        if scores:
            keys = list(scores)
            pos = self._arrays.positions(keys)
            ok = pos >= 0
            vec[pos[ok]] = np.fromiter(scores.values(), dtype=np.float64, count=len(keys))[ok]
        return vec

    def _build_native_graph(self) -> None:
        assert self._arrays is not None and self._edge_risk is not None and self._PyGraph is not None
        fam = self._familiarity_vector()
        columns = self._arrays.to_native_columns(self._edge_risk, fam)
        graph = self._PyGraph.from_arrays(*columns)
        with self._native_lock:
            self._py_graph = graph
        self._familiarity.get_familiarity_map().mark_synced()
        logger.info(
            "Graphe '%s' chargé dans Rust : %d nœuds, %d arcs",
            self._current_city, graph.node_count(), graph.edge_count(),
        )

    def _sync_familiarity(self) -> None:
        """Envoie à Rust UNIQUEMENT les arcs dont la familiarité a changé (verrou natif déjà pris)."""
        fmap = self._familiarity.get_familiarity_map()
        pending = fmap.pending()
        if not pending or self._py_graph is None or self._arrays is None:
            return
        keys = list(pending)
        pos = self._arrays.positions(keys)
        ok = pos >= 0
        values = np.fromiter(pending.values(), dtype=np.float64, count=len(keys))
        self._py_graph.update_familiarity(
            pos[ok].astype("<u8").tobytes(), values[ok].astype("<f8").tobytes()
        )
        fmap.mark_synced(keys)
        logger.debug("Familiarité synchronisée : %d arcs mis à jour (sur %d)", int(ok.sum()), self._arrays.n_edges)

    def _download_graph(self, city_key: str):
        """Télécharge le graphe OSMnx pour une ville."""
        config = SUPPORTED_CITIES[city_key]
        method = config.get("method", "place")
        try:
            if method == "place":
                logger.info("Téléchargement OSMnx (place) : %s...", config["query"])
                G = ox.graph_from_place(config["query"], network_type=config["network_type"])
            else:
                logger.info(
                    "Téléchargement OSMnx (point) : %s, rayon %sm...", config["point"], config["dist"]
                )
                G = ox.graph_from_point(config["point"], dist=config["dist"], network_type=config["network_type"])

            if not any("length" in d for _, _, d in G.edges(data=True)):
                G = ox.distance.add_edge_lengths(G)
            return G
        except Exception as e:
            raise RuntimeError(f"Échec téléchargement OSMnx pour '{city_key}': {e}") from e

Attributes

native_available property

native_available: bool

True si le noyau Rust est chargé et le calcul d'itinéraire disponible.

Methods:

__init__

__init__(eps: float = 0.1, cache_dir: Path | str | None = None, bandwidth_m: float | str | None = None, simulate_familiarity: bool = False, familiarity_trips: int = 50, max_labels: int | None = None)

Parameters:

Name	Type	Description	Default
eps	float	approximation de la recherche. 0.0 = frontière de Pareto exacte ; 0.1 = chaque solution exacte est couverte à 10 % près	0.1
cache_dir	Path | str | None	répertoire de cache	None
bandwidth_m	float | str | None	écart-type du noyau KDE en mètres. None → Silverman (borné à [200 ; 800] m) ; "auto" → choisi par validation sur des crimes mis de côté (voir kde_scorer.select_bandwidth)	None
simulate_familiarity	bool	si True, simule des trajectoires au chargement	False
familiarity_trips	int	nombre de trajets simulés	50
max_labels	int | None	budget de la recherche (None → 5 000 000 labels, ~200 Mo au plus). Dépassé → SearchBudgetExceededError, jamais un crash du processus.	None

Source code in src/saferoute/engine.py

def __init__(
    self,
    eps: float = 0.1,
    cache_dir: Path | str | None = None,
    bandwidth_m: float | str | None = None,
    simulate_familiarity: bool = False,
    familiarity_trips: int = 50,
    max_labels: int | None = None,
):
    """
    Args:
        eps: approximation de la recherche. 0.0 = frontière de Pareto exacte ;
                               0.1 = chaque solution exacte est couverte à 10 % près
        cache_dir: répertoire de cache
        bandwidth_m: écart-type du noyau KDE en mètres. None → Silverman (borné à [200 ; 800] m) ;
            "auto" → choisi par validation sur des crimes mis de côté (voir kde_scorer.select_bandwidth)
        simulate_familiarity: si True, simule des trajectoires au chargement
        familiarity_trips: nombre de trajets simulés
        max_labels: budget de la recherche (None → 5 000 000 labels, ~200 Mo au plus).
                               Dépassé → SearchBudgetExceededError, jamais un crash du processus.
    """
    self.eps = eps
    self.max_labels = max_labels
    self.bandwidth_m = bandwidth_m
    self.simulate_familiarity = simulate_familiarity
    self.familiarity_trips = familiarity_trips
    self._cache = GraphCache(cache_dir) if cache_dir else GraphCache()
    self._familiarity = FamiliarityEngine()

    self._py_graph = None
    self._nx_graph = None
    self._arrays: GraphArrays | None = None
    self._edge_risk: np.ndarray | None = None
    self._current_city: str | None = None
    self._kde_result: KDEResult | None = None
    # Les appels natifs sont sérialisés : une mise à jour de familiarité (écriture) ne doit pas
    # chevaucher un calcul d'itinéraire (lecture, GIL relâché).
    self._native_lock = threading.Lock()

    self._PyGraph, self._compute, self._BudgetError, self._native_error = _import_native()
    if self._native_error is None:
        logger.info("Module natif saferoute.saferoute_core chargé")
    else:
        logger.warning(
            "Module natif introuvable (%s) : le calcul d'itinéraire sera indisponible. "
            "Réinstallez le paquet ou lancez `maturin develop --release`.",
            self._native_error,
        )

load_city

load_city(city_key: str, crime_points: list[dict] | None = None, force_download: bool = False) -> dict

Charge une ville : graphe OSMnx + scores KDE. Utilise le cache si disponible, télécharge sinon.

Parameters:

Name	Type	Description	Default
city_key	str	"london" ou "cape_town" (autre ville : voir load_graph)	required
crime_points	list[dict] | None	liste de {"lat","lon","weight"}. Si None → charge depuis cache	None
force_download	bool	ignore le cache et re-télécharge	False

Returns:

Type	Description
dict	dict de statistiques de chargement

Source code in src/saferoute/engine.py

def load_city(
    self,
    city_key: str,
    crime_points: list[dict] | None = None,
    force_download: bool = False,
) -> dict:
    """
    Charge une ville : graphe OSMnx + scores KDE.
    Utilise le cache si disponible, télécharge sinon.

    Args:
        city_key: "london" ou "cape_town" (autre ville : voir load_graph)
        crime_points: liste de {"lat","lon","weight"}. Si None → charge depuis cache
        force_download: ignore le cache et re-télécharge

    Returns:
        dict de statistiques de chargement
    """
    if city_key not in SUPPORTED_CITIES:
        raise UnsupportedCityError(city_key, list(SUPPORTED_CITIES.keys()))

    t0 = time.perf_counter()
    G = None
    if not force_download:
        G = self._cache.load_graph(city_key)

    if G is None:
        G = self._download_graph(city_key)
        report = validate_graph(G, city_key)
        if not report.is_valid:
            raise RuntimeError(
                f"Graphe '{city_key}' invalide après téléchargement :\n{report.summary()}"
            )
        G = extract_largest_scc(G)
        self._cache.save_graph(city_key, G, stats={"bbox": report.bbox})
    t_graph = time.perf_counter() - t0

    if crime_points is None:
        crime_points = self._cache.load_crimes(city_key)
    if not crime_points:
        logger.warning(
            "Aucun crime disponible pour '%s'. Scores de risque tous à 0. "
            "Lancez scripts/download_cities.py pour pré-charger les données.",
            city_key,
        )
        crime_points = []

    stats = self._ingest(G, crime_points, city_key)
    stats["timings_s"]["load_graph"] = round(t_graph, 3)
    return stats

load_graph

load_graph(nx_graph: Any, crime_points: list[dict] | None = None, name: str = 'custom', keep_largest_component: bool = True) -> dict

Charge VOTRE graphe et VOS données de risque (n'importe quelle ville, sans toucher au code).

Parameters:

Name	Type	Description	Default
nx_graph	Any	networkx.MultiDiGraph de type OSMnx : nœuds à identifiant entier avec attributs y (lat) et x (lon) ; arcs avec length en mètres (osmnx.distance.add_edge_lengths).	required
crime_points	list[dict] | None	liste de {"lat", "lon", "weight"} (poids ∈ ]0 ; 10]). None ou vide → scores de risque nuls.	None
name	str	nom affiché (_current_city).	'custom'
keep_largest_component	bool	ne garde que la plus grande composante fortement connexe (évite les nœuds inatteignables).	True

Returns:

Type	Description
dict	dict de statistiques de chargement (mêmes clés que load_city).

Raises:

Type	Description
ValueError	graphe invalide (vide, nœud sans coordonnées, trop d'arcs sans longueur).

Source code in src/saferoute/engine.py

def load_graph(
    self,
    nx_graph: Any,
    crime_points: list[dict] | None = None,
    name: str = "custom",
    keep_largest_component: bool = True,
) -> dict:
    """
    Charge VOTRE graphe et VOS données de risque (n'importe quelle ville, sans toucher au code).

    Args:
        nx_graph: ``networkx.MultiDiGraph`` de type OSMnx : nœuds à identifiant
                                 entier avec attributs ``y`` (lat) et ``x`` (lon) ; arcs avec
                                 ``length`` en mètres (``osmnx.distance.add_edge_lengths``).
        crime_points: liste de ``{"lat", "lon", "weight"}`` (poids ∈ ]0 ; 10]).
                                 ``None`` ou vide → scores de risque nuls.
        name: nom affiché (``_current_city``).
        keep_largest_component: ne garde que la plus grande composante fortement connexe
                                 (évite les nœuds inatteignables).

    Returns:
        dict de statistiques de chargement (mêmes clés que load_city).

    Raises:
        ValueError: graphe invalide (vide, nœud sans coordonnées, trop d'arcs sans longueur).
    """
    report = validate_graph(nx_graph, name)
    if not report.is_valid:
        raise ValueError(f"Graphe '{name}' invalide :\n{report.summary()}")
    G = extract_largest_scc(nx_graph) if keep_largest_component else nx_graph
    return self._ingest(G, list(crime_points or []), name)

record_trip

record_trip(path: list[int]) -> None

Enregistre un trajet réel pour mettre à jour la familiarité. Appeler après chaque trajet effectué par l'utilisateur.

Parameters:

Name	Type	Description	Default
path	list[int]	liste de NodeId OSM du trajet effectué	required

Source code in src/saferoute/engine.py

def record_trip(self, path: list[int]) -> None:
    """
    Enregistre un trajet réel pour mettre à jour la familiarité.
    Appeler après chaque trajet effectué par l'utilisateur.

    Args:
        path: liste de NodeId OSM du trajet effectué
    """
    if self._nx_graph is None:
        raise GraphNotLoadedError("record_trip")
    # La familiarité retient les arcs modifiés ; ils seront envoyés à Rust au prochain
    # compute_routes() (mise à jour partielle, pas de reconstruction du graphe).
    self._familiarity.update_from_path(path, self._nx_graph)
    logger.info(
        "Trajet enregistré. Total trajets : %d", self._familiarity.get_familiarity_map().total_trips
    )

compute_routes

compute_routes(source_node: int, target_node: int) -> ParetoSet

Calcule les itinéraires Pareto-optimaux (shortest, safest, balanced) entre deux nœuds OSM.

Parameters:

Name	Type	Description	Default
source_node	int	NodeId OSM de départ	required
target_node	int	NodeId OSM d'arrivée	required

Returns:

Type	Description
ParetoSet	ParetoSet avec shortest, safest, balanced

Raises:

Type	Description
GraphNotLoadedError	aucun graphe chargé
NativeModuleNotFoundError	module Rust absent
RouteNotFoundError	nœud inconnu ou aucun chemin
SearchBudgetExceededError	requête trop complexe (voir max_labels, eps)

Source code in src/saferoute/engine.py

def compute_routes(self, source_node: int, target_node: int) -> ParetoSet:
    """
    Calcule les itinéraires Pareto-optimaux (shortest, safest, balanced) entre deux nœuds OSM.

    Args:
        source_node: NodeId OSM de départ
        target_node: NodeId OSM d'arrivée

    Returns:
        ParetoSet avec shortest, safest, balanced

    Raises:
        GraphNotLoadedError       : aucun graphe chargé
        NativeModuleNotFoundError : module Rust absent
        RouteNotFoundError        : nœud inconnu ou aucun chemin
        SearchBudgetExceededError : requête trop complexe (voir max_labels, eps)
    """
    if self._nx_graph is None:
        raise GraphNotLoadedError("compute_routes")
    if not self.native_available:
        raise NativeModuleNotFoundError(self._native_error or "")
    if self._py_graph is None:
        raise GraphNotLoadedError("compute_routes")

    try:
        with self._native_lock:
            self._sync_familiarity()
            results = self._compute(
                self._py_graph, int(source_node), int(target_node), float(self.eps), self.max_labels
            )
    except self._BudgetError as exc:  # sous-classe de RuntimeError : à traiter en premier
        raise SearchBudgetExceededError(source_node, target_node, str(exc)) from exc
    except (ValueError, RuntimeError) as exc:
        raise RouteNotFoundError(source_node, target_node, str(exc)) from exc

    return ParetoSet.from_results(results)

nearest_node

nearest_node(lat: float, lon: float) -> int

Retourne le NodeId OSM le plus proche d'un point GPS.

Source code in src/saferoute/engine.py

def nearest_node(self, lat: float, lon: float) -> int:
    """Retourne le NodeId OSM le plus proche d'un point GPS."""
    if self._nx_graph is None:
        raise GraphNotLoadedError("nearest_node")
    return int(ox.nearest_nodes(self._nx_graph, lon, lat))

get_node_coords

get_node_coords(node_id: int) -> tuple[float, float]

Retourne (lat, lon) d'un nœud OSM.

Source code in src/saferoute/engine.py

def get_node_coords(self, node_id: int) -> tuple[float, float]:
    """Retourne (lat, lon) d'un nœud OSM."""
    if self._nx_graph is None:
        raise GraphNotLoadedError("get_node_coords")
    data = self._nx_graph.nodes[node_id]
    return float(data["y"]), float(data["x"])

get_risk_map_geojson

get_risk_map_geojson(bbox: tuple[float, float, float, float] | None = None, max_features: int = 5000) -> dict

Retourne la heatmap de risque au format GeoJSON, filtrée par bbox.

Parameters:

Name	Type	Description	Default
bbox	tuple[float, float, float, float] | None	(min_lat, max_lat, min_lon, max_lon). Si None, retourne tout.	None
max_features	int	Nombre maximum d'arcs retournés (défaut 5 000).	5000

Returns:

Type	Description
dict	GeoJSON FeatureCollection.

Source code in src/saferoute/engine.py

def get_risk_map_geojson(
    self,
    bbox: tuple[float, float, float, float] | None = None,
    max_features: int = 5_000,
) -> dict:
    """Retourne la heatmap de risque au format GeoJSON, filtrée par bbox.

    Args:
        bbox: (min_lat, max_lat, min_lon, max_lon). Si None, retourne tout.
        max_features: Nombre maximum d'arcs retournés (défaut 5 000).

    Returns:
        GeoJSON FeatureCollection.
    """
    if self._nx_graph is None or self._kde_result is None or self._arrays is None:
        return {"type": "FeatureCollection", "features": []}

    arrays = self._arrays
    mid_lat, mid_lon = arrays.midpoints()
    if bbox is not None:
        min_lat, max_lat, min_lon, max_lon = bbox
        mask = (mid_lat >= min_lat) & (mid_lat <= max_lat) & (mid_lon >= min_lon) & (mid_lon <= max_lon)
        idx = np.flatnonzero(mask)[:max_features]
    else:
        idx = np.arange(min(arrays.n_edges, max_features))
    if idx.size == 0:
        return {"type": "FeatureCollection", "features": []}

    lat_u, lon_u, lat_v, lon_v = arrays.endpoint_coords(idx)
    risk = self._edge_risk[idx] if self._edge_risk is not None else np.zeros(idx.size)
    features = [
        {
            "type": "Feature",
            "geometry": {"type": "LineString", "coordinates": [[lo_u, la_u], [lo_v, la_v]]},
            "properties": {
                "risk_score": round(r, 4),
                "from": int(u),
                "to": int(v),
                "length_m": ln,
            },
        }
        for la_u, lo_u, la_v, lo_v, r, u, v, ln in zip(
            lat_u.tolist(), lon_u.tolist(), lat_v.tolist(), lon_v.tolist(),
            risk.tolist(), arrays.edge_u[idx].tolist(), arrays.edge_v[idx].tolist(),
            arrays.edge_length[idx].tolist(),
            strict=True,
        )
    ]
    return {"type": "FeatureCollection", "features": features}