129 lines
5.3 KiB
Python
129 lines
5.3 KiB
Python
from flask import Flask, request
|
|
|
|
from coordinates import check_coordinates_args, adapt_coordinates_to_max_area
|
|
from eligibility_api.elig_api_exceptions import ApiParamException
|
|
from ipe_fetcher.axione import AXIONE_MAX_AREA, Axione
|
|
from netwo.netwo import Netwo, NetwooEligibility, NETWO_DEPLOYED_STATUS
|
|
|
|
|
|
class EligibilityApiRoutes:
|
|
def __init__(self, flask_app: Flask, axione_ipe: Axione, netwo: Netwo):
|
|
self.flask_app = flask_app
|
|
self.axione_ipe = axione_ipe
|
|
self.netwo = netwo
|
|
|
|
def add_routes(self):
|
|
@self.flask_app.route("/eligibilite/axione", methods=["GET"])
|
|
def get_axione_eligibility_per_immeuble():
|
|
refimmeuble = request.args.get("refimmeuble")
|
|
if not refimmeuble:
|
|
raise ApiParamException(
|
|
"You need to specify path parameter 'refimmeuble'"
|
|
)
|
|
return self.axione_ipe.get_eligibilite_per_id_immeuble(refimmeuble)
|
|
|
|
@self.flask_app.route("/eligibilite/axione/coord", methods=["GET"])
|
|
def get_axione_eligibility_per_coordinates():
|
|
args = request.args
|
|
try:
|
|
processed_args = check_coordinates_args(args)
|
|
coordinates = adapt_coordinates_to_max_area(
|
|
processed_args, AXIONE_MAX_AREA
|
|
)
|
|
return self.axione_ipe.get_area_buildings(coordinates, {})
|
|
except ValueError:
|
|
raise ApiParamException(
|
|
"You need to specify path parameters 'swx' 'swy' 'nex' 'ney'"
|
|
)
|
|
|
|
@self.flask_app.route("/eligibilite/netwo", methods=["GET"])
|
|
def get_netwo_eligibility():
|
|
"""
|
|
Intérroge l'éligibilité Netwo
|
|
---
|
|
tags:
|
|
- API
|
|
description: |
|
|
Déclenche une recherche d'éligibilité chez Netwo:
|
|
- soit par une ref d'immeuble
|
|
- soit par les coordonnées d'un immeuble
|
|
La recherche est longue, l'API renvoie un event stream qui renvoie un statut toutes les secondes
|
|
Il est possible de retrouver les résultats d'une éligibilité via son ID.
|
|
Par défaut on renvoie uniquement les offres ftth, il est possible via ftto=true d'ajouter la ftto
|
|
parameters:
|
|
- in: query
|
|
name: ftto
|
|
required: false
|
|
schema:
|
|
type: boolean
|
|
nullable: true
|
|
allowEmptyValue: true
|
|
description: |
|
|
Si 'ftto' set alors on rajoutera les résultats FTTO à la recherche
|
|
- in: query
|
|
name: timeout_sec
|
|
required: false
|
|
schema:
|
|
type: int
|
|
nullable: false
|
|
allowEmptyValue: false
|
|
description: |
|
|
Timeout au bout du quel on retourne les offres trouvées même si la recherche n'est pas terminée
|
|
responses:
|
|
200:
|
|
description: Retourne le statut d'éligibilité et d'éventuelles offres
|
|
schema:
|
|
$ref: '#/definitions/Contribution'
|
|
examples:
|
|
application/json:
|
|
400:
|
|
description: Bad inputs
|
|
schema:
|
|
$ref: '#/definitions/ApiError'
|
|
examples:
|
|
application/json:
|
|
code: 400
|
|
name: Bad input parameter
|
|
description: Bad json format
|
|
500:
|
|
description: Erreur lors d'un appel à l'API de Netwo
|
|
schema:
|
|
$ref: '#/definitions/ApiError'
|
|
examples:
|
|
application/json:
|
|
netwo_status_code: 404
|
|
name: Error contacting Netwo API
|
|
description: ID not found
|
|
"""
|
|
args = request.args
|
|
ref_imb = args.get("ref_imb")
|
|
elig_id = args.get("id")
|
|
timeout_sec = None
|
|
search_ftto = args.get("ftto") is not None
|
|
|
|
imb_info = self.netwo.get_netwo_imb_coordinates(ref_imb)
|
|
if elig_id:
|
|
elig_offers = self.netwo.get_netwo_eligibility_results(
|
|
elig_id, search_ftto
|
|
)
|
|
return NetwooEligibility(
|
|
imb_info=imb_info,
|
|
eligOffers=elig_offers,
|
|
eligDone=True,
|
|
elig_id=elig_id,
|
|
)
|
|
|
|
if args.get("timeout_sec"):
|
|
try:
|
|
timeout_sec = int(args.get("timeout_sec"))
|
|
except ValueError:
|
|
raise ApiParamException("timeout_sec param must be an integer")
|
|
|
|
if imb_info.get("imb_status") != NETWO_DEPLOYED_STATUS:
|
|
return NetwooEligibility(
|
|
imb_info=imb_info,
|
|
)
|
|
print(f"start elig with {imb_info}")
|
|
return self.netwo.start_netwo_eligibility(
|
|
imb_info, search_ftto, timeout_sec
|
|
)
|