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

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