Sélectionner une révision Git
acc_operation.py 19,97 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, time
from dateutil.relativedelta import relativedelta
from odoo import _, models
from odoo.exceptions import ValidationError
from odoo.osv import expression
from odoo.tools import date_utils
class AccOperation(models.Model):
_inherit = "acc.operation"
# ------------------------------------------------------
# Actions
# ------------------------------------------------------
def action_view_curves(self):
"""
Action qui ouvre le portail sur l'opération
:return: Vue Qweb
"""
self.ensure_one()
if self.id:
return {
"type": "ir.actions.act_url",
"url": "/operation/%s" % (self.id),
"target": "new",
}
# ------------------------------------------------------
# Business methods
# ------------------------------------------------------
def get_last_month_power_tot(self):
"""
Fonction retournant la consommation totale, l'autoconsommation totale
et la production totale du dernier mois.
:return: la somme de la conso totale de tous les consommateurs,
la somme de l autoconso totale de tous les consommateurs,
la sommme de la prod totale de tous les producteurs
la date de début de mois
"""
self.ensure_one()
# Get last date slot recorded
last_record = self.get_last_cdc_record()
start_date, end_date = self.get_interval("month", last_record)
end_date_month = end_date + relativedelta(days=1)
query = (
self._select_clause("month", ["cons", "prod", "autocons"])
+ self._from_clause()
+ self._where_clause(
self.id,
start_date,
end_date_month,
)
+ self._group_clause("month")
+ self._order_clause()
)
self.env.cr.execute(query)
raw_data = self.env.cr.fetchone()
return raw_data[1], raw_data[2], raw_data[3], start_date
def get_last_cdc_record(self, partner_id=None):
"""
@returns: last acc.enedis.cdc record in operations
"""
domain = [("acc_operation_id", "in", self.ids)]
if partner_id:
domain = expression.AND([domain, [("partner_id", "=", partner_id)]])
# Get last date slot recorded
last_record = (
self.env["acc.enedis.cdc"]
.sudo()
.search(
domain,
limit=1,
order="date_slot DESC",
)
)
if not last_record:
raise ValidationError(_("L'opération ne possède pas de données"))
return last_record
def get_interval(self, scale, cdc_record):
"""
Fonction retournant une date de début et une date de fin.
Ces dates sont calculées en fonction de l'échelle choisie et du dernier
élément enregistré
- day: la date de début et la date de fin = dernier jour de données
- week: date de fin = fin du jour du dernier jour de données
date de début = date de fin moins 7 jours
- month: date de fin = fin du mois du dernier jour de données
date de début = début du mois du dernier jour de données
- year: date de fin = fin du mois du dernier jour de données
date de début = début de l'année du dernier jour de données
:param str scale: type d'affichage des graphes
(day/week/month/year)
:param object cdc_record: courbe de charge (table acc.enedis.cdc)
@returns: une date de début et une date de fin
"""
# Convert end datetime to timezone
last_day_start = datetime.combine(cdc_record.date_slot, time.min)
last_day_end = datetime.combine(cdc_record.date_slot, time.max)
start_month, end_month = date_utils.get_month(cdc_record.date_slot)
if scale == "year":
end_date = end_month
start_date = start_month.replace(month=1)
elif scale == "month":
end_date = end_month
start_date = start_month
elif scale == "week":
end_date = last_day_end
start_date = last_day_end - relativedelta(days=7)
elif scale == "day":
end_date = last_day_end
start_date = last_day_start
else:
raise ValueError(
_(
"La période d'affichage est incorrecte : %s, "
"valeurs attendues : day/week/month/year"
)
% scale
)
return start_date, end_date
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 > 0 and delta < 31:
step_display_curve = "day"
display_hourly_curves = True
elif delta >= 31 and delta <= 365:
step = "month"
step_display_curve = "month"
elif delta > 365:
step = "year"
step_display_curve = "year"
return display_hourly_curves, step, step_display_curve
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 "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 ("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
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
"""
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
datetime start_date : first datetime to be retrieved
datetime end_date : last datetime 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"))
result = f"""
WHERE cdc.acc_operation_id = {operation_id}
AND cdc.date_slot >= '{start_date}'
AND cdc.date_slot < '{end_date}'
"""
if prm_id and isinstance(prm_id, int):
if extra_curve_type and extra_curve_type in (
"cons",
"autocons",
"allocons",
"prod",
"surplus",
"autoprod",
):
result += f"""
AND (cdc.comp_data_type = '{extra_curve_type}'
OR cdc.acc_counter_id = {prm_id})
"""
else:
result += f"""
AND cdc.acc_counter_id = {prm_id}
"""
if partner_id and isinstance(partner_id, int):
result += f" AND cdc.partner_id = {partner_id} "
return result
def _group_clause(self, date_slot):
"""
Function to build GROUP BY section of query for retrieving curves
@param
char date_slot : granularity (one of "hour", "day", "month", "year")
"""
if date_slot not in ("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')
"""
def _order_clause(self):
return "ORDER BY date_slot ASC;"
def get_cdc_by_query_cons(
self, 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: char slot_type: type de slot pour la query ("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: un dictionnaire de données
(labels et data pour les charts à afficher)
"""
self.ensure_one()
label = []
data_autocons = []
data_allocons = []
data_cons = []
data_prod = []
query = (
self._select_clause(slot_type, ["cons", "autocons", "prod"])
+ self._from_clause()
+ self._where_clause(
self.id, start_date, end_date, prm_id, "prod", partner_id
)
+ "AND cdc.comp_data_type IN ('autocons', 'cons', 'prod')"
+ self._group_clause(slot_type)
+ self._order_clause()
)
self.env.cr.execute(query)
raw_data = self.env.cr.fetchall()
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
def get_cdc_by_query_daily_histo_cons(
self, 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: 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("day", ["autocons", "allocons"])
+ self._from_clause()
+ self._where_clause(self.id, start_date, end_date, prm_id, partner_id)
+ self._group_clause("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
def get_cdc_by_query_prod(
self, 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: char slot_type: type de slot pour la query ("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 = []
query = (
self._select_clause(slot_type, ["autoprod", "surplus"])
+ self._from_clause()
+ self._where_clause(self.id, start_date, end_date, prm_id, partner_id)
+ self._group_clause(slot_type)
+ self._order_clause()
)
self.env.cr.execute(query)
raw_data = self.env.cr.fetchall()
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
def get_cdc_by_query_daily_histo_prod(
self, 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: 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("day", ["autoprod", "surplus"])
+ self._from_clause()
+ self._where_clause(self.id, start_date, end_date, prm_id, partner_id)
+ self._group_clause("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
def get_date_min_max(self, partner_id=None):
self.ensure_one()
last_record = self.get_last_cdc_record(partner_id)
date_max = last_record.date_slot.strftime("%d/%m/%Y")
if partner_id:
first_date = self.env["acc.counter.period"].search(
[("acc_operation_id", "=", self.id), ("partner_id", "=", partner_id)],
order="start_date ASC",
limit=1,
)
date_min = first_date.start_date.strftime("%d/%m/%Y")
else:
date_min = self.date_start_contract.strftime("%d/%m/%Y")
return {"date_min": date_min, "date_max": date_max}
# ------------------------------------------------------
# Functions to manage route
# ------------------------------------------------------
def graph_view_global(
self,
start_date=None,
end_date=None,
partner_id=None,
prm_id=None,
data_type=None,
):
"""
Fonction appelée pour l'affichage des courbes globales
sur le portail
:param: start_date: date début
end_date: date de fin
partner_id: données uniquement de ce contact
prm_id: données uniquement de ce PRM (mutuellement exclusif de
partner_id)
data_type: type de courbes à afficher:
- "pmo" : vue globale (cons + prod sans filtrage)
- "cons" : vue mon suivi conso (avec filtrage possible)
- "prod" : vue mon suivi production (avec filtrage possible)
@returns: dictionnaire pour la construction des graphes
"""
self.ensure_one()
result_graph = {}
start_date = datetime.strptime(start_date, "%d/%m/%Y")
end_date = datetime.combine(datetime.strptime(end_date, "%d/%m/%Y"), time.max)
display_hourly_curves, step_curve, step_display_curve = self.get_step_from_date(
start_date=start_date, end_date=end_date
)
chart_data = {}
if data_type == "cons" or data_type == "pmo":
chart_data_cons = self.get_cdc_by_query_cons(
step_curve, start_date, end_date, prm_id, partner_id
)
if display_hourly_curves:
chart_data_histo = self.get_cdc_by_query_daily_histo_cons(
start_date, end_date, prm_id, partner_id
)
chart_data_cons.update(chart_data_histo)
chart_data.update(chart_data_cons)
if data_type == "prod" or data_type == "pmo":
chart_data_prod = self.get_cdc_by_query_prod(
step_curve, start_date, end_date, prm_id, partner_id
)
if display_hourly_curves:
chart_data_histo = self.get_cdc_by_query_daily_histo_prod(
start_date, end_date, prm_id, partner_id
)
chart_data_prod.update(chart_data_histo)
chart_data.update(chart_data_prod)
result_graph["chart_data"] = chart_data
result_graph.update(self.get_date_min_max(partner_id))
result_graph.update(
{
"start_date": start_date,
"end_date": end_date,
"scale": step_display_curve,
"is_curve_line": display_hourly_curves,
}
)
return result_graph