''' 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 from path import Path # @UnusedImport from core import logconf from core.model import Sql from core.pde import WincanDb, QGisRegard, QGisChantier, QGisTroncon, \ ControlesDb, CSigDb, SRID import qgis_sync_wincan logger = logging.getLogger("qgis_sync_video") logconf.start("qgis_sync_video", logging.DEBUG) # # POUR TESTER, décommenter les lignes suivantes ##----------------------------------------------- # ControlesDb._path = Path(r"\\h2o\local\4-transversal\BDD\mdb_test\cg67Parc_data.mdb") # WincanDb._path = Path(r"\\h2o\local\4-transversal\BDD\mdb_test\Wincan\parc_2007\DB\PARC_2007.mdb") # CSigDb.server = "TR-POSTGIS-02" # CSigDb.pwd = "Am5VOMkdFHU7WwrfVOs9" # logger.handlers = [h for h in logger.handlers if (type(h) == logging.StreamHandler)] # logger.warning("<<<<<<<<<<<<<< Mode TEST >>>>>>>>>>>>>>>>>") ##----------------------------------------------- def main(): # 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.read(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: 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-%s (%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 (%s)", regard.number, essai.SI_InspName) chantier.items.append(troncon) if not chantiers: logger.info("Aucun nouveaux chantiers") return 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.read_all(Sql.format("SELECT id FROM t_chantiers WHERE nom = {:text} AND id_type_chantier=1", 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)) logger.info("Le chantier existant a été supprimé") 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]) - 5 chantier.x1 = max([regard.x for regard in regards]) + 5 chantier.y0 = min([regard.y for regard in regards]) - 5 chantier.y1 = max([regard.y for regard in regards]) + 5 # Créé le chantier logger.info("> Création du chantier") row = csig_db.first(Sql.format("""INSERT INTO t_chantiers(id_type_chantier, numero, nom, archive, geom) VALUES ({chantier_type}, {number}, {name:text}, False, ST_GeomFromText('POLYGON(({x0} {y0}, {x0} {y1}, {x1} {y1}, {x1} {y0}, {x0} {y0}))', {srid})) RETURNING id """, chantier_type=1, number=chantier.number, name=chantier.name, x0=chantier.x0, x1=chantier.x1, y0=chantier.y0, y1=chantier.y1, srid=SRID ) ) # 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:text}, {s_autonumber}, ST_GeomFromText('POINT({x} {y})', {srid}), {label:text}, False) RETURNING id """, name=regard.number, chantier_id=chantier.pgid, res_ce=regard.res_ce, s_autonumber=troncon.s_autonumber, x=regard.x, y=regard.y, srid=SRID, label=label) ) regard.pgid = row.id logger.debug("> Création du tronçon {}".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:text}, {res_ce:text}, ST_GeomFromText('LINESTRING ({x1} {y1}, {x2} {y2})', {srid}), False) 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.res_itv, res_ce=troncon.res_ce, x1=troncon.r1.x, y1=troncon.r1.y, x2=troncon.r2.x, y2=troncon.r2.y, srid=SRID ) ) 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)) wincan_db.commit() logger.info("Chantier créé") if __name__ == "__main__": # lance la synchro compactage avant qgis_sync_wincan.main() main() logger.info("-- Fin --")