Ajout d'une procédure pour l'import automatique de lames dans Geotrek-admin

[bruhnild]

Description :

Actuellement, il n'existe pas de procédure permettant l'import automatique de lames dans Geotrek-admin. La procédure décrite dans la documentation officielle (documentation import) ne couvre que l'import des poteaux, mais ne permet pas l'import des lames (blades).

Le seul moyen d'importer des lames est de réaliser un script spécifique. Ce processus devrait être amélioré pour permettre un import automatique, au même titre que celui des poteaux.

Côté Makina Corpus nous avons déjà eu l'occasion de réaliser des scripts d'import pour ces données. Cela pourrait être un point de départ pour un territoire qui souhaiterai importer des données manuellement en attendant une commande d'import packagée directement avec Geotrek.

Le code suivant doit être inséré dans un fichier python pour ensuite être lancé. Une fois lancé il cherche un fichier csv contenant les lames, fait l'import et créé un second fichier avec les lames correctement importées (permettant ainsi de comparer les deux fichiers csv pour savoir quelle lame a été importée ou non) :

from ctypes import sizeof
from dataclasses import replace
from gettext import textdomain
import json
from django.contrib.gis.geos import Point
from geotrek.authent.models import Structure
from geotrek.core.models import Topology
from geotrek.signage.models import Blade, BladeType, Color, Direction, Line, Sealing, Signage, SignageType, LinePictogram
import csv
from psycopg2.errors import UniqueViolation
from django.db.utils import IntegrityError


##################################
structure = Structure.objects.get(name='Default')  # < Remplacer ici par la nom de la structure de rattachement des lames
provider = 'ORIGINE DES DONNEES'   # < A modifier
f = open('/opt/geotrek-admin/var/conf/lames.csv', 'r')   # < Chemin vers le fichier importé
f_out = open('/opt/geotrek-admin/var/conf/lames_importees.csv', 'w')  # < Chemin vers le second fichier créé
###############################

color_mapping = {
    'V': 'Vert',
    'J': 'Jaune',
    'G': 'Gris',
    'M': 'Marron',
    'B': 'Blanc',
    'GB': 'Gris',
    'Bois': 'Marron'
}


def create_or_update_line(blade, number, text, picto, alt):
    if number or text or picto or alt:
        lines = Line.objects.filter(blade=blade, number=number)
        line = None
        if lines.exists():
            if lines.count() == 1:
                line = lines.first()
                line.text = text
                line.pictogram = picto
                line.save()
        else:
            line, _ = Line.objects.get_or_create(
                blade=blade,
                number=number,
                text=text,
            )
            if picto and picto != "":
                line_pictogram, _ = LinePictogram.objects.get_or_create(label=picto)
                line.pictograms.add(line_pictogram)
        if alt:
            line.distance = float(alt)
            line.save()


def create_bl_blade(signa, color_str):
    blade_type, _ =BladeType.objects.get_or_create(label='Bague de lieudit (BL)')

    blade, created = Blade.objects.get_or_create(
        topology=signa.topo_object,
        signage=signa,
        number="-BL",
        type=blade_type,
    )
    if created:
        handled_blade, handled_pk = "created", blade.pk
    else:
        handled_blade, handled_pk = "updated", blade.pk
    blade.color = Color.objects.get(label=color_mapping[color_str])
    blade.save()
    create_or_update_line(blade, 1, text=row[3], picto="", alt="")
    return handled_blade, handled_pk


def update_blade(signa, blade_code, row, blade=None):
    type_str = row[1]
    color_str = row[2]
    direction_str = row[6]
    handled_blade = False
    type_mapping = {
        'BL': 'Bague de lieudit (BL)',
        'LD1': 'Lame LD1',
        'LD2': 'Lame LD2',
        'LD3': 'Lame LD3',
        'LDS1': 'Lame LD1',
        'LDS2': 'Lame LD2',
        'LDS3': 'Lame LD3',
        'LT1': 'Lame LT1',
        'LT2': 'Lame LT2',
        'LT3': 'Lame LT3',
        'BLR': 'Bague de lieudit réduite (BLr)',
        'LDR1': 'Lame LD1 réduite',
        'LDR2': 'Lame LD2 réduite',
        'LDR3': 'Lame LD3 réduite',
        'LTR1': 'Lame LT1 réduite',
        'LTR2': 'Lame LT2 réduite',
        'LTR3': 'Lame LT3 réduite',
        'PB-PR': 'Plaque Balisage PR (PB-PR)',
        'PB-GR': 'Plaque Balisage - GR (PB-GR)',
        'H22': 'Signalétique routière',
        'H23': 'Signalétique routière',
        'TR': 'Triangle Réglementaire (TR)',
        'PAR': "Plaque d'Avertissement (PLA)",
        'PLR': "Plaque Réglementaire (PLR)",
    }
    if type_str == 'PPI' or type_str == 'BL' and row[5].strip(" ").strip("m"):
        if row[5]:
            signa.printed_elevation = int(row[5].strip(" ").strip("m"))
            signa.save()
        handled_blade, handled_pk = create_bl_blade(signa, color_str=row[2])
        return (handled_blade, handled_pk)
    else:
        if type_str == 'PB':
            replace_with = row[3].split('-')
            if 'Changement parcours' in replace_with:
                type_str = "PB-PR"
            else:
                type_str = f"{replace_with[0]}-{replace_with[1]}"
        elif type_str == 'JB' or not type_str:
            return ("No type - skipped", 0)

        try:
            type_obj = BladeType.objects.get(label=type_mapping[type_str])
        except:
            return ("No type - skipped", 0)
        color = None
        if color_str:
            color = Color.objects.get(label=color_mapping[color_str])
        direction = None
        if direction_str:
            if direction_str == "CO" or direction_str == "H":
                direction_str = "Continuité"
            direction = Direction.objects.get(label=direction_str)
        if not blade:
            blade = Blade.objects.create(
                topology=signa.topo_object,
                signage=signa,
                number=blade_code,
                type=type_obj,
                color=color,
                direction=direction
            )
            # print(f"Blade {blade.pk} created")
            handled_blade, handled_pk = "created", blade.pk
        else:
            blade.type = type_obj
            blade.color = color
            blade.direction = direction
            blade.save()
            handled_blade, handled_pk = "updated", blade.pk
        row[5] = row[5].replace(',', '.')
        create_or_update_line(blade, 1, text=row[3], picto=row[4], alt=row[5].strip(" ").strip("m"))
        if row[7]:
            row[9] = row[9].replace(',', '.')
            create_or_update_line(blade, 2, text=row[7], picto=row[8], alt=row[9].strip(" ").strip("m"))
        if row[10]:
            row[12] = row[12].replace(',', '.')
            create_or_update_line(blade, 3, text=row[10], picto=row[11], alt=row[12].strip(" ").strip("m"))
        return handled_blade, handled_pk


csvreader = csv.reader(f)
output_writer = csv.writer(f_out, delimiter=',', quotechar=',', quoting=csv.QUOTE_MINIMAL)
i = 0
dw = None
fail = False
handled_blade = False
handled_pk = 0
for row in csvreader:
    if i < 2:
        pass

    # Code	Type	Coul	Texte Ligne 1	Picto	Alt/Dist	Sens	Texte Ligne 2
    # 0     1       2       3               4       5           6           7       
    # Picto	Alt/Dist	Texte Ligne 3	Picto	Alt/Dist
    # 8     9                10           11        12 
    else:
        handled_blade = False
        handled_pk = 0
        # print(row)
        try:
            signa_code = None
            blade_code = None
            status_str = row[1]
            code_str = row[0]
            slashes_count = code_str.count("/")
            # print(f"{code_str=}")

            if slashes_count == 2:
                signa_code = f"{code_str.split('/')[0]}/{code_str.split('/')[1]}"
                blade_code = f"{code_str.split('/')[2]}"
            elif slashes_count == 1:
                signa_code = code_str
            else:
                continue
            signa = Signage.objects.filter(deleted=False,
             code=signa_code)
            if signa.count() == 0:
                signa = Signage.objects.filter(deleted=False, code=signa_code.replace('/', '_'))
            if signa.count() == 0:
                row.append("No signage")
                continue
            if signa.count() > 1:
                row.append("Several signages")
                continue
            else:
                signa = signa.first()
            if blade_code:
                if blade_code == "BL":
                    blade_code = "-BL"
                blade = Blade.objects.filter(deleted=False, number=blade_code, signage=signa)
                if blade.count() == 0:
                    handled_blade, handled_pk = update_blade(signa, blade_code, row)
                elif blade.count() > 1:
                    continue
                elif blade.count() == 1:
                    blade = blade.first()
                    handled_blade, handled_pk = update_blade(signa, blade_code, row, blade)
        except UniqueViolation as e:
            row.append(str(e).replace(',', '').replace('\n', ' '))
        except IntegrityError as e:
            row.append(str(e).replace(',', '').replace('\n', ' '))
        if handled_blade and handled_pk:
            while len(row) < 12:
                row.append('')
            row.append(handled_blade)
            row.append(handled_pk)
        output_writer.writerow(row)
    i += 1

f.close()

Ce fichier doit ensuite être lancé dans une console Geotrek via la commande suivante : sudo geotrek shell < mon_script.py

Le fichier csv qui sert pour l'import doit être formaté de la manière suivante :

Code,Type,Coul,Texte Ligne 1,Picto,Alt/Dist,Sens,Texte Ligne 2,Picto,Alt/Dist,Texte Ligne 3,Picto,Alt/Dist,Charte SEN,Quantité
07100/01/A,BL,V,Les Vernèdes,,873,,,,,,,,Charte CD30,1
30153/02/A,BL,V,Malôns et Elze,,860,,,,,,,,Charte CD30,2
30153/03/A,BL,V,Le Frontal,,596,,,,,,,,Charte CD30,3

Attention il faut pour que cela fonctionne que le code de la lame référence le code du poteau signalétique correspondant.