Axione-IPE-Viewer/webapp/ipe_fetcher/axione.py
2022-04-11 09:49:51 +02:00

105 lines
3.9 KiB
Python

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):
self.db_axione_ipe_path = db_axione_ipe_path
# 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(
# """
# SELECT
# X(ImmeubleGeoPoint),
# Y(ImmeubleGeoPoint),
# IdentifiantImmeuble,
# EtatImmeuble,
# NumeroVoieImmeuble,
# TypeVoieImmeuble,
# NomVoieImmeuble
# FROM ipe
# WHERE ROWID IN (
# SELECT ROWID FROM SpatialIndex
# WHERE f_table_name = 'ipe' AND
# search_frame = BuildMBR(:swx, :swy, :nex, :ney, 4326))
# """,
# areaCoordinates,
# )
cur.execute(
"""
SELECT
X(ImmeubleGeoPoint),
Y(ImmeubleGeoPoint),
imb_id,
imb_etat,
num_voie,
type_voie,
nom_voie
FROM arcep
WHERE ROWID IN (
SELECT ROWID FROM SpatialIndex
WHERE f_table_name = 'arcep' 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():
print(b)
etatImm = b[3]
idImm = b[2]
isEligible = etatImm == AXIONE_ETAT_DEPLOYE or etatImm == "deploye"
aquilenetEligStatus = FAIEligibilityStatus(
isEligible=isEligible,
ftthStatus=etatImm,
reasonNotEligible=None if isEligible else "Pas encore deploye",
)
if buildings.get(idImm):
buildings[idImm]["aquilenetEligStatus"] = aquilenetEligStatus
else:
building = Building(
x=b[0],
y=b[1],
idImm=idImm,
numVoieImm=b[4],
typeVoieImm=b[5],
nomVoieImm=b[6],
aquilenetEligStatus=aquilenetEligStatus,
)
buildings[idImm] = building
return buildings
else:
raise ValueError("The requested area is too wide, please reduce it")