Skip to content
Extraits de code Groupes Projets
Sélectionner une révision Git
  • 2f93caa0327218e171b31a0fe792e45dd007cea9
  • 14.0 par défaut
  • 16.0
  • 18.0
4 résultats

bodacc_mixin.py

Blame
  • api_enedis.py 9,79 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
    
    from datetime import datetime, date
    
    from odoo import models, exceptions
    from requests.auth import _basic_auth_str
    
    _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
            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):
            """ 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)
    
            token = self.access_token()
            url_enedis = self.env.user.company_id.url_enedis
            # url_enedis = 'https://gw.hml.api.enedis.fr/'
    
            header = {
                'Content-Type': 'application/json;charset=UTF-8',
                'Accept': 'application/json',
                'Authorization': 'Bearer ' + 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
    
            try:
                response.raise_for_status()
            except Exception as error:
                raise ValueError('Autentication failed: {}'.format(error))
    
            # Gestion erreur API
            if response.status_code not in [200]:
                try:
                    message = response.json().get('content')
                except:
                    message = response.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):
            """ 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)
    
            return True
    
        def load_data(self, url, date_start, date_end, type_courbe=None, usage_point_name=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)
    
            # 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_id', '=', 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
    
                for point in curve['interval_reading']:
                    date_slot = datetime.strptime(point['timestamp'], "%Y-%m-%dT%H:%M:%SZ")
                    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,
                    })
    
                # 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=[])
    
            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([
                        ('enercoop_operation_ids', 'in', self.id),
                        ('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
    
                        prm_id = self.env['enercoop.counter'].create({
                            'name': usage_point['usage_point_id'],
                            'enercoop_operation_ids': [(4, self.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