diagram_ph/app/services/diagram_generator.py

342 lines
12 KiB
Python

"""
Service de génération de diagrammes Pression-Enthalpie (PH).
Basé sur le code original diagram_PH.py qui fonctionne correctement.
"""
import io
import base64
import numpy as np
import matplotlib
# Configurer le backend Agg (non-interactif) pour éviter les problèmes sur Windows
matplotlib.use('Agg')
import matplotlib.pyplot as plt
from matplotlib.figure import Figure
from typing import Dict, List, Tuple, Optional, Any
from dataclasses import dataclass
from app.core.refrigerant_loader import RefrigerantLibrary
# Cache pour éviter de recalculer les courbes lourdes par réfrigérant
_diagram_cache: Dict[str, Any] = {}
@dataclass
class DiagramPoint:
"""Point dans le diagramme PH."""
pressure: float # bar
enthalpy: float # kJ/kg
temperature: Optional[float] = None # Celsius
entropy: Optional[float] = None # kJ/kg.K
quality: Optional[float] = None # 0-1
class DiagramGenerator:
"""Générateur de diagrammes Pression-Enthalpie."""
def __init__(self, refrigerant: RefrigerantLibrary):
"""
Initialise le générateur.
Args:
refrigerant: Bibliothèque du réfrigérant
"""
self.refrigerant = refrigerant
self.refrig_name = refrigerant.refrig_name
self.fig_width = 15
self.fig_height = 10
self.dpi = 100
# Utiliser le cache partagé pour éviter des recalculs coûteux
cached = _diagram_cache.get(self.refrig_name)
if cached and isinstance(cached, tuple) and len(cached) == 9:
(
self.Hsl,
self.Hsv,
self.Psat,
self.Tsat,
self.Tmax,
self.Tmin,
self.T_lst,
self.P,
self.IsoT_lst,
) = cached
else:
# Calculer et stocker dans le cache
Hsl, Hsv, Psat, Tsat = self.get_psat_values()
# Assign Tsat early because get_IsoT_values relies on self.Tsat
self.Hsl, self.Hsv, self.Psat, self.Tsat = Hsl, Hsv, Psat, Tsat
Tmax, Tmin, T_lst, P, IsoT_lst = self.get_IsoT_values()
self.Tmax, self.Tmin, self.T_lst, self.P, self.IsoT_lst = Tmax, Tmin, T_lst, P, IsoT_lst
_diagram_cache[self.refrig_name] = (
self.Hsl,
self.Hsv,
self.Psat,
self.Tsat,
self.Tmax,
self.Tmin,
self.T_lst,
self.P,
self.IsoT_lst,
)
def get_psat_values(self) -> Tuple[List[float], List[float], List[float], List[float]]:
"""
Calcule les valeurs de saturation (courbe en cloche).
COPIE EXACTE de diagram_PH.py lignes 39-63
Returns:
Tuple (Hsl, Hsv, Psat, Tsat)
"""
Hsl, Hsv, Psat, Tsat = [], [], [], []
# COPIE EXACTE ligne 49 du code original
for p in np.arange(self.refrigerant.p_begin(), self.refrigerant.p_end(), 0.5e5):
# Lignes 51-57 du code original
Hsl.append(self.refrigerant.hsl_px(p, 0) / 1e3)
Hsv.append(self.refrigerant.hsv_px(p, 1) / 1e3)
# Store Psat in Pa (internal SI), convert to bar only when displaying/exporting
Psat.append(p)
Tsat.append(self.refrigerant.T_px(p, 0.5))
# Lignes 60-61 du code original
if len(Hsl) > 2 and Hsl[-1] == Hsl[-2]:
break
return Hsl, Hsv, Psat, Tsat
def find_whole_10_numbers(self, Tmin: float, Tmax: float) -> np.ndarray:
"""
Trouve les températures rondes (multiples de 10) dans la plage.
COPIE EXACTE de refDLL.py lignes 131-133
Args:
Tmin: Température minimale (°C)
Tmax: Température maximale (°C)
Returns:
Array des températures
"""
# COPIE EXACTE lignes 131-133 de refDLL.py
start = int(Tmin // 10 + 1)
end = int(Tmax // 10 + 1)
return np.arange(start * 10, end * 10, 10)
def get_IsoT_values(self) -> Tuple[float, float, np.ndarray, np.ndarray, List[List[float]]]:
"""
Calcule les valeurs isothermes.
COPIE EXACTE de diagram_PH.py lignes 129-162
Returns:
Tuple (Tmax, Tmin, T_lst, P, IsoT_lst)
"""
# COPIE EXACTE ligne 138 du code original
# T = [self.callref.refrig.T_px(p, 0.5) - 273.15 for p in np.arange(self.callref.refrig.p_begin(), self.callref.refrig.p_end(), 50e5)]
# Lignes 141 du code original
Tmax = max(self.Tsat) - 273.15 - 1
Tmin = min(self.Tsat) - 273.15
# Ligne 144 du code original
T_lst = self.find_whole_10_numbers(Tmin, Tmax)
# Ligne 147 du code original
P = np.arange(self.refrigerant.p_begin(), self.refrigerant.p_end(), 0.05e5)
# Ligne 150 du code original
IsoT_lst = [[self.refrigerant.h_pT(p, temp + 273.15) / 1e3 for p in P] for temp in T_lst]
return Tmax, Tmin, T_lst, P, IsoT_lst
def plot_diagram(
self,
cycle_points: Optional[List[Tuple[float, float]]] = None,
title: Optional[str] = None,
p_min: Optional[float] = None,
p_max: Optional[float] = None,
h_min: Optional[float] = None,
h_max: Optional[float] = None,
include_isotherms: bool = True,
) -> Figure:
"""
Génère le diagramme PH complet.
COPIE EXACTE de diagram_PH.py lignes 183-224
Args:
cycle_points: Points du cycle [(h, p), ...]
title: Titre du diagramme
Returns:
Figure matplotlib
"""
# Configuration des tailles de police - COPIE EXACTE lignes 184-190
SMALL_SIZE = 10
MEDIUM_SIZE = 22
BIGGER_SIZE = 28
plt.rc('font', size=SMALL_SIZE)
plt.rc('axes', titlesize=SMALL_SIZE)
plt.rc('axes', labelsize=MEDIUM_SIZE)
plt.rc('xtick', labelsize=SMALL_SIZE)
plt.rc('ytick', labelsize=SMALL_SIZE)
plt.rc('legend', fontsize=SMALL_SIZE)
plt.rc('figure', titlesize=BIGGER_SIZE)
# Ligne 191 du code original - taille configurable
fig = Figure(figsize=[self.fig_width, self.fig_height])
ax = fig.add_subplot(111)
# Lignes 193-194 du code original: Plot saturation lines
# Psat is stored in Pa internally; plot in bar for readability
ax.plot(self.Hsl, [p / 1e5 for p in self.Psat], 'k-', label='Liquid Saturation')
ax.plot(self.Hsv, [p / 1e5 for p in self.Psat], 'k-', label='Vapor Saturation')
# Lignes 196-202 du code original: Plot isotherms (optional)
if include_isotherms:
for Th_lst, temp in zip(self.IsoT_lst, self.T_lst):
ax.plot(Th_lst, self.P / 1e5, 'g--', label=f'{temp}°C Isotherm', alpha=0.5)
try:
ax.annotate('{:.0f}°C'.format(temp),
(self.refrigerant.h_px(self.refrigerant.p_Tx(temp + 273.15, 0.5), 0.1) / 1e3,
self.refrigerant.p_Tx(temp + 273.15, 0.5) / 1e5),
ha='center',
backgroundcolor="white")
except Exception:
# Non-fatal: annotation failure shouldn't break plotting
pass
# Ligne 204 du code original
ax.set_yscale('log')
# Tracer les points du cycle si fournis (adapté pour l'API)
if cycle_points and len(cycle_points) > 0:
h_cycle = [p[0] for p in cycle_points]
p_cycle = [p[1] for p in cycle_points]
ax.plot(h_cycle, p_cycle, 'r-o', linewidth=2, markersize=8,
label='Cycle', zorder=10)
# Lignes 218-221 du code original
ax.set_xlabel('Enthalpy [kJ/kg]')
ax.set_ylabel('Pressure [bar]')
if title:
ax.set_title(title)
else:
ax.set_title(f'PH Diagram for {self.refrig_name}')
ax.grid(True, which='both', linestyle='--')
# Ligne 223 du code original
ax.legend(loc='best', fontsize=SMALL_SIZE)
# Ligne 224 du code original
fig.tight_layout()
# Apply axis limits if provided (p_min/p_max are in bar, h_min/h_max in kJ/kg)
try:
if p_min is not None or p_max is not None:
y_min = p_min if p_min is not None else ax.get_ylim()[0]
y_max = p_max if p_max is not None else ax.get_ylim()[1]
ax.set_ylim(float(y_min), float(y_max))
if h_min is not None or h_max is not None:
x_min = h_min if h_min is not None else ax.get_xlim()[0]
x_max = h_max if h_max is not None else ax.get_xlim()[1]
ax.set_xlim(float(x_min), float(x_max))
except Exception:
# ignore axis limit errors
pass
return fig
def export_to_base64(self, fig: Figure) -> str:
"""
Exporte la figure en PNG base64.
Args:
fig: Figure matplotlib
Returns:
String base64 de l'image PNG
"""
buf = io.BytesIO()
# NE PAS utiliser bbox_inches='tight' car ça peut tronquer le graphique
# Utiliser pad_inches pour ajouter une marge
fig.savefig(buf, format='png', dpi=self.dpi, bbox_inches='tight', pad_inches=0.2)
buf.seek(0)
img_base64 = base64.b64encode(buf.read()).decode('utf-8')
buf.close()
return img_base64
def generate_diagram_data(
self,
cycle_points: Optional[List[Tuple[float, float]]] = None
) -> Dict[str, Any]:
"""
Génère les données du diagramme en format JSON.
Utilise les données déjà calculées à l'initialisation.
Args:
cycle_points: Points du cycle
Returns:
Dictionnaire avec les données du diagramme
"""
data = {
"refrigerant": self.refrig_name,
"saturation_curve": [
{"enthalpy": float(h), "pressure": float(p / 1e5)}
for h, p in zip(self.Hsl + self.Hsv, self.Psat + self.Psat)
]
}
# Ajouter isothermes
isotherms_data = []
for Th_lst, temp in zip(self.IsoT_lst, self.T_lst):
points = []
for h, p in zip(Th_lst, self.P / 1e5):
if h is not None and not np.isnan(h):
points.append({"enthalpy": float(h), "pressure": float(p)})
if len(points) > 0:
isotherms_data.append({
"temperature": float(temp),
"points": points
})
data["isotherms"] = isotherms_data
# Ajouter points du cycle
if cycle_points:
data["cycle_points"] = [
{"enthalpy": float(h), "pressure": float(p)}
for h, p in cycle_points
]
return data
def generate_complete_diagram(
self,
cycle_points: Optional[List[Tuple[float, float]]] = None,
title: Optional[str] = None,
export_format: str = "both"
) -> Dict[str, Any]:
"""
Génère le diagramme complet.
Utilise les données déjà calculées à l'initialisation.
Args:
cycle_points: Points du cycle
title: Titre
export_format: "png", "json", "both"
Returns:
Dictionnaire avec image et/ou données
"""
result = {}
if export_format in ["png", "both"]:
fig = self.plot_diagram(cycle_points, title)
result["image_base64"] = self.export_to_base64(fig)
plt.close(fig)
if export_format in ["json", "both"]:
result["data"] = self.generate_diagram_data(cycle_points)
return result