Axione-IPE-Viewer/webapp/ipe_fetcher/axione.py

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")