from datetime import datetime from os.path import exists from ipe_fetcher.model import AreaCoordinates, Building, FAIEligibilityStatus from ipe_fetcher.sqlite_connector.cursor import ( get_cursor_with_spatialite, get_base_cursor, ) 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 = get_base_cursor(self.db_axione_ipe_path) except Exception as err: print("Error while connecting to DB with base cursor: ", err) raise RuntimeError("Could not connect to axione DB with base cursor") 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 get_area_buildings( self, area_coordinates: AreaCoordinates, existing_buildings: dict ) -> dict: # Try to get cursor on Axione database try: cur = get_cursor_with_spatialite(self.db_axione_ipe_path) except Exception as err: print("Error while connecting to DB with spatialite cursor: ", err) raise RuntimeError("Could not connect to axione DB") # 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)) """, area_coordinates, ) 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)) """, area_coordinates, ) res = cur.fetchall() if not existing_buildings: existing_buildings = dict() buildings = existing_buildings for b in res: etat_imm = b[3] id_imm = b[2] is_eligible = etat_imm == AXIONE_ETAT_DEPLOYE date_debut = b[9] reason_not_eligible = "" if is_eligible 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 date_debut: try: date_formatted = datetime.strptime(date_debut, "%Y%m%d").date() if date_formatted >= datetime.now().date(): if is_eligible: etat_imm = 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, ) aquilenet_elig_status = FAIEligibilityStatus( isEligible=is_eligible, ftthStatus=etat_imm, reasonNotEligible=reason_not_eligible, dateCommandable=date_commandable, ) etat_priority = self._get_etat_priority(etat_imm) if buildings.get(id_imm): buildings[id_imm]["aquilenetEligStatus"] = aquilenet_elig_status buildings[id_imm]["etat_imm_priority"] = etat_priority if buildings[id_imm].get("found_in"): buildings[id_imm]["found_in"].append("axione") else: buildings[id_imm]["found_in"] = ["axione"] else: building = Building( x=b[0], y=b[1], idImm=id_imm, 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=aquilenet_elig_status, fdnEligStatus=FAIEligibilityStatus( isEligible=False, reasonNotEligible="", ftthStatus="" ), othersEligStatus=FAIEligibilityStatus( isEligible=False, reasonNotEligible="", ftthStatus="" ), ) buildings[id_imm] = 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 = get_base_cursor(self.db_axione_ipe_path) except Exception as err: print("Error while connecting to DB: ", err) raise RuntimeError("Could not connect to axione DB") try: cur.execute( f""" SELECT EtatImmeuble, DateDebutAcceptationCmdAcces FROM {self.db_name_refimm} WHERE IdentifiantImmeuble == '{id_immeuble}' """ ) res = cur.fetchone() if res: imm_elig = res[0] is_eligible = imm_elig == AXIONE_ETAT_DEPLOYE reason_not_eligible = "" if is_eligible else "Pas encore deploye" date_commandable = res[1] else: imm_elig = "NOT_AXIONE" is_eligible = False reason_not_eligible = "Axione ne gere pas ce batiment" date_commandable = "" except Exception: imm_elig = "NOT_AXIONE" is_eligible = False reason_not_eligible = "Axione ne gere pas ce batiment" date_commandable = "" return FAIEligibilityStatus( isEligible=is_eligible, ftthStatus=imm_elig, reasonNotEligible=reason_not_eligible, dateCommandable=date_commandable, )