from datetime import datetime from os.path import exists from ipe_fetcher.model import AreaCoordinates, Building, FAIEligibilityStatus from ipe_fetcher.sqlite_connector.cursor import getCursorWithSpatialite, getBasicCursor AXIONE_ETAT_DEPLOYE = "DEPLOYE" AXIONE_ETAT_DEPLOYE_NON_COMMANDABLE = "DEPLOYE MAIS NON COMMANDABLE" AXIONE_ETAT_DEPLOIEMENT = "EN COURS DE DEPLOIEMENT" AXIONE_ETAT_ABANDONNE = "ABANDONNE" AXIONE_ETAT_CIBLE = "CIBLE" AXIONE_ETAT_SIGNE = "SIGNE" AXIONE_ETAT_RAD_DEPLOIEMENT = "RAD EN COURS DE DEPLOIEMENT" AXIONE_ETAT_RACCORDABLE_DEMANDE = "RACCORDABLE DEMANDE" AXIONE_REFIMM_TABLE_NAME = "refimm" AXIONE_MAX_AREA = 0.08 class Axione: def __init__(self, db_axione_ipe_path: str, db_name: str): self.db_axione_ipe_path = db_axione_ipe_path self.db_name = db_name # Check at least that the file exists if not exists(self.db_axione_ipe_path): raise ValueError(f"File {self.db_axione_ipe_path} does not exist") try: cur = getBasicCursor(self.db_axione_ipe_path) except Exception as err: print("Error while connecting to DB: ", err) raise "Could not connect to axione DB" cur.execute(f''' SELECT count(name) FROM sqlite_master WHERE type='table' AND name='{AXIONE_REFIMM_TABLE_NAME}' ''') self.db_name_refimm = db_name if cur.fetchone()[0] == 1: self.db_name_refimm = AXIONE_REFIMM_TABLE_NAME @staticmethod def _get_etat_priority(etat_imm): if etat_imm in AXIONE_ETAT_DEPLOYE: return 0 elif etat_imm == AXIONE_ETAT_DEPLOYE_NON_COMMANDABLE: return 1 elif etat_imm == AXIONE_ETAT_DEPLOIEMENT: return 2 elif etat_imm == AXIONE_ETAT_RAD_DEPLOIEMENT: return 3 elif etat_imm != AXIONE_ETAT_ABANDONNE: return 20 else: return 21 def getAreaBuildings( self, areaCoordinates: AreaCoordinates, existing_buildings: dict ) -> dict: cur = None # Try to get cursor on Axone database try: cur = getCursorWithSpatialite(self.db_axione_ipe_path) except Exception as err: print("Error while connecting to DB: ", err) raise "Could not get Axione data" # Let's first see how big is the area we're about to query. # If it's too big, abort the request to prevent a server DOS. cur.execute( """ SELECT Area(BuildMBR(:swx,:swy,:nex,:ney,4326)) """, areaCoordinates, ) req_area = cur.fetchone()[0] if req_area <= AXIONE_MAX_AREA: cur.execute( f""" SELECT X(ImmeubleGeoPoint), Y(ImmeubleGeoPoint), IdentifiantImmeuble, EtatImmeuble, NumeroVoieImmeuble, TypeVoieImmeuble, NomVoieImmeuble, CodePostalImmeuble, CommuneImmeuble, DateDebutAcceptationCmdAcces FROM {self.db_name} WHERE ROWID IN ( SELECT ROWID FROM SpatialIndex WHERE f_table_name = '{self.db_name}' AND search_frame = BuildMBR(:swx, :swy, :nex, :ney, 4326)) """, areaCoordinates, ) res = cur.fetchall() if not existing_buildings: existing_buildings = dict() buildings = existing_buildings for b in res: etatImm = b[3] idImm = b[2] isEligible = etatImm == AXIONE_ETAT_DEPLOYE date_debut = b[9] reasonNotEligible = "" if isEligible else "Pas encore deploye" date_commandable = "" # C'est bien déployé, cependant ce n'est pas encore commandable (donc bientôt et on a la date) # On laisse isEligible = True, côté JS il faut regarder le statut pour laj l'affichage en conséquence if isEligible and date_debut: try: date_formatted = datetime.strptime(date_debut, '%Y%m%d').date() if date_formatted >= datetime.now().date(): etatImm = AXIONE_ETAT_DEPLOYE_NON_COMMANDABLE date_commandable = date_formatted.strftime('%d/%m/%Y') except ValueError as err: print("Error while mainpulating DateDebutAcceptationCmdAcces from Axione DB: ", err) aquilenetEligStatus = FAIEligibilityStatus( isEligible=isEligible, ftthStatus=etatImm, reasonNotEligible=reasonNotEligible, dateCommandable=date_commandable ) etat_priority = self._get_etat_priority(etatImm) if buildings.get(idImm): buildings[idImm]["aquilenetEligStatus"] = aquilenetEligStatus buildings[idImm]['etat_imm_priority'] = etat_priority if buildings[idImm].get('found_in'): buildings[idImm]['found_in'].append("axione") else: buildings[idImm]['found_in'] = ["axione"] else: building = Building( x=b[0], y=b[1], idImm=idImm, numVoieImm=b[4], typeVoieImm=b[5], nomVoieImm=b[6], codePostal=b[7], commune=b[8], bat_info="", found_in = ["axione"], etat_imm_priority=etat_priority, aquilenetEligStatus=aquilenetEligStatus, fdnEligStatus=FAIEligibilityStatus(isEligible=False, reasonNotEligible="", ftthStatus=""), othersEligStatus=FAIEligibilityStatus(isEligible=False, reasonNotEligible="", ftthStatus=""), ) buildings[idImm] = building return buildings else: raise ValueError("The requested area is too wide, please reduce it") def get_eligibilite_per_id_immeuble(self, id_immeuble: str): # Try to get cursor on Axione database try: cur = getBasicCursor(self.db_axione_ipe_path) except Exception as err: print("Error while connecting to DB: ", err) raise "Could not get Axione data" cur.execute( f""" SELECT EtatImmeuble, DateDebutAcceptationCmdAcces FROM {self.db_name_refimm} WHERE IdentifiantImmeuble == '{id_immeuble}' """ ) res = cur.fetchone() if res: imm_elig = res[0] isEligible = imm_elig == AXIONE_ETAT_DEPLOYE reasonNotEligible = "" if isEligible else "Pas encore deploye" else: isEligible = False imm_elig = "NOT_AXIONE" reasonNotEligible = "Axione ne gere pas ce batiment" eligStatus = FAIEligibilityStatus( isEligible=isEligible, ftthStatus=imm_elig, reasonNotEligible=reasonNotEligible, dateCommandable=res[1] ) return eligStatus