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

210 lines
8.1 KiB
Python
Raw Permalink Normal View History

2022-12-09 12:01:55 +01:00
from datetime import datetime
from os.path import exists
2022-11-16 16:10:30 +01:00
2022-03-23 22:54:04 +01:00
from ipe_fetcher.model import AreaCoordinates, Building, FAIEligibilityStatus
2023-02-28 22:31:35 +01:00
from ipe_fetcher.sqlite_connector.cursor import (
get_cursor_with_spatialite,
get_base_cursor,
)
2022-11-16 16:10:30 +01:00
2022-03-23 22:54:04 +01:00
AXIONE_ETAT_DEPLOYE = "DEPLOYE"
2022-12-09 12:01:55 +01:00
AXIONE_ETAT_DEPLOYE_NON_COMMANDABLE = "DEPLOYE MAIS NON COMMANDABLE"
2022-03-23 22:54:04 +01:00
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"
2022-11-16 16:10:30 +01:00
AXIONE_REFIMM_TABLE_NAME = "refimm"
2022-03-23 22:54:04 +01:00
2023-01-11 15:35:12 +01:00
AXIONE_MAX_AREA = 0.08
2022-03-23 22:54:04 +01:00
class Axione:
2022-04-12 20:39:33 +02:00
def __init__(self, db_axione_ipe_path: str, db_name: str):
2022-03-23 22:54:04 +01:00
self.db_axione_ipe_path = db_axione_ipe_path
2022-04-12 20:39:33 +02:00
self.db_name = db_name
2022-03-23 22:54:04 +01:00
# 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")
2022-11-16 16:10:30 +01:00
try:
2023-02-28 22:31:35 +01:00
cur = get_base_cursor(self.db_axione_ipe_path)
2022-11-16 16:10:30 +01:00
except Exception as err:
2023-02-28 22:31:35 +01:00
print("Error while connecting to DB with base cursor: ", err)
raise RuntimeError("Could not connect to axione DB with base cursor")
2023-01-11 16:44:06 +01:00
cur.execute(
f""" SELECT count(name) FROM sqlite_master WHERE type='table' AND name='{AXIONE_REFIMM_TABLE_NAME}' """
)
2022-11-16 16:10:30 +01:00
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):
2022-12-09 12:01:55 +01:00
if etat_imm in AXIONE_ETAT_DEPLOYE:
return 0
2022-12-09 12:01:55 +01:00
elif etat_imm == AXIONE_ETAT_DEPLOYE_NON_COMMANDABLE:
return 1
2022-12-09 12:01:55 +01:00
elif etat_imm == AXIONE_ETAT_DEPLOIEMENT:
return 2
2022-12-09 12:01:55 +01:00
elif etat_imm == AXIONE_ETAT_RAD_DEPLOIEMENT:
return 3
2022-12-09 12:01:55 +01:00
elif etat_imm != AXIONE_ETAT_ABANDONNE:
2023-01-11 15:35:12 +01:00
return 20
2022-12-09 12:01:55 +01:00
else:
2023-01-11 15:35:12 +01:00
return 21
2023-02-28 22:31:35 +01:00
def get_area_buildings(
self, area_coordinates: AreaCoordinates, existing_buildings: dict
2022-03-23 22:54:04 +01:00
) -> dict:
2023-02-28 22:31:35 +01:00
# Try to get cursor on Axione database
2022-03-23 22:54:04 +01:00
try:
2023-02-28 22:31:35 +01:00
cur = get_cursor_with_spatialite(self.db_axione_ipe_path)
2022-03-23 22:54:04 +01:00
except Exception as err:
2023-02-28 22:31:35 +01:00
print("Error while connecting to DB with spatialite cursor: ", err)
raise RuntimeError("Could not connect to axione DB")
2022-03-23 22:54:04 +01:00
# 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))
""",
2023-02-28 22:31:35 +01:00
area_coordinates,
2022-03-23 22:54:04 +01:00
)
req_area = cur.fetchone()[0]
2023-01-11 15:35:12 +01:00
if req_area <= AXIONE_MAX_AREA:
2022-03-23 22:54:04 +01:00
cur.execute(
2022-04-12 20:39:33 +02:00
f"""
2022-03-23 22:54:04 +01:00
SELECT
X(ImmeubleGeoPoint),
Y(ImmeubleGeoPoint),
2022-04-12 20:39:33 +02:00
IdentifiantImmeuble,
EtatImmeuble,
NumeroVoieImmeuble,
TypeVoieImmeuble,
2022-04-12 23:17:40 +02:00
NomVoieImmeuble,
CodePostalImmeuble,
2022-12-09 12:01:55 +01:00
CommuneImmeuble,
DateDebutAcceptationCmdAcces
2022-04-12 20:39:33 +02:00
FROM {self.db_name}
2022-03-23 22:54:04 +01:00
WHERE ROWID IN (
SELECT ROWID FROM SpatialIndex
2022-04-12 20:39:33 +02:00
WHERE f_table_name = '{self.db_name}' AND
2022-03-23 22:54:04 +01:00
search_frame = BuildMBR(:swx, :swy, :nex, :ney, 4326))
""",
2023-02-28 22:31:35 +01:00
area_coordinates,
2022-03-23 22:54:04 +01:00
)
2022-11-16 16:10:30 +01:00
res = cur.fetchall()
2022-03-23 22:54:04 +01:00
if not existing_buildings:
existing_buildings = dict()
buildings = existing_buildings
2022-11-16 16:10:30 +01:00
for b in res:
2023-02-28 22:31:35 +01:00
etat_imm = b[3]
id_imm = b[2]
is_eligible = etat_imm == AXIONE_ETAT_DEPLOYE
2022-12-09 12:01:55 +01:00
date_debut = b[9]
2023-02-28 22:31:35 +01:00
reason_not_eligible = "" if is_eligible else "Pas encore deploye"
2022-12-09 12:01:55 +01:00
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
2023-01-11 15:35:12 +01:00
2023-01-11 16:44:06 +01:00
if date_debut:
2022-12-09 12:01:55 +01:00
try:
2023-01-11 16:44:06 +01:00
date_formatted = datetime.strptime(date_debut, "%Y%m%d").date()
2023-01-11 15:35:12 +01:00
2022-12-09 12:01:55 +01:00
if date_formatted >= datetime.now().date():
2023-02-28 22:31:35 +01:00
if is_eligible:
etat_imm = AXIONE_ETAT_DEPLOYE_NON_COMMANDABLE
2023-01-11 16:44:06 +01:00
date_commandable = date_formatted.strftime("%d/%m/%Y")
2022-12-09 12:01:55 +01:00
except ValueError as err:
2023-01-11 16:44:06 +01:00
print(
"Error while mainpulating DateDebutAcceptationCmdAcces from Axione DB: ",
err,
)
2022-12-09 12:01:55 +01:00
2023-02-28 22:31:35 +01:00
aquilenet_elig_status = FAIEligibilityStatus(
isEligible=is_eligible,
ftthStatus=etat_imm,
reasonNotEligible=reason_not_eligible,
2023-01-11 16:44:06 +01:00
dateCommandable=date_commandable,
2022-03-23 22:54:04 +01:00
)
2023-02-28 22:31:35 +01:00
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")
2022-04-12 20:39:33 +02:00
else:
2023-02-28 22:31:35 +01:00
buildings[id_imm]["found_in"] = ["axione"]
2022-12-09 12:01:55 +01:00
2022-03-23 22:54:04 +01:00
else:
building = Building(
x=b[0],
y=b[1],
2023-02-28 22:31:35 +01:00
idImm=id_imm,
2022-03-23 22:54:04 +01:00
numVoieImm=b[4],
typeVoieImm=b[5],
nomVoieImm=b[6],
2022-04-12 23:17:40 +02:00
codePostal=b[7],
commune=b[8],
2022-04-12 20:39:33 +02:00
bat_info="",
2023-01-11 16:44:06 +01:00
found_in=["axione"],
etat_imm_priority=etat_priority,
2023-02-28 22:31:35 +01:00
aquilenetEligStatus=aquilenet_elig_status,
2023-01-11 16:44:06 +01:00
fdnEligStatus=FAIEligibilityStatus(
isEligible=False, reasonNotEligible="", ftthStatus=""
),
othersEligStatus=FAIEligibilityStatus(
isEligible=False, reasonNotEligible="", ftthStatus=""
),
2022-03-23 22:54:04 +01:00
)
2023-02-28 22:31:35 +01:00
buildings[id_imm] = building
2022-03-23 22:54:04 +01:00
return buildings
else:
raise ValueError("The requested area is too wide, please reduce it")
2022-11-16 16:10:30 +01:00
def get_eligibilite_per_id_immeuble(self, id_immeuble: str):
# Try to get cursor on Axione database
try:
2023-02-28 22:31:35 +01:00
cur = get_base_cursor(self.db_axione_ipe_path)
2022-11-16 16:10:30 +01:00
except Exception as err:
print("Error while connecting to DB: ", err)
2023-02-28 22:31:35 +01:00
raise RuntimeError("Could not connect to axione DB")
2022-11-16 16:10:30 +01:00
2023-01-11 16:44:06 +01:00
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]
2023-02-28 22:31:35 +01:00
is_eligible = imm_elig == AXIONE_ETAT_DEPLOYE
reason_not_eligible = "" if is_eligible else "Pas encore deploye"
date_commandable = res[1]
2023-01-11 16:44:06 +01:00
else:
imm_elig = "NOT_AXIONE"
2023-02-28 22:31:35 +01:00
is_eligible = False
reason_not_eligible = "Axione ne gere pas ce batiment"
date_commandable = ""
except Exception:
2022-11-16 16:10:30 +01:00
imm_elig = "NOT_AXIONE"
2023-02-28 22:31:35 +01:00
is_eligible = False
reason_not_eligible = "Axione ne gere pas ce batiment"
date_commandable = ""
2022-11-16 16:10:30 +01:00
2023-02-28 22:31:35 +01:00
return FAIEligibilityStatus(
isEligible=is_eligible,
2022-11-16 16:10:30 +01:00
ftthStatus=imm_elig,
2023-02-28 22:31:35 +01:00
reasonNotEligible=reason_not_eligible,
dateCommandable=date_commandable,
2022-11-16 16:10:30 +01:00
)