Sélectionner une révision Git
api_enedis.py 10,86 Kio
# © 2021 Le Filament (<http://www.le-filament.com>)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import requests
import logging
import pytz
import json
from datetime import datetime, date
from odoo import models, exceptions
from requests.auth import _basic_auth_str
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 Enercoop -> 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 = response.error_description
raise exceptions.Warning(
"L'appel url '%s' a échoué\n"
"- Code erreur : %d\n"
"- Message : %s" % (
url,
response.error,
message))
response = auth_request.json()
token = response.get('access_token')
print("--- token ---", str(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)
}
path = {
# ID De l'opération
'agreement_id': self.name,
}
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
# Gestion erreur API
if response.status_code not in [200]:
try:
message = response.json().get('error_description')
except:
message = response.json().get('content')
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, type, usage_point_name=None, token=None):
""" Fonction permettant d'appeler l'API Enedis et retourne les courbes
de chare en fonction d'un intervalle de date et d'un type
:param date_start: une date de début
date_end: une date de fin
type: type de courbe à récupérer
(cons/autocons/surplus/prod)
usage_point_name: nom 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_name:
name = usage_point_name + '_' + str(date_start) + '_' + str(date_end) + '_' + type
else:
name = self.name + '_' + str(date_start) + '_' + str(date_end) + '_' + type
log_id = self.env['enercoop.enedis.import.logs'].search([
('name', '=', name),
('enercoop_operation_id', '=', self.id)
])
if not log_id:
self.load_data(url, date_start, date_end, type, usage_point_name, token)
return True
def load_data(self, url, date_start, date_end, type_courbe=None, usage_point_name=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_name: nom du PRM
@return object: Retourner les courbes de charge
à la maille d'une opération ou
d'un PRM en particulier
"""
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': type_courbe,
# 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')
for curve in curves:
usage_point = curve['usage_point_id']
# Get the counter ID from data
counter_id = self.env['enercoop.counter']
if usage_point_name:
counter_id = self.env['enercoop.counter'].search([
('name', '=', usage_point_name),
('enercoop_operation_ids', 'in', self.id)])
type = curve['type']
if counter_id:
name = counter_id.name + '_' + str(date_start) + '_' + str(date_end) + '_' + type
else:
name = self.name + '_' + str(date_start) + '_' + str(date_end) + '_' + type
enercoop_enedis_cdc_ids = self.env['enercoop.enedis.cdc'].search([
('date_slot', '>=', date_start),
('date_slot', '<=', date_end),
('enercoop_operation_id', '=', self.id),
'|', ('enercoop_counter_id', '=', counter_id.id), ('enercoop_counter_id', '=', False),
('comp_data_type', '=', type),
])
record_created = False
timezone = self.env.context.get('tz') or self.env.user.tz
user_tz = pytz.timezone(timezone)
for point in curve['interval_reading']:
date_slot = pytz.utc.localize(datetime.strptime(point['timestamp'], "%Y-%m-%dT%H:%M:%SZ")).astimezone(user_tz).replace(tzinfo=None)
record = enercoop_enedis_cdc_ids.filtered(lambda l: l.date_slot == date_slot)
if not record:
self.env['enercoop.enedis.cdc'].create({
'name': name,
'enercoop_operation_id': self.id,
'enercoop_counter_id': counter_id.id or False,
'comp_data_type': type,
'power': point['value'],
'date_slot': date_slot,
})
record_created = True
if record_created:
# Logs information loaded
self.env['enercoop.enedis.import.logs'].create({
'name': name,
'enercoop_operation_id': self.id,
})
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['enercoop.counter.operation'].search([
('operation_id', '=', self.id),
('enercoop_counter_id', '=', usage_point['usage_point_id'])])
if usage_id:
usage_id.write({
'date_start_contract': date.fromisoformat(usage_point['start']),
'date_end_contract': date.fromisoformat(usage_point['end']),
})
else:
prm_id = self.env['enercoop.counter'].search([
('name', '=', usage_point['usage_point_id'])])
if not prm_id:
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'])
self.date_end_contract = date.fromisoformat(usage_point['end'])
prm_id = self.env['enercoop.counter'].create({
'name': usage_point['usage_point_id'],
'is_delivery': is_delivery,
'is_injection': is_injection,
})
self.env['enercoop.counter.operation'].create({
'enercoop_counter_id': prm_id.id,
'operation_id': self.id,
'date_start_contract': date.fromisoformat(usage_point['start']),
'date_end_contract': date.fromisoformat(usage_point['end']),
})
return True