301 lines
9 KiB
Python
Executable file
301 lines
9 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
Script principal pour lancer l'analyse historique d'une ville.
|
|
|
|
Ce script:
|
|
1. Demande à l'utilisateur quelle ville il souhaite traiter
|
|
2. Trouve le code INSEE de la ville demandée
|
|
3. Vérifie si le polygone de la ville existe, sinon le récupère
|
|
4. Traite les données historiques OSM pour cette ville
|
|
|
|
Usage:
|
|
python historize_zone.py [--input fichier_historique.osh.pbf]
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import csv
|
|
import argparse
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
# Chemin vers le répertoire du script
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
# Chemin vers le fichier CSV contenant les données des villes
|
|
CITIES_CSV = os.path.join(SCRIPT_DIR, "osm-commerces-villes-export.csv")
|
|
|
|
# Chemin vers le répertoire des polygones
|
|
POLYGONS_DIR = os.path.join(SCRIPT_DIR, "polygons")
|
|
|
|
# Chemin par défaut pour le fichier d'historique OSM France
|
|
DEFAULT_HISTORY_FILE = os.path.join(SCRIPT_DIR, "osm_data", "france-internal.osh.pbf")
|
|
|
|
|
|
def run_command(command):
|
|
"""Exécute une commande shell et retourne la sortie"""
|
|
print(f"Exécution: {command}")
|
|
try:
|
|
result = subprocess.run(
|
|
command,
|
|
shell=True,
|
|
check=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
)
|
|
return result.stdout
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"Erreur lors de l'exécution de la commande: {e}")
|
|
print(f"Sortie de la commande: {e.stdout}")
|
|
print(f"Erreur de la commande: {e.stderr}")
|
|
return None
|
|
|
|
|
|
def load_cities():
|
|
"""
|
|
Charge les données des villes depuis le fichier CSV.
|
|
|
|
Returns:
|
|
dict: Dictionnaire des villes avec le nom comme clé et les données comme valeur
|
|
"""
|
|
cities = {}
|
|
try:
|
|
with open(CITIES_CSV, "r", encoding="utf-8") as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
if row.get("name") and row.get("zone"):
|
|
cities[row["name"].lower()] = row
|
|
except Exception as e:
|
|
print(f"Erreur lors du chargement du fichier CSV des villes: {e}")
|
|
sys.exit(1)
|
|
|
|
return cities
|
|
|
|
|
|
def find_city(city_name, cities):
|
|
"""
|
|
Recherche une ville par son nom dans le dictionnaire des villes.
|
|
|
|
Args:
|
|
city_name (str): Nom de la ville à rechercher
|
|
cities (dict): Dictionnaire des villes
|
|
|
|
Returns:
|
|
dict: Données de la ville si trouvée, None sinon
|
|
"""
|
|
# Recherche exacte
|
|
if city_name.lower() in cities:
|
|
return cities[city_name.lower()]
|
|
|
|
# Recherche partielle
|
|
matches = []
|
|
for name, data in cities.items():
|
|
if city_name.lower() in name:
|
|
matches.append(data)
|
|
|
|
if not matches:
|
|
return None
|
|
|
|
# Si plusieurs correspondances, demander à l'utilisateur de choisir
|
|
if len(matches) > 1:
|
|
print(f"Plusieurs villes correspondent à '{city_name}':")
|
|
for i, city in enumerate(matches):
|
|
print(f"{i+1}. {city['name']} (INSEE: {city['zone']})")
|
|
|
|
choice = input("Entrez le numéro de la ville souhaitée (ou 'q' pour quitter): ")
|
|
if choice.lower() == "q":
|
|
sys.exit(0)
|
|
|
|
try:
|
|
index = int(choice) - 1
|
|
if 0 <= index < len(matches):
|
|
return matches[index]
|
|
else:
|
|
print("Choix invalide.")
|
|
return None
|
|
except ValueError:
|
|
print("Veuillez entrer un numéro valide.")
|
|
return None
|
|
|
|
return matches[0]
|
|
|
|
|
|
def check_polygon_exists(insee_code):
|
|
"""
|
|
Vérifie si le polygone d'une commune existe déjà.
|
|
|
|
Args:
|
|
insee_code (str): Code INSEE de la commune
|
|
|
|
Returns:
|
|
str: Chemin vers le fichier polygone s'il existe, None sinon
|
|
"""
|
|
poly_file = os.path.join(POLYGONS_DIR, f"commune_{insee_code}.poly")
|
|
if os.path.isfile(poly_file):
|
|
return poly_file
|
|
return None
|
|
|
|
|
|
def get_polygon(insee_code):
|
|
"""
|
|
Récupère le polygone d'une commune à partir de son code INSEE.
|
|
|
|
Args:
|
|
insee_code (str): Code INSEE de la commune
|
|
|
|
Returns:
|
|
str: Chemin vers le fichier polygone créé, None en cas d'erreur
|
|
"""
|
|
get_poly_script = os.path.join(SCRIPT_DIR, "get_poly.py")
|
|
command = f"python3 {get_poly_script} {insee_code}"
|
|
|
|
output = run_command(command)
|
|
if output:
|
|
# Vérifier si le polygone a été créé
|
|
poly_file = check_polygon_exists(insee_code)
|
|
if poly_file:
|
|
return poly_file
|
|
|
|
return None
|
|
|
|
|
|
def process_city_history(input_file, poly_file, cleanup=False, benchmark=False):
|
|
"""
|
|
Traite l'historique OSM pour une ville.
|
|
|
|
Args:
|
|
input_file (str): Chemin vers le fichier d'historique OSM
|
|
poly_file (str): Chemin vers le fichier polygone de la ville
|
|
cleanup (bool): Si True, nettoie les fichiers temporaires après traitement
|
|
benchmark (bool): Si True, affiche des informations de performance détaillées
|
|
|
|
Returns:
|
|
bool: True si le traitement a réussi, False sinon
|
|
"""
|
|
loop_script = os.path.join(
|
|
SCRIPT_DIR, "loop_thematics_history_in_zone_to_counts.py"
|
|
)
|
|
output_dir = os.path.join(SCRIPT_DIR, "test_results")
|
|
temp_dir = os.path.join(SCRIPT_DIR, "test_temp")
|
|
|
|
# Créer les répertoires de sortie si nécessaires
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
os.makedirs(temp_dir, exist_ok=True)
|
|
|
|
# Construire la commande avec les options supplémentaires
|
|
command = f"python3 {loop_script} --input {input_file} --poly {poly_file} --output-dir {output_dir} --temp-dir {temp_dir} --max-dates 100"
|
|
|
|
# Ajouter les options de nettoyage et de benchmark si activées
|
|
if cleanup:
|
|
command += " --cleanup"
|
|
if benchmark:
|
|
command += " --benchmark"
|
|
|
|
print(f"Exécution de la commande: {command}")
|
|
output = run_command(command)
|
|
if output is not None:
|
|
return True
|
|
return False
|
|
|
|
|
|
def main():
|
|
"""Fonction principale"""
|
|
parser = argparse.ArgumentParser(
|
|
description="Analyse historique d'une ville dans OpenStreetMap."
|
|
)
|
|
parser.add_argument(
|
|
"--input",
|
|
"-i",
|
|
default=DEFAULT_HISTORY_FILE,
|
|
help=f"Fichier d'historique OSM (.osh.pbf). Par défaut: {DEFAULT_HISTORY_FILE}",
|
|
)
|
|
parser.add_argument(
|
|
"--cleanup",
|
|
"-c",
|
|
action="store_true",
|
|
help="Nettoyer les fichiers temporaires après traitement",
|
|
)
|
|
parser.add_argument(
|
|
"--benchmark",
|
|
"-b",
|
|
action="store_true",
|
|
help="Afficher des informations de performance détaillées",
|
|
)
|
|
parser.add_argument(
|
|
"--city",
|
|
"-v",
|
|
help="Nom de la ville à traiter (si non spécifié, demande interactive)",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Vérifier que le fichier d'historique existe
|
|
if not os.path.isfile(args.input):
|
|
print(f"Erreur: Le fichier d'historique {args.input} n'existe pas.")
|
|
print(
|
|
f"Veuillez spécifier un fichier d'historique valide avec l'option --input."
|
|
)
|
|
sys.exit(1)
|
|
|
|
# Charger les données des villes
|
|
cities = load_cities()
|
|
if not cities:
|
|
print("Aucune ville n'a été trouvée dans le fichier CSV.")
|
|
sys.exit(1)
|
|
|
|
print(f"Données chargées pour {len(cities)} villes.")
|
|
|
|
# Obtenir le nom de la ville à traiter
|
|
city_name = args.city
|
|
if not city_name:
|
|
# Mode interactif si aucune ville n'est spécifiée
|
|
city_name = input("Quelle ville souhaitez-vous traiter ? ")
|
|
|
|
# Rechercher la ville
|
|
city = find_city(city_name, cities)
|
|
if not city:
|
|
print(f"Aucune ville correspondant à '{city_name}' n'a été trouvée.")
|
|
sys.exit(1)
|
|
|
|
insee_code = city["zone"]
|
|
print(f"Ville trouvée: {city['name']} (INSEE: {insee_code})")
|
|
|
|
# Vérifier si le polygone existe
|
|
poly_file = check_polygon_exists(insee_code)
|
|
if poly_file:
|
|
print(f"Le polygone pour {city['name']} existe déjà: {poly_file}")
|
|
else:
|
|
print(f"Le polygone pour {city['name']} n'existe pas. Récupération en cours...")
|
|
poly_file = get_polygon(insee_code)
|
|
if not poly_file:
|
|
print(f"Erreur: Impossible de récupérer le polygone pour {city['name']}.")
|
|
sys.exit(1)
|
|
print(f"Polygone récupéré avec succès: {poly_file}")
|
|
|
|
# Afficher les options activées
|
|
if args.benchmark:
|
|
print("\n=== Options ===")
|
|
print(
|
|
f"Nettoyage des fichiers temporaires: {'Activé' if args.cleanup else 'Désactivé'}"
|
|
)
|
|
print(f"Benchmark: Activé")
|
|
print("===============\n")
|
|
|
|
# Traiter l'historique pour cette ville
|
|
print(f"Traitement de l'historique OSM pour {city['name']}...")
|
|
success = process_city_history(args.input, poly_file, args.cleanup, args.benchmark)
|
|
|
|
if success:
|
|
print(f"Traitement terminé avec succès pour {city['name']}.")
|
|
else:
|
|
print(f"Erreur lors du traitement de l'historique pour {city['name']}.")
|
|
sys.exit(1)
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|