osm-commerces/counting_osm_objects/update.py

#!/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())