210 lines
8 KiB
Python
210 lines
8 KiB
Python
from datetime import datetime
|
|
from os.path import exists
|
|
|
|
from ipe_fetcher.model import AreaCoordinates, Building, FAIEligibilityStatus
|
|
from ipe_fetcher.sqlite_connector.cursor import getCursorWithSpatialite, getBasicCursor
|
|
|
|
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 = getBasicCursor(self.db_axione_ipe_path)
|
|
except Exception as err:
|
|
print("Error while connecting to DB: ", err)
|
|
raise "Could not connect to axione DB"
|
|
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 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 <= 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))
|
|
""",
|
|
areaCoordinates,
|
|
)
|
|
|
|
res = cur.fetchall()
|
|
|
|
if not existing_buildings:
|
|
existing_buildings = dict()
|
|
buildings = existing_buildings
|
|
for b in res:
|
|
etatImm = b[3]
|
|
idImm = b[2]
|
|
isEligible = etatImm == AXIONE_ETAT_DEPLOYE
|
|
date_debut = b[9]
|
|
reasonNotEligible = "" if isEligible 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 isEligible:
|
|
etatImm = 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,
|
|
)
|
|
|
|
aquilenetEligStatus = FAIEligibilityStatus(
|
|
isEligible=isEligible,
|
|
ftthStatus=etatImm,
|
|
reasonNotEligible=reasonNotEligible,
|
|
dateCommandable=date_commandable,
|
|
)
|
|
etat_priority = self._get_etat_priority(etatImm)
|
|
if buildings.get(idImm):
|
|
buildings[idImm]["aquilenetEligStatus"] = aquilenetEligStatus
|
|
buildings[idImm]["etat_imm_priority"] = etat_priority
|
|
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"],
|
|
etat_imm_priority=etat_priority,
|
|
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")
|
|
|
|
def get_eligibilite_per_id_immeuble(self, id_immeuble: str):
|
|
|
|
# Try to get cursor on Axione database
|
|
try:
|
|
cur = getBasicCursor(self.db_axione_ipe_path)
|
|
except Exception as err:
|
|
print("Error while connecting to DB: ", err)
|
|
raise "Could not get Axione data"
|
|
|
|
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]
|
|
isEligible = imm_elig == AXIONE_ETAT_DEPLOYE
|
|
reasonNotEligible = "" if isEligible else "Pas encore deploye"
|
|
dateCommandable = res[1]
|
|
else:
|
|
imm_elig = "NOT_AXIONE"
|
|
isEligible = False
|
|
reasonNotEligible = "Axione ne gere pas ce batiment"
|
|
dateCommandable = ""
|
|
except Exception as err:
|
|
imm_elig = "NOT_AXIONE"
|
|
isEligible = False
|
|
reasonNotEligible = "Axione ne gere pas ce batiment"
|
|
dateCommandable = ""
|
|
|
|
eligStatus = FAIEligibilityStatus(
|
|
isEligible=isEligible,
|
|
ftthStatus=imm_elig,
|
|
reasonNotEligible=reasonNotEligible,
|
|
dateCommandable=dateCommandable,
|
|
)
|
|
|
|
return eligStatus
|