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

from os.path import exists

from ipe_fetcher.model import AreaCoordinates, Building, FAIEligibilityStatus
from ipe_fetcher.sqlite_connector.cursor import get_cursor_with_spatialite

ARCEP_ETAT_DEPLOYE = "deploye"


class Arcep:
    def __init__(self, db_arcep_ipe_path: str, db_name: str):
        self.db_arcep_ipe_path = db_arcep_ipe_path
        self.db_name = db_name
        # Check at least that the file exists
        if not exists(self.db_arcep_ipe_path):
            raise ValueError(f"File {self.db_arcep_ipe_path} does not exist")

    @staticmethod
    def _get_etat_priority(etat_imm):
        if etat_imm == ARCEP_ETAT_DEPLOYE:
            return 10
        elif etat_imm == "en cours de deploiement":
            return 11
        elif etat_imm != "abandonne":
            return 30
        else:
            return 31

    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_arcep_ipe_path)
        except Exception as err:
            print("Error while connecting to DB: ", err)
            raise RuntimeError("Could not get ARCEP 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))
        """,
            area_coordinates,
        )
        req_area = cur.fetchone()[0]
        if req_area <= 0.08:
            cur.execute(
                f"""
                SELECT
                    X(ImmeubleGeoPoint),
                    Y(ImmeubleGeoPoint),
                    imb_id,
                    imb_etat,
                    num_voie,
                    type_voie,
                    nom_voie,
                    batiment,
                    code_poste,
                    nom_com
                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,
            )
            if not existing_buildings:
                existing_buildings = dict()
            buildings = existing_buildings
            for b in cur.fetchall():
                x = b[0]
                y = b[1]
                id_imm = b[2]
                etat_imm = b[3]
                num_voie_imm = b[4]
                type_voie_imm = b[5]
                nom_voie_imm = b[6]
                bat_info = b[7]
                code_postal = b[8]
                commune = b[9]
                is_eligible = etat_imm == ARCEP_ETAT_DEPLOYE
                others_elig_status = FAIEligibilityStatus(
                    isEligible=is_eligible,
                    ftthStatus=etat_imm,
                    reasonNotEligible=None if is_eligible else "Pas encore deploye",
                )

                etat_priority = self._get_etat_priority(etat_imm)
                if buildings.get(id_imm):
                    buildings[id_imm]["othersEligStatus"] = others_elig_status
                    buildings[id_imm]["bat_info"] = bat_info
                    buildings[id_imm]["etat_imm_priority"] = etat_priority
                    if buildings[id_imm].get("found_in"):
                        buildings[id_imm]["found_in"].append("arcep")
                    else:
                        buildings[id_imm]["found_in"] = ["arcep"]
                else:
                    building = Building(
                        x=x,
                        y=y,
                        idImm=id_imm,
                        numVoieImm=num_voie_imm,
                        typeVoieImm=type_voie_imm,
                        nomVoieImm=nom_voie_imm,
                        codePostal=code_postal,
                        commune=commune,
                        bat_info=bat_info,
                        found_in=["arcep"],
                        etat_imm_priority=etat_priority,
                        aquilenetEligStatus=FAIEligibilityStatus(
                            isEligible=False, reasonNotEligible="", ftthStatus=""
                        ),
                        fdnEligStatus=FAIEligibilityStatus(
                            isEligible=False, reasonNotEligible="", ftthStatus=""
                        ),
                        othersEligStatus=others_elig_status,
                    )
                    buildings[id_imm] = building
            return buildings
        else:
            raise ValueError("The requested area is too wide, please reduce it")