Axione-IPE-Viewer/webapp/ipe_fetcher/arcep.py
Johan Le Baut 0fb3de88e1 clean code
2023-02-28 22:31:35 +01:00

122 lines
4.7 KiB
Python

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