class GraphCache:
"""
Gestionnaire de cache pour les graphes OSMnx et les données de criminalité.
Usage :
cache = GraphCache()
G = cache.load_graph("london") or cache.save_graph("london", G)
crimes = cache.load_crimes("london") or cache.save_crimes("london", crimes)
"""
def __init__(self, cache_dir: Path | str | None = None):
self.cache_dir = Path(cache_dir) if cache_dir else _default_cache_dir()
self.cache_dir.mkdir(parents=True, exist_ok=True)
logger.info(f"Cache initialisé dans : {self.cache_dir}")
# ── Graphe OSMnx ──────────────────────────────────────────────────────────
def graph_path(self, city_key: str) -> Path:
"""Retourne le chemin du fichier .graphml pour une ville."""
return self.cache_dir / f"{city_key}_graph.graphml"
def meta_path(self, city_key: str) -> Path:
return self.cache_dir / f"{city_key}_meta.json"
def has_graph(self, city_key: str) -> bool:
return self.graph_path(city_key).exists()
def save_graph(self, city_key: str, G: Any, stats: dict | None = None) -> None:
"""
Sauvegarde un graphe NetworkX en .graphml + métadonnées JSON.
Args:
city_key: identifiant court ex. "london", "cape_town"
G: graphe NetworkX (MultiDiGraph OSMnx)
stats: dict optionnel de statistiques de validation
"""
path = self.graph_path(city_key)
ox.save_graphml(G, filepath=str(path))
meta = {
"city_key": city_key,
"downloaded_at": datetime.now(timezone.utc).isoformat(),
"node_count": G.number_of_nodes(),
"edge_count": G.number_of_edges(),
"stats": stats or {},
}
self.meta_path(city_key).write_text(json.dumps(meta, indent=2), encoding="utf-8")
logger.info(
f"Graphe '{city_key}' sauvegardé : "
f"{meta['node_count']} nœuds, {meta['edge_count']} arcs → {path}"
)
def load_graph(self, city_key: str):
"""
Charge un graphe depuis le cache .graphml.
Retourne None si le cache n'existe pas.
"""
path = self.graph_path(city_key)
if not path.exists():
return None
logger.info(f"Chargement graphe '{city_key}' depuis le cache...")
try:
G = ox.load_graphml(filepath=str(path))
except (MemoryError, OSError):
# Saturation mémoire ou erreur d'E/S : le fichier n'est PAS corrompu. Avant 0.1.2, TOUTE
# exception supprimait le cache — un simple manque de RAM détruisait un graphe de Londres
# (527 Mo) qui demande des minutes de téléchargement. On laisse l'erreur remonter, intacte.
raise
except Exception as e:
# Corruption avérée (XML invalide, attributs manquants…). On met le fichier de côté au lieu
# de le supprimer : il reste récupérable et inspectable.
quarantined = path.with_name(path.name + ".corrupt")
logger.error(f"Cache corrompu pour '{city_key}' : {e} — mis de côté : {quarantined.name}")
path.replace(quarantined)
self.meta_path(city_key).unlink(missing_ok=True)
raise CacheCorruptionError(path, str(e)) from e
meta = self._load_meta(city_key)
if meta:
logger.info(
f"Cache '{city_key}' : {meta.get('node_count')} nœuds, "
f"{meta.get('edge_count')} arcs (téléchargé le {meta.get('downloaded_at', '?')})"
)
return G
def _load_meta(self, city_key: str) -> dict | None:
path = self.meta_path(city_key)
if not path.exists():
return None
try:
meta: dict = json.loads(path.read_text(encoding="utf-8"))
return meta
except Exception:
return None
# ── Données de criminalité ─────────────────────────────────────────────────
def crimes_path(self, city_key: str) -> Path:
return self.cache_dir / f"{city_key}_crimes.csv.gz"
def has_crimes(self, city_key: str) -> bool:
return self.crimes_path(city_key).exists()
def save_crimes(self, city_key: str, crimes: list[dict]) -> None:
"""
Sauvegarde une liste de crimes en CSV compressé gzip.
Chaque entrée : {"lat": float, "lon": float, "weight": float}, plus "month" ("YYYY-MM")
si au moins un crime le porte (colonne optionnelle, nécessaire à l'évaluation temporelle).
"""
path = self.crimes_path(city_key)
with_month = any(c.get("month") for c in crimes)
fields = ["lat", "lon", "weight"] + (["month"] if with_month else [])
# Écriture ATOMIQUE : un processus interrompu en cours d'écriture laissait un .gz tronqué que
# le chargement lisait ensuite comme valide. On écrit un fichier temporaire, puis on renomme.
tmp = path.with_name(path.name + ".tmp")
with gzip.open(tmp, "wt", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
writer.writeheader()
writer.writerows(crimes)
os.replace(tmp, path)
logger.info(f"Crimes '{city_key}' sauvegardés : {len(crimes)} entrées → {path}")
def load_crimes(self, city_key: str) -> list[dict] | None:
"""
Charge les crimes depuis le cache CSV.gz.
Retourne None si le cache n'existe pas.
"""
path = self.crimes_path(city_key)
if not path.exists():
return None
crimes = []
with gzip.open(path, "rt", encoding="utf-8", newline="") as f:
reader = csv.DictReader(f)
for row in reader:
try:
entry: dict[str, float | str] = {
"lat": float(row["lat"]),
"lon": float(row["lon"]),
"weight": float(row["weight"]),
}
except (KeyError, ValueError):
continue
if row.get("month"):
entry["month"] = row["month"]
crimes.append(entry)
logger.info(f"Crimes '{city_key}' chargés depuis le cache : {len(crimes)} entrées")
return crimes
def cache_info(self) -> dict:
"""Retourne un résumé de l'état du cache."""
info = {}
for meta_file in self.cache_dir.glob("*_meta.json"):
city_key = meta_file.stem.replace("_meta", "")
meta = json.loads(meta_file.read_text(encoding="utf-8"))
crimes_path = self.crimes_path(city_key)
info[city_key] = {
"graph": meta,
"crimes_cached": crimes_path.exists(),
"crimes_size_kb": round(crimes_path.stat().st_size / 1024, 1)
if crimes_path.exists() else 0,
}
return info