scripts-pde/core/pde.py

'''
    Configuration des scripts
'''

from datetime import datetime

from path import Path

from core.db import AccessSDb, PostgresDb
from core.model import Model


MAIN = Path(__file__).parent.parent.abspath()

# Web url of the WsPde web service
# PDE_WS_URL = r"http://localhost:2890/public/WsPDE.asmx"
PDE_WS_URL = r"http://webservices.bas-rhin.fr/CG67.AstreGF.WebServices/public/WsPDE.asmx"

# Chemin d'acces au fichier de secu Access, et login/mdp
MDW_PATH = r"\\h2o\local\4-transversal\BDD\mda\cg67Parc.mdw"
UID = "olivier"
PWD = "massot"

# Connexion à la base postgres ControlesSig
CSIG_SERVER = "TR-POSTGIS-02"
CSIG_DB = "ControlesSIG"
CSIG_PORT = "5432"
CSIG_USER = "ControlesSIG_userrw"
CSIG_PWD = "Am5VOMkdFHU7WwrfVOs9"

# Dirs
WRKDIR = MAIN / "work"
RSCDIR = MAIN / "resources"
DEFAULT_LOG_DIR = MAIN / "log"
LOG_CONFIG_FILE = MAIN / "logging.yaml"


PDA_FILES_DEST = Path(r"\\h2o\LOCAL\4-transversal\BDD\mdb\PDA\Fichiers_PDA")

FACTURES_DB_PATH = Path(r"\\h2o\local\4-transversal\BDD\mdb\Facture_data.mdb")
CONTROLES_DB_PATH = Path(r"\\h2o\local\4-transversal\BDD\mdb\cg67Parc_data.mdb")
WINCAN_DB_PATH = Path(r"\\h2o\local\4-transversal\BDD\mdb\Wincan\parc_2007\DB\PARC_2007.mdb")
COMMUN_DB_PATH = Path(r"\\h2o\local\4-transversal\BDD\mdb\Commun_Data.mdb")
ANALYTIQUE_DB_PATH = Path(r"\\h2o\local\4-transversal\BDD\mdb\Db_analytique.mdb")
BO_DB_PATH = Path(r"\\h2o\local\4-transversal\BDD\mdb\dbBO.mdb")
AGRHUM_DB_PATH = Path(r"\\h2o\local\4-transversal\BDD\mdb\BDD_ParcRH.mdb")
PDA_DB_PATH = Path(r"\\h2o\local\4-transversal\BDD\mdb\PDA\db_PDA.mdb")
PIRACA_DB_PATH = Path(r"\\h2o\local\4-transversal\BDD\mdb\db_Piraca.mdb")

# Config de _qgis_sync
COMPACTAGE_DIR = Path(r"\\h2o\local\1-PARC\activités\assainissement\Controle Réseaux\compactage\gps compactage")
ITV_DIR = Path(r"\\h2o\local\1-PARC\activités\assainissement\Controle Réseaux\inspection\Projets ITV\GPS")



### ***********  Ne pas modifier en dessous   *************** ###

DEFAULT_LOG_DIR.mkdir_p()
WRKDIR.mkdir_p()

def mk_workdir(name):
    """ créé le répertoire de travail pour le script 'name' """
    workdir = WRKDIR / name
    workdir.mkdir_p()
    return workdir

# ## Bases du PDE

class ParcDb(AccessSDb):
    _path = ""
    def __init__(self, **kwargs):
        super(AccessSDb, self).__init__(self._path, systemdb=MDW_PATH, uid=UID, pwd=PWD, **kwargs)

class FacturesDb(ParcDb):
    _path = FACTURES_DB_PATH

class ControlesDb(ParcDb):
    _path = CONTROLES_DB_PATH

class WincanDb(ParcDb):
    _path = WINCAN_DB_PATH

class CommunDb(ParcDb):
    _path = FACTURES_DB_PATH

class AnalytiqueDb(ParcDb):
    _path = ANALYTIQUE_DB_PATH

class BoDb(ParcDb):
    _path = BO_DB_PATH

class AgrhumDb(ParcDb):
    _path = AGRHUM_DB_PATH

class PdaDb(ParcDb):
    _path = PDA_DB_PATH

class PiracaDb(ParcDb):
    _path = PIRACA_DB_PATH

class CSigDb(PostgresDb):
    server = CSIG_SERVER
    db = CSIG_DB
    port = CSIG_PORT
    user = CSIG_USER
    pwd = CSIG_PWD

    def __init__(self, **kwargs):
        super(CSigDb, self).__init__(server=self.server, dbname=self.db, user=self.user, pwd=self.pwd, **kwargs)

# ## Modeles

class Affaire(Model):
    """ Modèle de données d'une affaire Analytique """
    def __init__(self):
        self.DblAffaireId = None, int
        self.strMOId = None, str
        self.strMOeId = None, str
        self.strEntrepriseId = None, str
        self.strCommneId = None, str
        self.strLieux = None, str
        self.strTypeId = None, str
        self.dtmCommande = None, datetime
        self.Ref = None, str
        self.blnMarche = None, bool
        self.dblMarche = 0, float
        self.intTypeContrat = None, int
        self.strCT = None, str
        self.strAvancement = None, str
        self.strLiaisonControle = None, str
        self.intDevisId = None, int
        self.blnTarification = False, bool
        self.blnAnalyse = False, bool
        self.remarques = None, str
        self.strSituation = None, str
        self.dtmFin = None, datetime
        self.intCoefFG = 0, float

class Interv(Model):
    """ Modèle de données d'une intervention de contrôle réseaux telle que représentée dan Analytique """
    def __init__(self):
        self.dblInterventionId = None, int
        self.dblAffaireId = 0, int
        self.strEquipeId = None, str
        self.strEnginId = None, str
        self.strRapportId = None, str
        self.strCatégorieInterventionId = None, str
        self.strTypeInterventionId = None, str
        self.dblquantite = 0, float
        self.strunite = None, str
        self.dtmIntervention = None, datetime
        self.dtmDureeIntervention = None, datetime
        self.dtmDureeInstallation = None, datetime
        self.strcaracteristique1 = None, str
        self.strgrandeur1 = None, str
        self.strunite1 = None, str
        self.strcaracteristique2 = None, str
        self.strgrandeur2 = None, str
        self.strunite2 = None, str
        self.strcaracteristique3 = None, str
        self.strgrandeur3 = None, str
        self.strunite3 = None, str
        self.strLiaisonControle = None, str
        self.strarticleId = None, str
        self.intPeriode = 0, int
        self.blnTarification = False, bool
        self.blnAnalyse = False, bool
        self.blnFacturer = None, bool
        self.remarques = None, str
        self.blnPeriode = None, bool
        self.dtnPeriodeDebut = None, datetime
        self.dtmImportation = None, datetime
        self.blnVerifFacture = None, bool
        self.strTest = None, str

class Tarification(Model):
    """ Modèle de donnée d'une ligne de tarification """
    def __init__(self):
        self.DblTarifId = None, int
        self.DblAffaireId = 0, int
        self.strRapportId = None, int
        self.strArticleId = None, str
        self.dblQuantite = 0, float
        self.strUnite = None, str
        self.dtmDebut = None, datetime
        self.dtmFin = None, datetime
        self.bytPeriode = 0, int
        self.dblPrixUnitaire = 0, float
        self.dblPrixTotal = 0, float
        self.bytCodeTiers = 0, int
        self.dblTauxTVA = 0, float
        self.dblPrixTVA = 0, float
        self.strStatut = None, str
        self.strNumFacture = None, str

class EnTete(Model):
    """ En-tête d'une facture dans FacturesDb """
    def __init__(self):
        self.lngPieceId = 0, int
        self.lngDocId = 0, int
        self.lngTiersId = 0, int
        self.lngASTRE = None, int
        self.strCodeProduit = None, str
        self.bytTitreId = 0, int
        self.strCodeAdresse = None, str
        self.strAdresse1 = None, str
        self.strAdresse2 = None, str
        self.strAdresse3 = None, str
        self.strAdresse4 = None, str
        self.strAdresse5 = None, str
        self.strPaysIdIso3166 = None , str  # (!) le nom réel du champ est '[strPaysIdIso3166-A2]'
        self.bytTVATiersId = 0, int
        self.strTelephone = None, str
        self.strTelecopie = None, str
        self.strPortable = None, str
        self.strEMail = None, str
        self.strWeb = None, str
        self.strLangueIdIso639 = None, str
        self.strCompteComptable = None, str
        self.strDeviseIdIso4217 = None, str
        self.bytReglementId = 0, int
        self.strReglement = None, str
        self.bytClasseTarifId = 0, int
        self.bytClasseRemiseTiersId = 0, int
        self.bytNbExFacture = 1, int
        self.strStatTiers = None, str
        self.bytSituationIdPrincipale = 0, int
        self.bytSituationIdSecondaire = 0, int
        self.memObsEntete = None, str
        self.memObsPied = None, str
        self.memObsInterne = None, str
        self.bytTypeDocumentId = 0, int
        self.strStatDocEntete = None, str
        self.dblCoursDevise = 1, int
        self.dtmCoursDevise = datetime.now(), datetime
        self.dtmCreation = datetime.now(), datetime
        self.strUserIdCreation = None, str
        self.dtmLastMod = datetime.now()
        self.strUserIdLastMod = None, str
        self.dtmPiece = datetime.now(), datetime
        self.dtmFirstPrint = None, datetime
        self.dtmLastPrint = None, datetime
        self.dtmCompta = None, datetime
        self.dtmLivraison = None, datetime
        self.dtmEcheance = None, datetime
        self.bytTypeTarif = 1, int
        self.idEtatEmission = None, str

class Ligne(Model):
    """ Ligne d'une facture dans FacturesDb """
    def __init__(self):
        self.lngPieceId = 0, int
        self.intLigneId = 0, int
        self.strArticleId = None, str
        self.strArticle = None, str
        self.bytTVAArticleId = 0, int
        self.bytTVAId = 0, int
        self.dblTVATaux = 0, float
        self.dblQte = 0, float
        self.bytUniteIdQuantite = 0, int
        self.bytNbDecQuantite = 2, int
        self.bytUniteIdPrix = 0, int
        self.dblCnvUniteCoef = 1.0, float
        self.bytNbDecQuantiteConvertie = 2, int
        self.dblPUhtBrutDev = 0, float
        self.dblPUttcBrutDev = 0, float
        self.dblPUhtNetDev = 0, float
        self.dblPUttcNetDev = 0, float
        self.bytNbDecPU = 2, int
        self.dblPThtBrutDev = 0, float
        self.dblPTttcBrutDev = 0, float
        self.dblPThtNetDev = 0, float
        self.dblPTttcNetDev = 0, float
        self.strStatArticle = None, str
        self.strStatDocLigne = None, str
        self.dblTauxRemise1 = 0, float
        self.dblTauxRemise2 = 0, float
        self.dblTauxRemise3 = 0, float
        self.blnPrestation = False, bool
        self.strCompteComptable = None, str
        self.memObs = None, str
        self.memObsInterne = None, str
        self.bytLigneSousTotal = 1, int
        self.intLigneIdRattachement = 0, int
        self.blnLigneVisible = True, bool
        self.blnLigneGeneree = False, bool
        self.bytGenerateurId = 0, int
        self.dblPUhtBrut = 0, float
        self.dblPUttcBrut = 0, float
        self.dblPUhtNet = 0, float
        self.dblPUttcNet = 0, float
        self.dblPThtBrut = 0, float
        self.dblPTttcBrut = 0, float
        self.dblPThtNet = 0, float
        self.dblPTttcNet = 0, float
        self.bytClasseRemiseArticleId = 0, int
        self.dblPUSaisie = 0, float
        self.strPUAff = "", str
        self.strQteAff = "", str



class InterventionITV(Model):
    """ Modèle d'une intervention vidéo sur un réseau d'assainissement dans Controles """
    def __init__(self):
        self.lngChantierId = 0, int
        self.bytIntervId = 0, int
        self.dtmIntervDu = None, datetime
        self.dtmIntervAu = None, datetime
        self.strEquipeId = "", str
        self.intlMaterielID = 0, int
        self.bytCommandeId = 0, int
        self.lngTroncon = 0, int
        self.SI_InspMethod = "", str
        self.SI_ReasonOfInspection = "", str


class InspectionTronconWincan(Model):
    """ Inspection vidéo d'un tronçon de réseau d'assainissement dans Wincan """
    def __init__(self):
        self.s_guid = "", str
        self.si_guid = "", str
        self.nom_chantier = "", str
        self.lng_chantier_id = 0, int
        self.byt_interv_id = 0, int
        self.si_autonumber = 0, int
        self.classement_troncons = "", str
        self.nom_troncon = "", str
        self.startnode_type = "", str
        self.endnode_type = "", str
        self.sens_ecoul = "", str
        self.startnode_z = 0.0, float
        self.endnode_z = 0.0, float
        self.section_length = 0.0, float
        self.section_purpose = "", str
        self.section_use = "", str
        self.section_type = "", str
        self.materiau = "", str
        self.annee_pose = 0, int
        self.diametre = "", str
        self.route = "", str
        self.n_route = "", str
        self.pipe_shape = "", str
        self.pipe_length = 0.0, float
        self.arbres = "", str
        self.test_ecoulement = "", str
        self.si_date = None, datetime
        self.nb_ops = 0, int
        self.nb_photos = 0, int
        self.rate_1 = None, int
        self.rate_2 = None, int
        self.rate_3 = None, int
        self.rate_4 = None, int
        self.rate_5 = None, int
        self.DG = "", str


#### Objets utilisés par la synchro QGis


class QGisChantier(object):
    """represent a chantier (Video+Etancheite or Compactage)"""
    def __init__(self):
        self.pgid = None
        self.dir_path = ""
        self.number = 0  # ex: 123000
        self.complement = ""  # ex: -S1
        self.name = ""  # ex: 123000-S1 COMMUNE
        self.x0 = 0
        self.y0 = 0
        self.x1 = 0
        self.y1 = 0
        self.error_msg = ""
        self.items = []

    def __repr__(self):
        return "[Chantier: {}]".format(self.name if self.name else self.number)

class QGisPoint(object):
    """represent a point
    could be a Point de Compactage or a Regard"""
    def __init__(self):
        self.number = 0
        self.name = ""
        self.x = 0
        self.y = 0

    def __repr__(self):
        return "[Point: {}, {}, ({}, {})]".format(self.number, self.name, self.x, self.y)

class QGisRegard(QGisPoint):
    """represent a Regard"""
    def __init__(self):
        QGisPoint.__init__(self)
        self.pgid = 0
        self.res_ce = None

    def sql_res_ce(self):
        """return a sql formatted result for Etancheite """
        return "'{}'".format(self.res_ce) if self.res_ce != None else "NULL"

    def __eq__(self, other):
        return self.name == other.name

    def __hash__(self):
        # used for comparison (in 'set' among others)
        return hash((self.number, self.x, self.y))

    def __repr__(self):
        return "[Regard: {}, {}, {}, {}, ({}, {})]".format(self.pgid,
                                                           self.number,
                                                           self.name,
                                                           self.res_ce,
                                                           self.x,
                                                           self.y)

class QGisTroncon(object):
    """ represent a troncon """
    def __init__(self):
        self.name = ""
        self.r1 = None
        self.r2 = None
        self.res_itv = None
        self.res_ce = None
        self.s_autonumber = 0
        self.si_autonumber = 0

    def __repr__(self):
        return "[Troncon > name='{}', r1={}, r2={}, res_ce='{}', \
        res_itv='{}', s_autonumber={}, si_autonumber{}]".format(self.name,
                                                                self.r1,
                                                                self.r2,
                                                                self.res_ce,
                                                                self.res_itv,
                                                                self.s_autonumber,
                                                                self.si_autonumber)

    def sql_res_ce(self):
        """return a sql formatted result for Etancheite """
        return "'{}'".format(self.res_ce) if self.res_ce != None else "NULL"

    def sql_res_itv(self):
        """return a sql formatted result for Video """
        return "'{}'".format(self.res_itv) if self.res_itv != None else "NULL"




if __name__ == "__main__":
    for cls in (FacturesDb, ControlesDb, WincanDb, CommunDb, AnalytiqueDb, BoDb, AgrhumDb):
        db = cls()
        db.assert_connected()
    print("connections ok")