Axione-IPE-Viewer/webapp/main.py

import configparser
import os
from typing import TypedDict

from flask import Flask, request, render_template, redirect

from eligibility_api.api_doc import start_swagger
from eligibility_api.elig_api_exceptions import FlaskExceptions
from eligibility_api.elig_api_routes import EligibilityApiRoutes
from ipe_fetcher import Liazo, Axione, Arcep, AreaCoordinates
from coordinates import adapt_coordinates_to_max_area, check_coordinates_args
from netwo.netwo import Netwo

LIAZO_MAX_X_INTERVAL = 0.0022
LIAZO_MAX_Y_INTERVAL = 0.0011
LIAZO_MAX_AREA = LIAZO_MAX_X_INTERVAL * LIAZO_MAX_Y_INTERVAL


class Config(TypedDict):
    axione_ipe_path: str
    axione_ipe_db_name: str
    arcep_ipe_path: str
    arcep_ipe_db_name: str
    netwo_api_key: str
    netwo_aquilenet_fixed_recurring_price: float


def parseConfig() -> Config:
    cfg_path = os.environ.get("CONFIG", "/etc/ftth-ipe-map/conf.ini")
    cfg = configparser.ConfigParser()
    with open(cfg_path, "r") as f:
        cfg.read_file(f)
    return {
        "axione_ipe_path": cfg.get("DB", "axione_ipe_path"),
        "axione_ipe_db_name": cfg.get("DB", "axione_ipe_db_name"),
        "arcep_ipe_path": cfg.get("DB", "arcep_ipe_path"),
        "arcep_ipe_db_name": cfg.get("DB", "arcep_ipe_db_name"),
        "netwo_api_key": cfg.get("NETWO", "api_key"),
        "netwo_aquilenet_fixed_recurring_price": float(
            cfg.get("NETWO", "aquilenet_fixed_recurring_price") or 0.00
        ),
    }


app = Flask(__name__)
cfg: Config = parseConfig()

axione = Axione(cfg.get("axione_ipe_path"), cfg.get("axione_ipe_db_name"))
arcep = Arcep(cfg.get("arcep_ipe_path"), cfg.get("arcep_ipe_db_name"))
liazo = Liazo()
netwo = Netwo(
    cfg.get("netwo_api_key"), cfg.get("netwo_aquilenet_fixed_recurring_price")
)
elig_api_routes = EligibilityApiRoutes(app, axione, netwo)
elig_api_routes.add_routes()
elig_api_exceptions = FlaskExceptions(app)
elig_api_exceptions.add_exceptions()


@app.route("/", methods=["GET"])
def getMap():
    return render_template("map.html")


@app.route("/eligdata", methods=["GET"])
def getEligData():
    toto = 1
    args = request.args
    valid_args = True
    processed_args = {}
    try:
        processed_args = check_coordinates_args(args)
    except ValueError:
        valid_args = False

    if valid_args:
        coordinates = adapt_coordinates_to_max_area(processed_args, LIAZO_MAX_AREA)

        buildings = dict()
        try:
            buildings = arcep.get_area_buildings(coordinates, buildings)
            buildings = axione.get_area_buildings(coordinates, buildings)
        except ValueError as err:
            print("Could not get Axione data for this area:", err)

        buildings = liazo.get_area_buildings(coordinates, buildings)

        return {"buildings": list(buildings.values())}
    else:
        return "Invalid bounding box coordinates", 400


@app.route("/eligdata/bounds", methods=["GET"])
def getEligDataBounds():
    args = request.args
    try:
        processed_args = check_coordinates_args(args)
        return {"bounds": adapt_coordinates_to_max_area(processed_args, LIAZO_MAX_AREA)}
    except ValueError:
        return "Invalid bounding box coordinates", 400


@app.route("/eligtest/ftth", methods=["GET"])
def testFtth():
    args = request.args
    idImm = args["idImm"]
    codePostal = args["codePostal"]
    axioneOk = args["axione"]
    liazoOk = args["liazo"]
    pto_url = f"https://tools.aquilenet.fr/cgi-bin/recherchepto.cgi?refimmeuble={idImm}&cp={codePostal}&axione={axioneOk}&liazo={liazoOk}"
    return redirect(pto_url)


# start_swagger(app)