Script synchro video complet (pas testé)

qgis_sync_video.py

@@ -0,0 +1,257 @@
+'''
+
+    Créé les chantiers, tronçons et regards correspondant aux contrôles Vidéo.
+
+    Les données des chantiers sont issues de la base Wincan
+    Les lignes (issues de la requête csig_sync_video) sont traitées si:
+    - SI_Spare3 est null
+    - les coordonnées des regards ont été mises à jour (non null)
+    Lorsqu'un chantier est correctement créé, la valeur SI_Spare3 est mise à jour à '1'
+
+@author: olivier.massot, mai 2018
+'''
+import logging
+import re
+import sys
+
+from path import Path
+
+from core import logconf
+from core.model import Sql
+from core.pde import WincanDb, QGisRegard, QGisChantier, QGisTroncon, \
+    ControlesDb, CSigDb, SRID
+
+
+logger = logging.getLogger("qgis_sync_wincan")
+logconf.start("qgis_sync_wincan", logging.DEBUG)
+debug = False
+
+
+# # POUR TESTER, décommenter les lignes suivantes
+##-----------------------------------------------
+
+ControlesDb._path = Path(r"\\h2o\local\4-transversal\BDD\mdb_test\cg67Parc_data.mdb")
+CSigDb.server = "TR-POSTGIS-02"
+WincanDb._path = Path(r"\\h2o\local\4-transversal\BDD\mdb_test\Wincan\parc_2007\DB\PARC_2007.mdb")
+logger.handlers = [h for h in logger.handlers if (type(h) == logging.StreamHandler)]
+logger.warning("<<<<<<<<<<<<<<   Mode TEST   >>>>>>>>>>>>>>>>>")
+
+##-----------------------------------------------
+
+# Connexion à CSig
+csig_db = CSigDb(autocommit=False)
+
+# Connexion à Wincan
+wincan_db = WincanDb(autocommit=False)
+
+# Connexion à Controles
+controles_db = ControlesDb(autocommit=False)
+
+
+logger.info("Chargement des données")
+qessais = wincan_db.execute(Sql.format("""SELECT S_AutoNumber, S_StartNode, S_EndNode, S_StartNodeCoord_X, S_StartNodeCoord_Y,
+                                         S_EndNodeCoord_X, S_EndNodeCoord_Y, SI_AutoNumber, SI_InspName
+                                         FROM csig_sync_video"""))
+chantiers = {}
+
+for essai in qessais:
+    chantier_name = "<inconnu>"
+
+    r1, r2 = QGisRegard(), QGisRegard()
+
+    r1.number = essai.S_StartNode
+    r1.x = essai.S_StartNodeCoord_X
+    r1.y = essai.S_StartNodeCoord_Y
+
+    r2.number = essai.S_EndNode
+    r2.x = essai.S_EndNodeCoord_X
+    r2.y = essai.S_EndNodeCoord_Y
+
+    if not essai.SI_AutoNumber:
+        logger.error("%s - Le SI_Autonumber du troncon est manquant", essai.SI_InspName)
+        continue
+
+    # Parse le nom du chantier
+    parsed = re.search(r"^(\d{5,6})(-S?\d{1,2})?[\s_]?(.*)$", essai.SI_InspName)
+    if not parsed:
+        logger.error("Nom de chantier illisible: '{}'".format(essai.SI_InspName))
+        continue
+
+    chantier_number = parsed.group(1)
+
+    # Créé un nouveau chantier si pas encore créé
+    try:
+        chantier = chantiers[chantier_number]
+    except KeyError:
+        chantier = QGisChantier()
+        chantier.name = essai.SI_InspName
+        chantier.number = chantier_number
+        chantiers[chantier_number] = chantier
+
+    # Créé un nouveau tronçon
+    troncon = QGisTroncon()
+    troncon.r1 = r1
+    troncon.r2 = r2
+    troncon.si_autonumber = essai.SI_AutoNumber
+    troncon.s_autonumber = essai.S_AutoNumber
+
+    # Importe le résultat de l'essai Vidéo depuis contrôle
+    row = controles_db.first(Sql.format("""SELECT DG FROM csig_itv_results
+                                             WHERE SI_AutoNumber={si_autonumber}
+                                             """, si_autonumber=essai.SI_AutoNumber))
+    if row:
+        troncon.res_itv = row.DG
+    else:
+        logger.error("Impossible de trouver le resultat ITV du tronçon {}-{} (%s)", r1.number, r2.number, essai.SI_InspName)
+
+    # Importe le résultat de l'essai d'Etanchéité du tronçon depuis contrôle
+    row = controles_db.first(Sql.format("""SELECT strResSigne FROM csig_etancheite_results
+                                   WHERE (strTrcRegard='{r1}-{r2}' OR strTrcRegard='{r2}-{r1}')
+                                        AND lngChantierId={lngChantierId}
+                                    """, r1=r1.number, r2=r2.number, lngChantierId=chantier_number))
+    if row:
+        troncon.res_ce = row.strResSigne
+    else:
+        logger.error("Impossible de trouver le resultat Etanchéité du tronçon {}-{} (%s)", r1.number, r2.number, essai.SI_InspName)
+
+
+    # Importe le résultat de l'essai d'Etanchéité des regards depuis contrôle
+    for regard in (troncon.r1, troncon.r2):
+
+        row = controles_db.first(Sql.format("""SELECT strResSigne FROM csig_etancheite_results
+                                                WHERE strTrcRegard={regard:text} AND lngChantierId={lngChantierId}
+                                                """, regard=regard.number, lngChantierId=chantier_number))
+        if row:
+            regard.res_ce = row.strResSigne
+        else:
+            logger.error("Impossible de trouver le resultat Etanchéité du regard {} (%s)", regard.number, essai.SI_InspName)
+
+    chantier.items.append(troncon)
+
+if not chantiers:
+    logger.info("Aucun nouveaux chantiers")
+    sys.exit(0)
+
+logger.info("{} chantiers chargés en mémoire".format(len(chantiers)))
+
+
+# Pour la génération des etiquettes
+_labbelled = []
+def label_for(id_chantier, regard_number):
+    """ Renvoie le contenu de l'etiquette pour le regard
+    Un regard est etiquetté seulement si c'est le premier du chantier portant ce numéro """
+    if not (id_chantier, regard_number) in _labbelled:
+        _labbelled.append((id_chantier, regard_number))
+        return regard_number
+    else:
+        return ""
+
+
+for chantier in chantiers.values():
+    logger.info("** Création du chantier: {}".format(chantier.name))
+
+    # Vérifie si le chantier existe déjà
+    existants = csig_db.readall(Sql.format("SELECT id FROM t_chantiers WHERE nom = {:text}", chantier.name))
+
+    cancel = False
+
+    for row in existants:
+        logger.info("Un chantier portant ce nom existe déjà dans ControlesSig {} (id {})".format(chantier.name, row.id))
+
+        replace = input("Voulez-vous le remplacer? (o/n)")
+        if replace:
+            # Supprime l'ancien chantier
+            csig_db.execute(Sql.format("""DELETE FROM t_troncons WHERE id_chantier = {}""", row.id))
+            csig_db.execute(Sql.format("""DELETE FROM t_regards WHERE id_chantier = {}""", row.id))
+            csig_db.execute(Sql.format("""DELETE FROM t_chantiers WHERE id = {}""", row.id))
+        else:
+            # pass the chantier, do not register any error
+            logger.warning("Import du chantier annulé par l'utilisateur")
+            cancel = True
+
+    if cancel:
+        continue
+
+    # Calcule l'emprise du chantier
+    regards = set([troncon.r1 for troncon in chantier.items]) | set([troncon.r2 for troncon in chantier.items])
+    chantier.x0 = min([regard.x for regard in regards])
+    chantier.x1 = max([regard.x for regard in regards])
+    chantier.y0 = min([regard.y for regard in regards])
+    chantier.y1 = max([regard.y for regard in regards])
+
+    # Créé le chantier
+    logger.info("> Création du chantier")
+    row = csig_db.first(Sql.format("""INSERT INTO t_chantiers(id_type_chantier, numero, nom, geom, archive)
+                                       VALUES ({chantier_type}, {number}, {name:text}, {geom}, {archive})
+                                       RETURNING id
+                                        """, chantier_type=1,
+                                            number=chantier.number,
+                                            name=chantier.name,
+                                            geom="ST_GeomFromText('POLYGON(({x0} {y0}, {x0} {y1}, {x1} {y1}, \
+                                            {x1} {y0}, {x0} {y0}))', {srid})".format(x0=chantier.x0,
+                                                                                     x1=chantier.x1,
+                                                                                     y0=chantier.y0,
+                                                                                     y1=chantier.y1,
+                                                                                     srid=SRID),
+                                            archive="FALSE"
+                                            )
+                       )
+    # Récupère son id postgis
+    chantier.pgid = row.id
+
+    # Créé les regards, puis le tronçon
+    for troncon in chantier.items:
+
+        for regard in (troncon.r1, troncon.r2):
+            logger.info("> Création du regard %s", regard.number)
+
+            label = label_for(chantier.pgid, regard.number)
+
+            row = csig_db.first(Sql.format("""INSERT INTO t_regards(nom, id_chantier, res_ce, s_autonumber, geom, label, archive)
+                                           VALUES ({name:text}, {chantier_id}, {res_ce}, {s_autonumber}, {geom}, {label:text}, {archive})
+                                           RETURNING id
+                                           """.format(
+                                                   name=regard.number,
+                                                   chantier_id=chantier.pgid,
+                                                   res_ce=regard.res_ce,
+                                                   s_autonumber=troncon.s_autonumber,
+                                                   geom="ST_GeomFromText('POINT({x} {y})', {srid})".format(x=regard.x, y=regard.y, srid=SRID),
+                                                   label=label,
+                                                   archive="FALSE"
+                                               )
+                                           ))
+            regard.pgid = row.id
+
+        logger.debug("> Création du tronçon %s".format("{}-{}".format(troncon.r1.number, troncon.r2.number)))
+
+        csig_db.first(Sql.format("""INSERT INTO t_troncons(nom, id_chantier, id_regard_depart, id_regard_fin, s_autonumber, si_autonumber, res_itv, res_ce, geom, archive)
+                                   VALUES ({name:text, {chantier_id}, {id_regard_depart}, {id_regard_fin}, {s_autonumber}, {si_autonumber}, {res_itv}, {res_ce}, {geom}, {archive})
+                                   RETURNING id
+                                 """, name="{}-{}".format(troncon.r1.number, troncon.r2.number),
+                                    chantier_id=chantier.pgid,
+                                    id_regard_depart=troncon.r1.pgid,
+                                    id_regard_fin=troncon.r2.pgid,
+                                    s_autonumber=troncon.s_autonumber,
+                                    si_autonumber=troncon.si_autonumber,
+                                    res_itv=troncon.sql_res_itv(),
+                                    res_ce=troncon.sql_res_ce(),
+                                    geom="ST_GeomFromText('LINESTRING ({x1} {y1}, {x2} {y2})', {srid})".format(x1=troncon.r1.x,
+                                                                                                               y1=troncon.r1.y,
+                                                                                                               x2=troncon.r2.x,
+                                                                                                               y2=troncon.r2.y,
+                                                                                                               srid=SRID),
+                                    archive="FALSE"
+                                )
+                       )
+
+
+    csig_db.commit()
+
+
+    # update si_spare3 in Wincan to mark the chantier as imported
+    logger.info("> Marque le chantier comme importé dans WincanDb")
+
+    wincan_db.execute(Sql.format("UPDATE SI_T SET SI_T.SI_Spare3 = '1' \
+                                  WHERE (((SI_T.SI_InspName) Like '{}%'));", chantier.number))
+
+    logger.info("Chantier créé")