from ipe_fetcher.model import AreaCoordinates, Building, FAIEligibilityStatus from ipe_fetcher.sqlite_connector.cursor import getCursorWithSpatialite from os.path import exists AXIONE_ETAT_DEPLOYE = "DEPLOYE" 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" 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") 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 <= 0.08: cur.execute( f""" SELECT X(ImmeubleGeoPoint), Y(ImmeubleGeoPoint), IdentifiantImmeuble, EtatImmeuble, NumeroVoieImmeuble, TypeVoieImmeuble, NomVoieImmeuble, CodePostalImmeuble, CommuneImmeuble 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, ) if not existing_buildings: existing_buildings = dict() buildings = existing_buildings for b in cur.fetchall(): etatImm = b[3] idImm = b[2] isEligible = etatImm == AXIONE_ETAT_DEPLOYE aquilenetEligStatus = FAIEligibilityStatus( isEligible=isEligible, ftthStatus=etatImm, reasonNotEligible="" if isEligible else "Pas encore deploye", ) if buildings.get(idImm): buildings[idImm]["aquilenetEligStatus"] = aquilenetEligStatus 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"], 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")