Aller au contenu

Familiarité

Fait par Gillesto

saferoute.familiarity.FamiliarityEngine

Moteur de familiarité : calcule et met à jour les scores d'arcs à partir de trajectoires utilisateur (réelles ou simulées).

Source code in src/saferoute/familiarity.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
class FamiliarityEngine:
    """
    Moteur de familiarité : calcule et met à jour les scores d'arcs
    à partir de trajectoires utilisateur (réelles ou simulées).
    """

    def __init__(self, decay_per_trip: float = 0.02):
        """
        Args:
            decay_per_trip: décroissance appliquée à chaque nouveau trajet
                             (simule l'oubli progressif des routes peu fréquentées)
        """
        self.decay_per_trip = decay_per_trip
        self._fmap = FamiliarityMap()

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

    def update_from_path(self, path: list[int], nx_graph: Any) -> None:
        """
        Met à jour la familiarité depuis un chemin (liste de NodeId OSM).

        Args:
            path: liste de NodeId OSM [n1, n2, n3, ...]
            nx_graph: graphe NetworkX pour trouver les clés d'arcs
        """
        if len(path) < 2:
            return

        # Décroissance légère à chaque nouveau trajet
        if self.decay_per_trip > 0:
            self._fmap.decay(factor=1.0 - self.decay_per_trip)

        for i in range(len(path) - 1):
            u, v = path[i], path[i + 1]
            # Trouve la clé de l'arc dans le multigraphe. S'il existe des arcs parallèles, on retient
            # le plus court (celui qu'un piéton emprunte), et non le premier de l'itération.
            if nx_graph.has_edge(u, v):
                parallel = nx_graph[u][v]
                key = min(parallel, key=lambda k: parallel[k].get("length", 100.0), default=0)
                # Incrément inversement proportionnel à la longueur de l'arc
                # (les arcs courts sont "plus familiers" par unité de distance)
                length = parallel[key].get("length", 100.0) if key in parallel else 100.0
                delta = min(0.15, 100.0 / max(length, 1.0))
                self._fmap.update(u, v, key, delta)

        self._fmap.total_trips += 1
        # Ne pas loguer les stats détaillées (données de mobilité utilisateur)
        logger.debug("Trajet enregistré : %d nœuds", len(path))

    def simulate_trajectories(
        self,
        nx_graph: Any,
        n_trips: int = 50,
        seed: int = 42,
    ) -> FamiliarityMap:
        """
        Génère des trajectoires fictives pour simuler un utilisateur habitué
        à certains corridors de la ville.

        Stratégie :
        - Sélectionne aléatoirement N paires (source, destination)
        - Calcule le plus court chemin NetworkX (proxy pour le comportement réel)
        - Met à jour la familiarité pour chaque arc emprunté

        Args:
            nx_graph: graphe NetworkX OSMnx
            n_trips: nombre de trajets simulés
            seed: graine pour reproductibilité

        Returns:
            FamiliarityMap mise à jour
        """
        import networkx as nx

        rng = random.Random(seed)
        nodes = list(nx_graph.nodes())

        if len(nodes) < 2:
            logger.warning("Graphe trop petit pour simuler des trajectoires")
            return self._fmap

        successful = 0
        for _ in range(n_trips):
            src = rng.choice(nodes)
            dst = rng.choice(nodes)
            if src == dst:
                continue

            try:
                # Plus court chemin par longueur (comportement utilisateur naïf)
                path = nx.shortest_path(nx_graph, src, dst, weight="length")
                self.update_from_path(path, nx_graph)
                successful += 1
            except (nx.NetworkXNoPath, nx.NodeNotFound):
                continue

        logger.info(
            f"Simulation : {successful}/{n_trips} trajets réussis | "
            f"stats={self._fmap.stats()}"
        )
        return self._fmap

    def get_familiarity_map(self) -> FamiliarityMap:
        return self._fmap

    def apply_to_graph_json(self, graph_json: str) -> str:
        """
        Injecte les scores de familiarité dans le JSON du graphe
        avant envoi au moteur Rust.

        Args:
            graph_json: JSON produit par engine._serialize_graph()

        Returns:
            JSON enrichi avec les scores de familiarité
        """
        data = json.loads(graph_json)
        fmap = self._fmap.scores

        updated = 0
        for edge in data.get("edges", []):
            key = (edge["from"], edge["to"], 0)
            score = fmap.get(key, 0.0)
            if score > 0:
                edge["familiarity"] = round(score, 4)
                updated += 1

        logger.debug(f"Familiarité injectée : {updated}/{len(data.get('edges', []))} arcs")
        return json.dumps(data)

    def save(self, path: Path | str) -> None:
        """Sauvegarde la carte de familiarité en JSON."""
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        data = {
            "total_trips": self._fmap.total_trips,
            # Clé : "u|v|k" avec séparateur | pour éviter les ambiguïtés
            "scores": {f"{int(u)}|{int(v)}|{int(k)}": v
                       for (u, v, k), v in self._fmap.scores.items()},
        }
        path.write_text(json.dumps(data, indent=2), encoding="utf-8")
        logger.info(f"Familiarité sauvegardée : {len(self._fmap.scores)} arcs → {path}")

    def load(self, path: Path | str) -> None:
        """Charge une carte de familiarité depuis JSON."""
        path = Path(path)
        if not path.exists():
            logger.warning(f"Fichier de familiarité introuvable : {path}")
            return
        data = json.loads(path.read_text(encoding="utf-8"))
        self._fmap.total_trips = data.get("total_trips", 0)
        self._fmap._dirty.update(self._fmap.scores)  # les anciens arcs doivent être remis à zéro
        self._fmap.scores = {}
        for key_str, val in data.get("scores", {}).items():
            parts = key_str.split("|")
            if len(parts) == 3:
                try:
                    u, v, k = int(parts[0]), int(parts[1]), int(parts[2])
                    self._fmap.scores[(u, v, k)] = float(val)
                except ValueError:
                    continue
        self._fmap.mark_all_pending()
        logger.info(f"Familiarité chargée : {len(self._fmap.scores)} arcs depuis {path}")

Methods:

__init__ 

__init__(decay_per_trip: float = 0.02)

Parameters:

Name Type Description Default
decay_per_trip float

décroissance appliquée à chaque nouveau trajet (simule l'oubli progressif des routes peu fréquentées)

 0.02
Source code in src/saferoute/familiarity.py 
 97
 98
 99
100
101
102
103
104
def __init__(self, decay_per_trip: float = 0.02):
    """
    Args:
        decay_per_trip: décroissance appliquée à chaque nouveau trajet
                         (simule l'oubli progressif des routes peu fréquentées)
    """
    self.decay_per_trip = decay_per_trip
    self._fmap = FamiliarityMap()

update_from_path 

update_from_path(path: list[int], nx_graph: Any) -> None

Met à jour la familiarité depuis un chemin (liste de NodeId OSM).

Parameters:

Name Type Description Default
path list[int]

liste de NodeId OSM [n1, n2, n3, ...]

 required
nx_graph Any

graphe NetworkX pour trouver les clés d'arcs

 required
Source code in src/saferoute/familiarity.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def update_from_path(self, path: list[int], nx_graph: Any) -> None:
    """
    Met à jour la familiarité depuis un chemin (liste de NodeId OSM).

    Args:
        path: liste de NodeId OSM [n1, n2, n3, ...]
        nx_graph: graphe NetworkX pour trouver les clés d'arcs
    """
    if len(path) < 2:
        return

    # Décroissance légère à chaque nouveau trajet
    if self.decay_per_trip > 0:
        self._fmap.decay(factor=1.0 - self.decay_per_trip)

    for i in range(len(path) - 1):
        u, v = path[i], path[i + 1]
        # Trouve la clé de l'arc dans le multigraphe. S'il existe des arcs parallèles, on retient
        # le plus court (celui qu'un piéton emprunte), et non le premier de l'itération.
        if nx_graph.has_edge(u, v):
            parallel = nx_graph[u][v]
            key = min(parallel, key=lambda k: parallel[k].get("length", 100.0), default=0)
            # Incrément inversement proportionnel à la longueur de l'arc
            # (les arcs courts sont "plus familiers" par unité de distance)
            length = parallel[key].get("length", 100.0) if key in parallel else 100.0
            delta = min(0.15, 100.0 / max(length, 1.0))
            self._fmap.update(u, v, key, delta)

    self._fmap.total_trips += 1
    # Ne pas loguer les stats détaillées (données de mobilité utilisateur)
    logger.debug("Trajet enregistré : %d nœuds", len(path))

simulate_trajectories 

simulate_trajectories(nx_graph: Any, n_trips: int = 50, seed: int = 42) -> FamiliarityMap

Génère des trajectoires fictives pour simuler un utilisateur habitué à certains corridors de la ville.

Stratégie : - Sélectionne aléatoirement N paires (source, destination) - Calcule le plus court chemin NetworkX (proxy pour le comportement réel) - Met à jour la familiarité pour chaque arc emprunté

Parameters:

Name Type Description Default
nx_graph Any

graphe NetworkX OSMnx

 required
n_trips int

nombre de trajets simulés

 50
seed int

graine pour reproductibilité

 42

Returns:

Type Description
FamiliarityMap

FamiliarityMap mise à jour
Source code in src/saferoute/familiarity.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def simulate_trajectories(
    self,
    nx_graph: Any,
    n_trips: int = 50,
    seed: int = 42,
) -> FamiliarityMap:
    """
    Génère des trajectoires fictives pour simuler un utilisateur habitué
    à certains corridors de la ville.

    Stratégie :
    - Sélectionne aléatoirement N paires (source, destination)
    - Calcule le plus court chemin NetworkX (proxy pour le comportement réel)
    - Met à jour la familiarité pour chaque arc emprunté

    Args:
        nx_graph: graphe NetworkX OSMnx
        n_trips: nombre de trajets simulés
        seed: graine pour reproductibilité

    Returns:
        FamiliarityMap mise à jour
    """
    import networkx as nx

    rng = random.Random(seed)
    nodes = list(nx_graph.nodes())

    if len(nodes) < 2:
        logger.warning("Graphe trop petit pour simuler des trajectoires")
        return self._fmap

    successful = 0
    for _ in range(n_trips):
        src = rng.choice(nodes)
        dst = rng.choice(nodes)
        if src == dst:
            continue

        try:
            # Plus court chemin par longueur (comportement utilisateur naïf)
            path = nx.shortest_path(nx_graph, src, dst, weight="length")
            self.update_from_path(path, nx_graph)
            successful += 1
        except (nx.NetworkXNoPath, nx.NodeNotFound):
            continue

    logger.info(
        f"Simulation : {successful}/{n_trips} trajets réussis | "
        f"stats={self._fmap.stats()}"
    )
    return self._fmap

apply_to_graph_json 

apply_to_graph_json(graph_json: str) -> str

Injecte les scores de familiarité dans le JSON du graphe avant envoi au moteur Rust.

Parameters:

Name Type Description Default
graph_json str

JSON produit par engine._serialize_graph()

 required

Returns:

Type Description
str

JSON enrichi avec les scores de familiarité
Source code in src/saferoute/familiarity.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def apply_to_graph_json(self, graph_json: str) -> str:
    """
    Injecte les scores de familiarité dans le JSON du graphe
    avant envoi au moteur Rust.

    Args:
        graph_json: JSON produit par engine._serialize_graph()

    Returns:
        JSON enrichi avec les scores de familiarité
    """
    data = json.loads(graph_json)
    fmap = self._fmap.scores

    updated = 0
    for edge in data.get("edges", []):
        key = (edge["from"], edge["to"], 0)
        score = fmap.get(key, 0.0)
        if score > 0:
            edge["familiarity"] = round(score, 4)
            updated += 1

    logger.debug(f"Familiarité injectée : {updated}/{len(data.get('edges', []))} arcs")
    return json.dumps(data)

save 

save(path: Path | str) -> None

Sauvegarde la carte de familiarité en JSON.

Source code in src/saferoute/familiarity.py 
221
222
223
224
225
226
227
228
229
230
231
232
def save(self, path: Path | str) -> None:
    """Sauvegarde la carte de familiarité en JSON."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    data = {
        "total_trips": self._fmap.total_trips,
        # Clé : "u|v|k" avec séparateur | pour éviter les ambiguïtés
        "scores": {f"{int(u)}|{int(v)}|{int(k)}": v
                   for (u, v, k), v in self._fmap.scores.items()},
    }
    path.write_text(json.dumps(data, indent=2), encoding="utf-8")
    logger.info(f"Familiarité sauvegardée : {len(self._fmap.scores)} arcs → {path}")

load 

load(path: Path | str) -> None

Charge une carte de familiarité depuis JSON.

Source code in src/saferoute/familiarity.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def load(self, path: Path | str) -> None:
    """Charge une carte de familiarité depuis JSON."""
    path = Path(path)
    if not path.exists():
        logger.warning(f"Fichier de familiarité introuvable : {path}")
        return
    data = json.loads(path.read_text(encoding="utf-8"))
    self._fmap.total_trips = data.get("total_trips", 0)
    self._fmap._dirty.update(self._fmap.scores)  # les anciens arcs doivent être remis à zéro
    self._fmap.scores = {}
    for key_str, val in data.get("scores", {}).items():
        parts = key_str.split("|")
        if len(parts) == 3:
            try:
                u, v, k = int(parts[0]), int(parts[1]), int(parts[2])
                self._fmap.scores[(u, v, k)] = float(val)
            except ValueError:
                continue
    self._fmap.mark_all_pending()
    logger.info(f"Familiarité chargée : {len(self._fmap.scores)} arcs depuis {path}")

saferoute.familiarity.FamiliarityMap dataclass

Carte de familiarité : score [0.0, 1.0] par arc (u, v, key). 0.0 = jamais emprunté, 1.0 = très familier.

La carte retient les arcs modifiés depuis la dernière synchronisation avec le moteur Rust (pending() / mark_synced()), pour ne renvoyer que ces arcs au lieu de reconstruire et re-sérialiser le graphe entier.

Source code in src/saferoute/familiarity.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@dataclass
class FamiliarityMap:
    """
    Carte de familiarité : score [0.0, 1.0] par arc (u, v, key).
    0.0 = jamais emprunté, 1.0 = très familier.

    La carte retient les arcs modifiés depuis la dernière synchronisation avec le moteur Rust
    (`pending()` / `mark_synced()`), pour ne renvoyer que ces arcs au lieu de reconstruire et
    re-sérialiser le graphe entier.
    """
    scores: dict = field(default_factory=dict)  # {(u, v, key): float}
    total_trips: int = 0
    _dirty: set = field(default_factory=set, repr=False)

    def get(self, u: int, v: int, key: int = 0) -> float:
        return float(self.scores.get((u, v, key), 0.0))

    def update(self, u: int, v: int, key: int, delta: float = 0.1) -> None:
        """Incrémente le score d'un arc (avec saturation à 1.0)."""
        current = self.scores.get((u, v, key), 0.0)
        self.scores[(u, v, key)] = min(1.0, current + delta)
        self._dirty.add((u, v, key))

    def decay(self, factor: float = 0.95) -> None:
        """
        Applique un facteur de décroissance temporelle.
        Les routes non empruntées récemment deviennent moins familières.
        factor=0.95 → perd 5% de familiarité par période.
        """
        self._dirty.update(self.scores)  # tous les arcs changent, y compris ceux qui disparaissent
        self.scores = {k: v * factor for k, v in self.scores.items() if v * factor > 0.01}

    def pending(self) -> dict:
        """Arcs modifiés depuis la dernière synchronisation : {(u, v, key): score actuel (0.0 si retiré)}."""
        return {k: self.scores.get(k, 0.0) for k in self._dirty}

    def mark_synced(self, keys=None) -> None:
        """Marque comme synchronisés les arcs `keys` (tous si None)."""
        if keys is None:
            self._dirty.clear()
        else:
            self._dirty.difference_update(keys)

    def mark_all_pending(self) -> None:
        """Force le renvoi de tous les arcs (après un chargement complet depuis un fichier)."""
        self._dirty.update(self.scores)

    def to_edge_dict(self) -> dict:
        """Retourne {(u, v, key): score} pour injection dans le graphe."""
        return dict(self.scores)

    def stats(self) -> dict:
        if not self.scores:
            return {"count": 0, "mean": 0.0, "max": 0.0, "familiar_edges": 0}
        vals = list(self.scores.values())
        return {
            "count": len(vals),
            "mean": sum(vals) / len(vals),
            "max": max(vals),
            "familiar_edges": sum(1 for v in vals if v > 0.3),
            "total_trips": self.total_trips,
        }

Methods:

update 

update(u: int, v: int, key: int, delta: float = 0.1) -> None

Incrémente le score d'un arc (avec saturation à 1.0).

Source code in src/saferoute/familiarity.py 
44
45
46
47
48
def update(self, u: int, v: int, key: int, delta: float = 0.1) -> None:
    """Incrémente le score d'un arc (avec saturation à 1.0)."""
    current = self.scores.get((u, v, key), 0.0)
    self.scores[(u, v, key)] = min(1.0, current + delta)
    self._dirty.add((u, v, key))

decay 

decay(factor: float = 0.95) -> None

Applique un facteur de décroissance temporelle. Les routes non empruntées récemment deviennent moins familières. factor=0.95 → perd 5% de familiarité par période.

Source code in src/saferoute/familiarity.py 
50
51
52
53
54
55
56
57
def decay(self, factor: float = 0.95) -> None:
    """
    Applique un facteur de décroissance temporelle.
    Les routes non empruntées récemment deviennent moins familières.
    factor=0.95 → perd 5% de familiarité par période.
    """
    self._dirty.update(self.scores)  # tous les arcs changent, y compris ceux qui disparaissent
    self.scores = {k: v * factor for k, v in self.scores.items() if v * factor > 0.01}

pending 

pending() -> dict

Arcs modifiés depuis la dernière synchronisation : {(u, v, key): score actuel (0.0 si retiré)}.

Source code in src/saferoute/familiarity.py 
59
60
61
def pending(self) -> dict:
    """Arcs modifiés depuis la dernière synchronisation : {(u, v, key): score actuel (0.0 si retiré)}."""
    return {k: self.scores.get(k, 0.0) for k in self._dirty}

mark_synced 

mark_synced(keys=None) -> None

Marque comme synchronisés les arcs keys (tous si None).

Source code in src/saferoute/familiarity.py 
63
64
65
66
67
68
def mark_synced(self, keys=None) -> None:
    """Marque comme synchronisés les arcs `keys` (tous si None)."""
    if keys is None:
        self._dirty.clear()
    else:
        self._dirty.difference_update(keys)

mark_all_pending 

mark_all_pending() -> None

Force le renvoi de tous les arcs (après un chargement complet depuis un fichier).

Source code in src/saferoute/familiarity.py 
70
71
72
def mark_all_pending(self) -> None:
    """Force le renvoi de tous les arcs (après un chargement complet depuis un fichier)."""
    self._dirty.update(self.scores)

to_edge_dict 

to_edge_dict() -> dict

Retourne {(u, v, key): score} pour injection dans le graphe.

Source code in src/saferoute/familiarity.py 
74
75
76
def to_edge_dict(self) -> dict:
    """Retourne {(u, v, key): score} pour injection dans le graphe."""
    return dict(self.scores)