Sélectionner une révision Git
acc_enedis_cdc.py 16,29 Kio
# Copyright 2021- Le Filament (https://le-filament.com)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html)
from datetime import datetime
from dateutil.relativedelta import relativedelta
from odoo import _, api, fields, models
from odoo.exceptions import ValidationError
from odoo.osv import expression
from odoo.tools import date_utils
from odoo.addons.api_connector.tools.date_utils import local_to_utc
class AccEnedisCdc(models.Model):
_inherit = "acc.enedis.cdc"
# ------------------------------------------------------
# Actions
# ------------------------------------------------------
# ------------------------------------------------------
# Business methods
# ------------------------------------------------------
@api.model
def _get_step_from_date(self, start_date, end_date):
"""
Fonction retournant le pas des courbes en fonction de 2 dates.
:return:
display_hourly_curves (boolean) : whether or not hourly curves
should be displayed
step: hour/month/year
step_display_curve: hour/day/month/year
"""
display_hourly_curves = False
step = "hour"
step_display_curve = "hour"
# Calculate delta between 2 dates
delta = (end_date - start_date).days
if delta > 1 and delta <= 31:
step_display_curve = "day"
display_hourly_curves = True
elif delta > 31 and delta <= 366:
step = "month"
step_display_curve = "month"
elif delta > 366:
step = "year"
step_display_curve = "year"
return display_hourly_curves, step, step_display_curve
@api.model
def _select_clause(self, date_slot, curve_types):
"""
Function to build SELECT section of query for retrieving curves
@param
char date_slot : granularity
(one of "minute", "hour", "day", "month", "year")
[char] curve_types : list of type of curves
(allowed values : 'cons', 'autocons', 'allocons',
'prod', 'surplus', 'autoprod')
"""
if date_slot not in ("minute", "hour", "day", "month", "year"):
raise ValidationError(_("Incorrect date_slot in SELECT section"))
result = f"""
SELECT date_trunc('{date_slot}',
cdc.date_slot AT TIME ZONE 'UTC' AT TIME ZONE 'Europe/Paris'
) AS date_slot
"""
for curve_type in curve_types:
if curve_type in ("cons", "autocons", "prod", "surplus"):
result += f"""
, (SUM(CASE WHEN cdc.comp_data_type = '{curve_type}'
THEN cdc.power ELSE 0 END)) /2 / 1000 as {curve_type}
"""
elif curve_type == "allocons":
result += """
,(SUM(CASE
WHEN cdc.comp_data_type = 'cons'
THEN cdc.power ELSE 0 END)
- SUM(CASE
WHEN cdc.comp_data_type = 'autocons'
THEN cdc.power ELSE 0 END)) / 2 / 1000 as allocons
"""
elif curve_type == "autoprod":
result += """
,(SUM(CASE
WHEN cdc.comp_data_type = 'prod'
THEN cdc.power ELSE 0 END)
- SUM(CASE
WHEN cdc.comp_data_type = 'surplus'
THEN cdc.power ELSE 0 END)) / 2 / 1000 as autoprod
"""
return result
@api.model
def _from_clause(self):
"""
Function to build FROM section of query for retrieving curves
"""
return """
FROM
acc_enedis_cdc cdc
INNER JOIN acc_operation ope
ON ope.id = cdc.acc_operation_id
"""
@api.model
def _where_clause(
self,
operation_id,
start_date,
end_date,
prm_id=None,
extra_curve_type=None,
partner_id=None,
):
"""
Function to build WHERE section of query for retrieving curves
@param
int operation_id : id of operation for which curves should be retrieved
date start_date : first date to be retrieved
date end_date : last date to be retrieved
int prm_id : id of PRM to be retrieved (optional)
[char] extra_curve_type : extra curve to be retrieved (optional)
(allowed values : 'cons', 'autocons', 'allocons',
'prod', 'surplus', 'autoprod')
int partner_id : id of partner to be retrieved (optional)
"""
if (
not isinstance(operation_id, int)
and not isinstance(start_date, datetime)
and not isinstance(end_date, datetime)
):
raise ValidationError(_("WHERE clause parameters incorrect"))
start_datetime = local_to_utc(start_date, "Europe/Paris")
end_datetime = local_to_utc(end_date, "Europe/Paris")
result = f"""
WHERE cdc.acc_operation_id = {operation_id}
AND cdc.date_slot >= '{start_datetime}'
AND cdc.date_slot < '{end_datetime}'
"""
if partner_id and isinstance(partner_id, int):
result += f" AND ((cdc.partner_id = {partner_id} "
if prm_id and isinstance(prm_id, int):
result += f" AND cdc.acc_counter_id = {prm_id}) "
else:
result = f"{result})"
if extra_curve_type and extra_curve_type in (
"cons",
"autocons",
"allocons",
"prod",
"surplus",
"autoprod",
):
result = f"{result} OR cdc.comp_data_type = '{extra_curve_type}' )"
else:
result = f"{result})"
return result
@api.model
def _group_clause(self, date_slot):
"""
Function to build GROUP BY section of query for retrieving curves
@param
char date_slot : granularity
(one of "minute", "hour", "day", "month", "year")
"""
if date_slot not in ("minute", "hour", "day", "month", "year"):
raise ValidationError(_("Incorrect date_slot in GROUP BY section"))
return f"""
GROUP BY date_trunc('{date_slot}',
cdc.date_slot AT TIME ZONE 'UTC' AT TIME ZONE 'Europe/Paris')
"""
@api.model
def _order_clause(self):
return "ORDER BY date_slot ASC;"
@api.model
def _cdc_by_query_cons(
self,
operation_id,
slot_type,
start_date,
end_date,
prm_id=None,
partner_id=None,
curve_types=None,
):
"""
Fonction permettant de récupérer les données pour les consommateurs
:param: int operation_id: opération concernée
char slot_type: type de slot pour la query ("minute", "hour", "month" ou
"year")
datetime start_date: date début
datetime end_date: date de fin
int prm_id : PRM de soutirage à récupérer
int partner_id: contact associé à la courbe
[char] curve_types: type de données
@returns: resultat de la requête
(labels et data pour les charts à afficher)
"""
if curve_types is None:
curve_types = ["cons", "autocons", "prod"]
query = (
self._select_clause(date_slot=slot_type, curve_types=curve_types)
+ self._from_clause()
+ self._where_clause(
operation_id,
start_date,
end_date,
prm_id,
"prod" if "prod" in curve_types else None,
partner_id,
)
+ "AND cdc.comp_data_type IN %s"
+ self._group_clause(date_slot=slot_type)
+ self._order_clause()
)
self.env.cr.execute(query, (tuple(curve_types),))
return self.env.cr.fetchall()
@api.model
def _get_cdc_by_query_cons(
self,
operation_id,
slot_type,
start_date,
end_date,
prm_id=None,
partner_id=None,
):
"""
Fonction permettant de récupérer les données pour la
construction des chart pour une ou des opérations données
pour les consommateurs
:param: int operation_id : Opération concernée
char slot_type: type de slot pour la query ("minute", "hour", "month" ou
"year")
datetime start_date: date début
datetime end_date: date de fin
int prm_id : PRM de soutirage à récupérer
int partner_id: contact associé à la courbe
@returns: resultat de la requête
(labels et data pour les charts à afficher)
"""
label = []
data_autocons = []
data_allocons = []
data_cons = []
data_prod = []
raw_data = self._cdc_by_query_cons(
operation_id, slot_type, start_date, end_date, prm_id, partner_id
)
for row in raw_data:
label.append(row[0])
data_cons.append({"x": row[0], "y": round(row[1], 2)})
data_autocons.append({"x": row[0], "y": round(row[2], 2)})
data_allocons.append({"x": row[0], "y": round(row[1] - row[2], 2)})
data_prod.append({"x": row[0], "y": round(row[3], 2)})
cdc_cons = {
"label": label,
"autocons": data_autocons,
"allocons": data_allocons,
"cons": data_cons,
"prod": data_prod,
}
return cdc_cons
@api.model
def _get_cdc_by_query_daily_histo_cons(
self,
operation_id,
start_date,
end_date,
prm_id=None,
partner_id=None,
):
"""
Fonction permettant de récupérer les données pour la construction
des chart pour une ou des opérations données pour les consommateurs
:param: int operation_id : Opération concernée
datetime start_date: date début
datetime end_date: date de fin
int prm_id : PRM de soutirage à récupérer
int partner_id: contact associé à la courbe
:return: un dictionnaire de données
(labels et data pour les charts à afficher)
"""
label_histo = []
data_autocons_histo = []
data_allocons_histo = []
query = (
self._select_clause(date_slot="day", curve_types=["autocons", "allocons"])
+ self._from_clause()
+ self._where_clause(
operation_id=operation_id,
start_date=start_date,
end_date=end_date,
prm_id=prm_id,
partner_id=partner_id,
)
+ self._group_clause(date_slot="day")
+ self._order_clause()
)
self.env.cr.execute(query)
raw_data = self.env.cr.fetchall()
for row in raw_data:
data_autocons_histo.append(round(row[1], 2))
data_allocons_histo.append(round(row[2], 2))
label_histo.append(row[0])
cdc_cons = {
"autocons_histo": data_autocons_histo,
"allocons_histo": data_allocons_histo,
"label_histo": label_histo,
}
return cdc_cons
@api.model
def _cdc_by_query_prod(
self,
operation_id,
slot_type,
start_date,
end_date,
prm_id=None,
partner_id=None,
curve_types=None,
):
"""
Fonction permettant de récupérer les données pour la construction des
chart pour une ou des opérations données pour les consommateurs
:param: int operation_id : Opération concernée
char slot_type: type de slot pour la query ("minute", "hour", "month" ou
"year")
datetime start_date: date début
datetime end_date: date de fin
int prm_id : PRM d'injection à récupérer
int partner_id: contact associé à la courbe
[char] curve_types: type de données
:return: un dictionnaire de données
(labels et data pour les charts à afficher)
"""
if curve_types is None:
curve_types = ["autoprod", "surplus"]
query = (
self._select_clause(date_slot=slot_type, curve_types=curve_types)
+ self._from_clause()
+ self._where_clause(
operation_id=operation_id,
start_date=start_date,
end_date=end_date,
prm_id=prm_id,
partner_id=partner_id,
)
+ self._group_clause(date_slot=slot_type)
+ self._order_clause()
)
self.env.cr.execute(query)
return self.env.cr.fetchall()
@api.model
def _get_cdc_by_query_prod(
self,
operation_id,
slot_type,
start_date,
end_date,
prm_id=None,
partner_id=None,
):
"""
Fonction permettant de récupérer les données pour la construction des
chart pour une ou des opérations données pour les consommateurs
:param: int operation_id : Opération concernée
char slot_type: type de slot pour la query ("minute", "hour", "month" ou
"year")
datetime start_date: date début
datetime end_date: date de fin
int prm_id : PRM d'injection à récupérer
int partner_id: contact associé à la courbe
:return: un dictionnaire de données
(labels et data pour les charts à afficher)
"""
label = []
data_autocons = []
data_surplus = []
raw_data = self._cdc_by_query_prod(
operation_id, slot_type, start_date, end_date, prm_id, partner_id
)
for row in raw_data:
label.append(row[0])
data_autocons.append({"x": row[0], "y": round(row[1], 2)})
data_surplus.append({"x": row[0], "y": round(row[2], 2)})
cdc_prod = {
"label": label,
"autocons_prod": data_autocons,
"surplus": data_surplus,
}
return cdc_prod
@api.model
def _get_cdc_by_query_daily_histo_prod(
self, operation_id, start_date, end_date, prm_id=None, partner_id=None
):
"""
Fonction permettant de récupérer les données pour la construction des
chart pour une ou des opérations données pour les consommateurs
:param: int operation_id : Opération concernée
datetime start_date: date début
datetime end_date: date de fin
int prm_id : PRM d'injection à récupérer
int partner_id: contact associé à la courbe
:return: un dictionnaire de données
(labels et data pour les charts à afficher)
"""
label_histo = []
data_autocons_prod_histo = []
data_surplus_histo = []
query = (
self._select_clause(date_slot="day", curve_types=["autoprod", "surplus"])
+ self._from_clause()
+ self._where_clause(
operation_id=operation_id,
start_date=start_date,
end_date=end_date,
prm_id=prm_id,
partner_id=partner_id,
)
+ self._group_clause(date_slot="day")
+ self._order_clause()
)
self.env.cr.execute(query)
raw_data = self.env.cr.fetchall()
for row in raw_data:
label_histo.append(row[0])
data_autocons_prod_histo.append(round(row[1], 2))
data_surplus_histo.append(round(row[2], 2))
cdc_jour = {
"label_histo": label_histo,
"autocons_prod_histo": data_autocons_prod_histo,
"surplus_histo": data_surplus_histo,
}
return cdc_jour
# ------------------------------------------------------
# Functions to manage route
# ------------------------------------------------------