Sélectionner une révision Git
api_enedis.py 10,80 Kio
# © 2021 Le Filament (<http://www.le-filament.com>)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import logging
from datetime import date, datetime
import pytz
import requests
from requests.auth import _basic_auth_str
from odoo import _, exceptions, fields, models
from odoo.exceptions import UserError
_logger = logging.getLogger(__name__)
class ApiEnedis(models.AbstractModel):
"""Cet Abstract Model partage l'ensemble des fonctions
de l'API Enedis.
"""
_name = "api.enedis"
_description = "Connecteur Enedis"
def access_token(self):
"""Retourne un access token valide pour 5 secondes."""
_logger.info("Création du Token de connexion Enedis")
url_enedis = self.env.user.company_id.url_enedis
if not url_enedis:
raise UserError(
_(
"L'API n'est pas configurée. Veuillez rentrer "
"l'URL de l'API dans app -> Configuration -> Configuration"
)
)
client_id = self.client_id
secret_id = self.secret_id
auth = _basic_auth_str(client_id, secret_id)
url = url_enedis + "v1/oauth2/token"
headers = {
"Authorization": auth,
"Content-Type": "application/x-www-form-urlencoded",
}
formData = {
"grant_type": "client_credentials",
}
auth_request = requests.post(url=url, headers=headers, data=formData)
try:
auth_request.raise_for_status()
except Exception as error:
raise ValueError("Autentication failed: {}".format(error))
# Gestion erreur API
if auth_request.status_code not in [200]:
try:
message = auth_request.json().get("error_description")
except:
message = auth_request.error_description
raise exceptions.Warning(
_(
"L'appel url '%s' a échoué\n"
"- Code erreur : %d\n"
"- Message : %s" % (url, auth_request.error, message)
)
)
response = auth_request.json()
token = response.get("access_token")
return token
def enedis_get_by_url(self, url, call_type, query=None, token=None):
"""Création de la requête à Enedis
:param url: action API Enedis
:param call_type: post/get
:param query: data to get
@return response au format JSON
"""
_logger.info("Calling %s" % url)
if not token:
token = self.access_token()
url_enedis = self.env.user.company_id.url_enedis
header = {
"Content-Type": "application/json;charset=UTF-8",
"Accept": "application/json",
"Authorization": "Bearer " + str(token),
}
try:
if call_type == "get":
response = requests.get(url_enedis + url, headers=header, params=query)
elif call_type == "post":
response = requests.post(
url_enedis + url,
)
except Exception as err:
_logger.warning(
"Erreur de connexion. URL: %s",
err.__str__(),
)
response = False
print("---response----", str(response.content))
# Gestion erreur API
if response.status_code not in [200]:
try:
message = response.json().get("error_description")
except:
message = response.json()
raise exceptions.Warning(
_(
"L'appel url '%s' a échoué\n"
"- Code erreur : %d\n"
"- Message : %s" % (url, response.status_code, message)
)
)
return response
def definitive_load_curves(
self, date_start, date_end, usage_point_id=None, token=None
):
"""Fonction permettant d'appeler l'API Enedis et retourne les courbes
de chare en fonction d'un intervalle de date
:param date_start: une date de début
date_end: une date de fin
usage_point_id: id du PRM
:return Retourner les courbes de charge
à la maille d'une opération ou
d'un PRM en particulier
"""
url = (
"v1/collective_self_consumption/agreements/"
+ self.name
+ "/definitive_load_curves"
)
if usage_point_id:
name = usage_point_id.name + "_" + str(date_start) + "_" + str(date_end)
else:
name = self.name + "_" + str(date_start) + "_" + str(date_end)
log_id = self.env["acc.enedis.cdc"].search(
[("name", "=", name), ("acc_operation_id", "=", self.id)]
)
if not log_id:
self.load_data(
url, date_start, date_end, usage_point_id=usage_point_id, token=token
)
return True
def load_data(self, url, date_start, date_end, usage_point_id=None, token=None):
"""
Read function
:param url : URL à appeler
date_start: une date de début
date_end: une date de fin
type_courbe: type de courbe à récupérer
(cons/autocons/surplus/prod)
usage_point_id: id du PRM
@return object: Retourner les courbes de charge
à la maille d'une opération ou
d'un PRM en particulier
"""
if usage_point_id:
usage_point_name = usage_point_id.name
counter_id = usage_point_id.id
else:
usage_point_name = None
counter_id = None
query = {
# Date de début de la période souhaitée.La date est incluse dans la période
"start": date_start,
# Date de fin de la période souhaitée.La date est incluse dans la période
"end": date_end,
# Type de la courbe (enum)
"type": ["cons,autocons,surplus,prod,complement"],
# Identifiant du PRM
"usage_point_id": usage_point_name,
}
response = self.enedis_get_by_url(
url=url, call_type="get", query=query, token=token
)
# get the data
curves = response.json().get("curves")
if curves:
name = usage_point_id.name + "_" + str(date_start) + "_" + str(date_end)
for curve in curves:
type = curve["type"]
domain_all = [
("date_slot", ">=", date_start),
("date_slot", "<=", date_end),
("acc_operation_id", "=", self.id),
("comp_data_type", "=", type),
]
domain_prm = domain_all + [("acc_counter_id", "=", counter_id)]
acc_enedis_cdc_ids = self.env["acc.enedis.cdc"].search(domain_prm)
record_created = False
for point in curve["interval_reading"]:
dt = pytz.utc.localize(
datetime.strptime(point["timestamp"], "%Y-%m-%dT%H:%M:%SZ")
)
date_slot = fields.Datetime.to_string(
dt.astimezone(pytz.timezone("Europe/Paris"))
)
date_slot_utc = fields.Datetime.to_string(dt)
record = acc_enedis_cdc_ids.filtered(
lambda l: l.date_slot == fields.Datetime.to_datetime(date_slot)
)
if not record:
self.env["acc.enedis.cdc"].create(
{
"name": name,
"acc_operation_id": self.id,
"acc_counter_id": counter_id or False,
"comp_data_type": type,
"power": point["value"],
"date_slot": date_slot,
"date_slot_utc": date_slot_utc,
}
)
record_created = True
if record_created:
# Logs information loaded
self.env["acc.enedis.import.logs"].create(
{
"name": name,
"acc_operation_id": self.id,
}
)
self.is_data_enedis = True
def perimeter(self):
"""Fonction permettant d'appeler l'API Enedis et retourne le périmètre
d'une opréation donées
:param
:return Retourne le périmètre d'une opération donnée
Date de début et fin de contrat
"""
url = "v1/collective_self_consumption/agreements/" + self.name + "/perimeter"
response = self.enedis_get_by_url(
url=url, call_type="get", query=[], token=None
)
usage_points = response.json().get("usage_points")
for usage_point in usage_points:
usage_id = self.env["acc.counter"].search(
[
("acc_operation_id", "=", self.id),
("name", "=", usage_point["usage_point_id"]),
]
)
if usage_id:
usage_id.write(
{
"date_start_contract": date.fromisoformat(usage_point["start"]),
}
)
if usage_point["end"] != "9999-12-31":
usage_id.write(
{
"date_end_contract": date.fromisoformat(usage_point["end"]),
}
)
else:
is_delivery = False
is_injection = False
if usage_point["type"] == "CONS":
is_delivery = True
if usage_point["type"] == "PROD":
is_injection = True
self.date_start_contract = date.fromisoformat(usage_point["start"])
if usage_point["end"] != "9999-12-31":
self.date_end_contract = date.fromisoformat(usage_point["end"])
else:
self.date_end_contract = None
if usage_point["end"] != "9999-12-31":
date_end = date.fromisoformat(usage_point["end"])
else:
date_end = None
self.env["acc.counter"].create(
{
"name": usage_point["usage_point_id"],
"is_delivery": is_delivery,
"is_injection": is_injection,
"acc_operation_id": self.id,
"date_start_contract": date.fromisoformat(usage_point["start"]),
"date_end_contract": date_end,
}
)
return True