up

counting_osm_objects/update.py

@@ -0,0 +1,194 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+"""
+Script pour mettre à jour le fichier historisé france internal.
+
+Ce script utilise osmupdate pour mettre à jour le fichier france-internal.osh.pbf
+avec les dernières modifications d'OpenStreetMap.
+
+Usage:
+    python update.py [--verbose]
+"""
+
+import os
+import sys
+import argparse
+import subprocess
+import logging
+from datetime import datetime
+
+# Chemin vers le répertoire du script
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+
+# Chemin vers le fichier historisé france internal
+FRANCE_INTERNAL_FILE = os.path.join(SCRIPT_DIR, "osm_data", "france-internal.osh.pbf")
+
+# Chemin vers le répertoire temporaire pour osmupdate
+TEMP_DIR = os.path.join(SCRIPT_DIR, "update_temp")
+
+# Configurer le logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+logger = logging.getLogger(__name__)
+
+
+def run_command(command, verbose=False):
+    """
+    Exécute une commande shell et retourne la sortie.
+    
+    Args:
+        command (str): Commande à exécuter
+        verbose (bool): Si True, affiche la sortie de la commande en temps réel
+        
+    Returns:
+        tuple: (code de retour, sortie standard, sortie d'erreur)
+    """
+    logger.info(f"Exécution: {command}")
+    
+    if verbose:
+        # Exécuter la commande avec sortie en temps réel
+        process = subprocess.Popen(
+            command,
+            shell=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            universal_newlines=True,
+            bufsize=1
+        )
+        
+        stdout_lines = []
+        stderr_lines = []
+        
+        # Lire la sortie standard en temps réel
+        for line in process.stdout:
+            line = line.strip()
+            stdout_lines.append(line)
+            print(line)
+            
+        # Lire la sortie d'erreur en temps réel
+        for line in process.stderr:
+            line = line.strip()
+            stderr_lines.append(line)
+            print(f"ERREUR: {line}", file=sys.stderr)
+            
+        # Attendre la fin du processus
+        return_code = process.wait()
+        return return_code, "\n".join(stdout_lines), "\n".join(stderr_lines)
+    else:
+        # Exécuter la commande sans sortie en temps réel
+        try:
+            result = subprocess.run(
+                command,
+                shell=True,
+                check=False,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                universal_newlines=True,
+            )
+            return result.returncode, result.stdout, result.stderr
+        except Exception as e:
+            logger.error(f"Erreur lors de l'exécution de la commande: {e}")
+            return 1, "", str(e)
+
+
+def update_france_internal(verbose=False):
+    """
+    Met à jour le fichier historisé france internal avec osmupdate.
+    
+    Args:
+        verbose (bool): Si True, affiche la sortie de la commande en temps réel
+        
+    Returns:
+        bool: True si la mise à jour a réussi, False sinon
+    """
+    # Vérifier si le fichier existe
+    if not os.path.isfile(FRANCE_INTERNAL_FILE):
+        logger.error(f"Le fichier {FRANCE_INTERNAL_FILE} n'existe pas.")
+        logger.error("Veuillez télécharger le fichier initial depuis Geofabrik ou une autre source.")
+        return False
+    
+    # Créer le répertoire temporaire s'il n'existe pas
+    os.makedirs(TEMP_DIR, exist_ok=True)
+    
+    # Chemin vers le fichier mis à jour
+    updated_file = os.path.join(TEMP_DIR, "france-internal-updated.osh.pbf")
+    
+    # Construire la commande osmupdate
+    command = f"osmupdate --verbose --keep-tempfiles -t={TEMP_DIR}/temp {FRANCE_INTERNAL_FILE} {updated_file}"
+    
+    # Exécuter la commande
+    logger.info("Mise à jour du fichier france-internal.osh.pbf en cours...")
+    return_code, stdout, stderr = run_command(command, verbose)
+    
+    if return_code != 0:
+        logger.error(f"Erreur lors de la mise à jour: {stderr}")
+        return False
+    
+    # Remplacer l'ancien fichier par le nouveau
+    if os.path.isfile(updated_file):
+        # Créer une sauvegarde de l'ancien fichier
+        backup_file = f"{FRANCE_INTERNAL_FILE}.bak"
+        try:
+            os.rename(FRANCE_INTERNAL_FILE, backup_file)
+            logger.info(f"Sauvegarde de l'ancien fichier créée: {backup_file}")
+        except Exception as e:
+            logger.error(f"Erreur lors de la création de la sauvegarde: {e}")
+            return False
+        
+        # Déplacer le nouveau fichier
+        try:
+            os.rename(updated_file, FRANCE_INTERNAL_FILE)
+            logger.info(f"Fichier mis à jour avec succès: {FRANCE_INTERNAL_FILE}")
+            return True
+        except Exception as e:
+            logger.error(f"Erreur lors du déplacement du fichier mis à jour: {e}")
+            # Restaurer l'ancien fichier en cas d'erreur
+            try:
+                os.rename(backup_file, FRANCE_INTERNAL_FILE)
+                logger.info("Restauration de l'ancien fichier réussie.")
+            except Exception as e2:
+                logger.error(f"Erreur lors de la restauration de l'ancien fichier: {e2}")
+            return False
+    else:
+        logger.error(f"Le fichier mis à jour {updated_file} n'a pas été créé.")
+        return False
+
+
+def main():
+    """Fonction principale"""
+    parser = argparse.ArgumentParser(
+        description="Met à jour le fichier historisé france internal avec osmupdate."
+    )
+    parser.add_argument(
+        "--verbose", "-v", action="store_true", help="Affiche la sortie des commandes en temps réel"
+    )
+    
+    args = parser.parse_args()
+    
+    # Afficher l'heure de début
+    start_time = datetime.now()
+    logger.info(f"Début de la mise à jour: {start_time.strftime('%Y-%m-%d %H:%M:%S')}")
+    
+    # Mettre à jour le fichier
+    success = update_france_internal(args.verbose)
+    
+    # Afficher l'heure de fin et la durée
+    end_time = datetime.now()
+    duration = end_time - start_time
+    logger.info(f"Fin de la mise à jour: {end_time.strftime('%Y-%m-%d %H:%M:%S')}")
+    logger.info(f"Durée totale: {duration}")
+    
+    if success:
+        logger.info("Mise à jour terminée avec succès.")
+        return 0
+    else:
+        logger.error("Échec de la mise à jour.")
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())