"""
Version: 1.2.1 (2024-12-20)
Auteur: Kahina et franck - Groupe 2
Commit: bigmoletos@yopmail.com
Routes principales pour l'API IHM.
Ce module gère les routes principales de l'application:
- Page d'accueil avec formulaire de prédiction
- Communication avec l'API de modélisation
- Affichage des résultats
- Informations sur le modèle
Routes:
- /: Page d'accueil avec formulaire
- /predict: Endpoint de prédiction
- /model-info: Informations sur le modèle
- /health: Vérification de l'état de l'API
Dépendances:
- httpx pour les requêtes asynchrones
- flask pour le routage
- database.models pour l'accès aux données
"""
from flask import Blueprint, render_template, request, jsonify, flash, redirect, send_from_directory, send_file, url_for
from flask_login import login_required, current_user
import httpx
from datetime import datetime
import json
import logging
from database.models import db, Simulation
import os
import requests
import pandas as pd
import numpy as np
import traceback
from werkzeug.exceptions import NotFound
# Configuration du logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
main_bp = Blueprint('main', __name__)
# Configuration du chemin des images UML
current_dir = os.path.dirname(os.path.abspath(__file__)) # Dossier main
project_root = os.path.abspath(
os.path.join(current_dir, '..', '..', '..',
'..')) # Remonte jusqu'à projet_qualite_air
DOCS_FOLDER = os.path.join(project_root, 'docs')
[docs]
@main_bp.route('/docs/<path:filename>')
def serve_docs(filename):
"""Sert les fichiers statiques depuis le dossier docs"""
logger.info(f"Demande de fichier: {filename} depuis {DOCS_FOLDER}")
return send_from_directory(DOCS_FOLDER, filename)
# Configuration de l'URL de l'API en fonction de l'environnement
[docs]
def get_api_url():
"""Détermine l'URL de l'API de modélisation selon l'environnement"""
# Vérifier si on est dans un conteneur Docker
if os.path.exists('/.dockerenv'):
# En production (Docker)
return os.environ.get('MODEL_API_URL', 'http://api_modelisation:5000')
else:
# En développement local
return os.environ.get('MODEL_API_URL', 'http://127.0.0.1:8092')
# Utiliser la fonction pour définir l'URL
MODEL_API_URL = get_api_url()
logger.info(f"Configuration de l'API de modélisation: {MODEL_API_URL}")
[docs]
async def get_model_info():
async with httpx.AsyncClient() as client:
try:
response = await client.get(f"{MODEL_API_URL}/model/info",
timeout=30.0)
return response.json()
except httpx.ConnectError:
logger.error(f"Impossible de se connecter à {MODEL_API_URL}")
return {"error": "Service de modélisation indisponible"}
except Exception as e:
logger.error(
f"Erreur lors de la récupération des infos du modèle: {e}")
return {"error": "Erreur interne"}
[docs]
async def make_prediction(features):
async with httpx.AsyncClient() as client:
try:
response = await client.post(f"{MODEL_API_URL}/model/predict",
json=features,
timeout=30.0)
response.raise_for_status()
return response.json()
except httpx.ConnectError:
logger.error(f"Impossible de se connecter à {MODEL_API_URL}")
return {"error": "Service de modélisation indisponible"}
except httpx.HTTPStatusError as e:
logger.error(
f"Erreur HTTP {e.response.status_code}: {e.response.text}")
return {
"error":
f"Erreur du service de modélisation: {e.response.text}"
}
except Exception as e:
logger.error(f"Erreur lors de la prédiction: {e}")
return {"error": "Erreur interne"}
[docs]
@main_bp.route('/')
def index():
"""
Route de la page d'accueil.
Version: 1.0.1 (2024-03-19 16:45)
"""
try:
model_info = httpx.get(f"{MODEL_API_URL}/model/info").json()
return render_template('index.html',
model_info=model_info,
current_time=datetime.now())
except Exception as e:
return render_template('error.html', error_message=str(e))
[docs]
@main_bp.route('/predict', methods=['POST'])
@login_required
def predict():
"""
Traite la soumission du formulaire de prédiction.
"""
features = request.form.to_dict()
try:
logger.debug(f"Features reçues brutes: {features}")
# Convertir les valeurs en nombres
processed_features = {}
for key, value in features.items():
try:
# Gérer les valeurs numériques
if key in [
'year', 'month', 'day', 'hour', 'TEMP', 'DEWP', 'PRES',
'Iws', 'Is', 'Ir'
]:
processed_features[key] = float(value)
# Gérer la direction du vent
elif key == 'cbwd':
processed_features[key] = value
else:
processed_features[key] = float(value)
except ValueError:
processed_features[key] = value
logger.debug(f"Features traitées: {processed_features}")
# Calculer les features dérivées
month = processed_features.get('month', 0)
hour = processed_features.get('hour', 0)
temp = processed_features.get('TEMP', 0)
dewp = processed_features.get('DEWP', 0)
pres = processed_features.get('PRES', 0)
cbwd = processed_features.get('cbwd', 'NW')
iws = processed_features.get('Iws', 0)
# Ajouter les features calculées
import math
processed_features.update({
'month_sin':
math.sin(2 * math.pi * month / 12),
'month_cos':
math.cos(2 * math.pi * month / 12),
'hour_sin':
math.sin(2 * math.pi * hour / 24),
'hour_cos':
math.cos(2 * math.pi * hour / 24),
'wind_x':
iws * {
'NW': -0.7071,
'NE': 0.7071,
'SE': 0.7071,
'SW': -0.7071,
'cv': 0
}.get(cbwd, 0),
'wind_y':
iws * {
'NW': 0.7071,
'NE': 0.7071,
'SE': -0.7071,
'SW': -0.7071,
'cv': 0
}.get(cbwd, 0),
'dewpoint_spread':
temp - dewp,
'relative_humidity':
100 * (math.exp((17.625 * dewp) / (243.04 + dewp)) / math.exp(
(17.625 * temp) / (243.04 + temp))),
'temp_wind':
temp * iws,
'pres_temp':
pres * temp,
'rolling_temp':
temp, # Simplifié car nous n'avons pas l'historique
'rolling_wind':
iws # Simplifié car nous n'avons pas l'historique
})
logger.debug(f"Features finales avec calculs: {processed_features}")
# Faire la prédiction
prediction_result = httpx.post(f"{MODEL_API_URL}/model/predict",
json={
"features": processed_features
},
timeout=30.0).json()
if "error" in prediction_result:
raise Exception(prediction_result["error"])
# Extraire la valeur de prédiction
prediction_value = float(prediction_result['prediction'][0])
# Sauvegarder la simulation dans la base de données
simulation = Simulation(user_id=current_user.id,
input_data=processed_features,
result=prediction_value)
db.session.add(simulation)
db.session.commit()
return render_template(
'prediction_result.html',
prediction=f"{prediction_value:.2f}",
features=features,
who_limit=10.0 # Limite OMS pour PM2.5
)
except Exception as e:
logger.error(f"Erreur lors de la prédiction : {str(e)}")
return render_template('error.html', error_message=str(e))
[docs]
@main_bp.route('/model-info')
def model_info():
try:
response = httpx.get(f"{MODEL_API_URL}/model/info")
model_info = response.json()
logger.debug(f"Réponse de l'API: {model_info}")
# Extraire les informations du modèle de la réponse
model_data = model_info.get('model_info', {})
# Formater les métriques
metrics = model_data.get('metrics', {})
formatted_metrics = {
"MSE": f"{metrics.get('mse', 0):.2f}",
"RMSE": f"{metrics.get('rmse', 0):.2f}",
"MAE": f"{metrics.get('mae', 0):.2f}",
"R²": f"{metrics.get('r2', 0):.4f}"
}
# Extraire et formater les feature importance
feature_importance = model_data.get('feature_importance', {})
features = feature_importance.get('feature', {})
importance = feature_importance.get('importance', {})
# Créer une liste triée des features avec leur importance
feature_list = [
{
"name": features.get(str(i), f"Feature {i}"),
"importance": float(importance.get(str(i), 0)) *
100 # Convertir en pourcentage
} for i in range(len(features))
]
# Trier par importance décroissante
feature_list.sort(key=lambda x: x["importance"], reverse=True)
# Formater les pourcentages
for feature in feature_list:
feature["importance"] = f"{feature['importance']:.2f}%"
# Extraire les statistiques des données
preprocessing_info = model_data.get('preprocessing_info', {})
data_stats = preprocessing_info.get('data_stats', {})
formatted_stats = {
"Échantillons d'entraînement":
f"{int(data_stats.get('n_train', 0)):,}",
"Échantillons de test": f"{int(data_stats.get('n_test', 0)):,}",
"PM2.5 minimum": f"{float(data_stats.get('y_min', 0)):.1f}",
"PM2.5 maximum": f"{float(data_stats.get('y_max', 0)):.1f}",
"PM2.5 moyen": f"{float(data_stats.get('y_mean', 0)):.1f}",
"Écart-type PM2.5": f"{float(data_stats.get('y_std', 0)):.1f}"
}
return render_template(
'model_info.html',
metrics=formatted_metrics,
feature_importance=feature_list,
data_stats=formatted_stats,
model_status=model_info.get('status', 'inconnu'),
cuda_available=model_info.get('cuda_available', False))
except Exception as e:
logger.error(
f"Erreur lors de la récupération des informations du modèle : {str(e)}"
)
logger.error(f"Type d'erreur: {type(e).__name__}")
import traceback
logger.error(f"Traceback:\n{traceback.format_exc()}")
return render_template(
'error.html',
error_message=
"Une erreur est survenue lors de la récupération des informations du modèle.",
error_details=str(e),
suggestions=[
"Vérifiez que l'API de modélisation est accessible",
"Vérifiez que le modèle est correctement chargé",
"Consultez les logs pour plus de détails"
])
[docs]
@main_bp.route('/health')
def health_check():
"""
Vérifie l'état de santé de l'API
"""
return jsonify({"status": "healthy", "service": "api_ihm"})
[docs]
@main_bp.route('/predictions')
@login_required
def prediction_history():
"""
Affiche l'historique des prédictions.
Version: 1.0.0 (2024-03-19 17:30)
"""
if current_user.is_admin:
# Admin voit toutes les prédictions
predictions = Simulation.query.order_by(Simulation.date.desc()).all()
else:
# Utilisateur normal ne voit que ses prédictions
predictions = Simulation.query.filter_by(
user_id=current_user.id).order_by(Simulation.date.desc()).all()
return render_template('prediction_history.html', predictions=predictions)
[docs]
@main_bp.route('/replay-prediction/<int:prediction_id>', methods=['POST'])
@login_required
def replay_prediction(prediction_id):
try:
# Récupérer la prédiction originale
original_prediction = Simulation.query.get_or_404(prediction_id)
if not original_prediction:
flash('Prédiction non trouvée.', 'error')
return redirect(url_for('main.prediction_history'))
# Créer une nouvelle simulation avec les mêmes paramètres
new_simulation = Simulation(user_id=current_user.id,
input_data=original_prediction.input_data,
result=original_prediction.result,
date=datetime.now())
try:
db.session.add(new_simulation)
db.session.commit()
flash('Prédiction rejouée avec succès!', 'success')
return redirect(
url_for('main.prediction_result',
prediction_id=new_simulation.id))
except Exception as e:
db.session.rollback()
logger.error(
f"Erreur lors de l'enregistrement de la nouvelle prédiction: {str(e)}"
)
flash(
"Une erreur est survenue lors de la sauvegarde de la prédiction.",
'error')
return redirect(url_for('main.prediction_history'))
except Exception as e:
logger.error(
f"Erreur lors du rejeu de la prédiction {prediction_id}: {str(e)}")
flash("Une erreur est survenue lors du rejeu de la prédiction.",
'error')
return redirect(url_for('main.prediction_history'))
[docs]
@main_bp.route('/wiki')
def wiki():
"""
Route pour afficher la page wiki contenant les informations sur l'impact
des conditions météorologiques sur les PM2.5.
Returns:
str: Template HTML de la page wiki
"""
return render_template('wiki.html')
[docs]
@main_bp.route('/sources')
def sources():
"""
Route pour afficher la page des sources et références sur la pollution de l'air.
Returns:
str: Template HTML de la page sources
"""
return render_template('sources.html')
[docs]
def get_environment_type():
"""Détermine le type d'environnement (production, test, local)"""
try:
python_env = os.getenv('PYTHON_ENV', '').lower()
is_docker = os.path.exists('/.dockerenv')
is_windows = os.name == 'nt'
logger.info(f"PYTHON_ENV: {python_env}")
logger.info(f"Is Docker: {is_docker}")
logger.info(f"Is Windows: {is_windows}")
logger.info(f"Current working directory: {os.getcwd()}")
# Si nous sommes sur Windows, c'est forcément local, peu importe PYTHON_ENV
if is_windows or os.getcwd().startswith('C:'):
logger.info("Environnement Windows détecté -> local")
return 'local'
# Sinon, on vérifie la variable d'environnement
elif python_env == 'production':
logger.info("Environnement production détecté via PYTHON_ENV")
return 'production'
elif python_env == 'test':
logger.info("Environnement test détecté via PYTHON_ENV")
return 'test'
# Par défaut en test si dans Docker
elif is_docker:
logger.info("Conteneur Docker détecté -> test")
return 'test'
# Par défaut
else:
logger.info("Environnement par défaut -> local")
return 'local'
except Exception as e:
logger.error(
f"Erreur lors de la détection de l'environnement: {str(e)}")
return 'local' # Par défaut en cas d'erreur
[docs]
def get_base_paths():
"""
Retourne les chemins de base selon l'environnement (production, test ou local).
Cette fonction détermine les chemins d'accès aux fichiers et URLs selon l'environnement:
- En production/test dans Docker: fichiers dans /app, URLs avec préfixe /testihm/...
- En production/test sur VM: fichiers et URLs dans /home/ubuntu/testihm/...
- En local (Windows): fichiers dans le dossier du projet, URLs relatives
Returns:
tuple: (paths, local_paths) où chaque dictionnaire contient:
- base_path: Chemin racine
- uml_path: Chemin des schémas UML
- data_path: Chemin des données
- uml_url_prefix: Préfixe URL pour les schémas UML
- data_url_prefix: Préfixe URL pour les données
- physical_uml_path: Chemin physique des schémas UML
- physical_data_path: Chemin physique des données
"""
logger.info("=== Début de get_base_paths() ===")
try:
# Initialisation des dictionnaires de chemins
paths = {}
local_paths = {}
logger.debug("Dictionnaires de chemins initialisés")
try:
# Détection de l'environnement
env_type = get_environment_type()
logger.info(f"Type d'environnement détecté: {env_type}")
logger.debug(
f"Variables d'environnement: PYTHON_ENV={os.getenv('PYTHON_ENV')}"
)
except Exception as e:
logger.error(
f"Erreur lors de la détection de l'environnement: {str(e)}")
logger.debug(f"Traceback: {traceback.format_exc()}")
raise
try:
# Vérification si on est dans Docker
is_docker = os.path.exists('/.dockerenv')
logger.info(f"Exécution dans Docker: {is_docker}")
logger.debug(f"Dossier courant: {os.getcwd()}")
except Exception as e:
logger.error(f"Erreur lors de la vérification Docker: {str(e)}")
logger.debug(f"Traceback: {traceback.format_exc()}")
is_docker = False
try:
if env_type in ['production', 'test']:
logger.info(
f"Configuration des chemins pour environnement {env_type}")
# Définition des chemins de base
vm_base = '/home/ubuntu/testihm/services/api_ihm'
url_prefix = '/testihm/services/api_ihm'
logger.debug(
f"Chemins de base - VM: {vm_base}, URL: {url_prefix}")
try:
if is_docker:
logger.info("Configuration des chemins pour Docker")
# Dans Docker, les fichiers sont montés dans /app
paths = {
'base_path': '/app',
'uml_path': '/app/uml/schemas',
'data_path': '/app/data/datasets',
'uml_url_prefix': f'{url_prefix}/uml/schemas',
'data_url_prefix': f'{url_prefix}/data/datasets',
'physical_uml_path': '/app/uml/schemas',
'physical_data_path': '/app/data/datasets'
}
logger.debug(
f"Chemins Docker configurés: {json.dumps(paths, indent=2)}"
)
# Vérification des chemins Docker
try:
for key, path in paths.items():
if os.path.exists(path):
logger.info(
f"✓ Chemin Docker {key} existe: {path}"
)
if os.path.isdir(path):
contents = os.listdir(path)
logger.debug(
f"Contenu de {key}: {contents}")
logger.info(
f" → Contient {len(contents)} fichiers/dossiers"
)
else:
logger.warning(
f"✗ Chemin Docker {key} n'existe pas: {path}"
)
except Exception as e:
logger.error(
f"Erreur lors de la vérification des chemins Docker: {str(e)}"
)
logger.debug(
f"Traceback: {traceback.format_exc()}")
else:
logger.info("Configuration des chemins pour VM")
# Sur la VM, utilisation des chemins complets
paths = {
'base_path': vm_base,
'uml_path': f"{vm_base}/uml/schemas",
'data_path': f"{vm_base}/data/datasets",
'uml_url_prefix': f'{url_prefix}/uml/schemas',
'data_url_prefix': f'{url_prefix}/data/datasets',
'physical_uml_path': f"{vm_base}/uml/schemas",
'physical_data_path': f"{vm_base}/data/datasets"
}
logger.debug(
f"Chemins VM configurés: {json.dumps(paths, indent=2)}"
)
# Vérification des chemins VM
try:
for key, path in paths.items():
if os.path.exists(path):
logger.info(
f"✓ Chemin VM {key} existe: {path}")
if os.path.isdir(path):
contents = os.listdir(path)
logger.debug(
f"Contenu de {key}: {contents}")
logger.info(
f" → Contient {len(contents)} fichiers/dossiers"
)
else:
logger.warning(
f"✗ Chemin VM {key} n'existe pas: {path}"
)
except Exception as e:
logger.error(
f"Erreur lors de la vérification des chemins VM: {str(e)}"
)
logger.debug(
f"Traceback: {traceback.format_exc()}")
except Exception as e:
logger.error(
f"Erreur lors de la configuration des chemins {env_type}: {str(e)}"
)
logger.debug(f"Traceback: {traceback.format_exc()}")
raise
else:
try:
logger.info(
"Configuration des chemins pour environnement local (Windows)"
)
# Construction des chemins locaux
current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.abspath(
os.path.join(current_dir, '..', '..', '..', '..'))
logger.debug(f"Dossier courant: {current_dir}")
logger.debug(f"Racine du projet: {project_root}")
local_paths = {
'base_path':
project_root,
'uml_path':
os.path.join(project_root, 'docs', 'uml',
'schemas').replace('\\', '/'),
'data_path':
os.path.join(project_root, 'services', 'api_ihm',
'src', 'database').replace('\\', '/'),
'uml_url_prefix':
'/docs/uml/schemas',
'data_url_prefix':
'/data/datasets',
'physical_uml_path':
os.path.join(project_root, 'docs', 'uml',
'schemas').replace('\\', '/'),
'physical_data_path':
os.path.join(project_root, 'services', 'api_ihm',
'src', 'database').replace('\\', '/')
}
logger.debug(
f"Chemins locaux configurés: {json.dumps(local_paths, indent=2)}"
)
# Vérification des chemins locaux
try:
for key, path in local_paths.items():
if os.path.exists(path):
logger.info(
f"✓ Chemin local {key} existe: {path}")
if os.path.isdir(path):
contents = os.listdir(path)
logger.debug(
f"Contenu de {key}: {contents}")
logger.info(
f" → Contient {len(contents)} fichiers/dossiers"
)
else:
logger.warning(
f"✗ Chemin local {key} n'existe pas: {path}"
)
except Exception as e:
logger.error(
f"Erreur lors de la vérification des chemins locaux: {str(e)}"
)
logger.debug(f"Traceback: {traceback.format_exc()}")
except Exception as e:
logger.error(
f"Erreur lors de la configuration des chemins locaux: {str(e)}"
)
logger.debug(f"Traceback: {traceback.format_exc()}")
raise
logger.info("=== Fin de get_base_paths() avec succès ===")
return paths, local_paths
except Exception as e:
logger.error(
f"Erreur lors de la configuration des chemins: {str(e)}")
logger.debug(f"Traceback: {traceback.format_exc()}")
raise
except Exception as e:
logger.error(f"Erreur générale dans get_base_paths: {str(e)}")
logger.error(f"Traceback complet: {traceback.format_exc()}")
# En cas d'erreur, retourner des chemins par défaut
default_paths = {
'base_path': '.',
'uml_path': './docs/uml/schemas',
'data_path': './src/database',
'uml_url_prefix': '/docs/uml/schemas',
'data_url_prefix': '/data/datasets',
'physical_uml_path': './docs/uml/schemas',
'physical_data_path': './src/database'
}
logger.info("Utilisation des chemins par défaut suite à une erreur")
logger.debug(
f"Chemins par défaut: {json.dumps(default_paths, indent=2)}")
return default_paths, default_paths
[docs]
def is_test_environment():
"""Détermine si l'application est en environnement de test/production"""
return get_environment_type() in ['production', 'test']
[docs]
@main_bp.route('/uml', methods=['GET'])
def uml():
"""
Route pour afficher la page des schémas UML du projet et gestion de projet.
Gère l'affichage des diagrammes UML depuis les chemins appropriés selon l'environnement.
"""
try:
logger.info("=== Début du traitement de la route /uml ===")
try:
env_type = get_environment_type()
paths, local_paths = get_base_paths()
except Exception as e:
logger.error(
f"Erreur lors de la récupération des chemins: {str(e)}")
raise
logger.info(f"Type d'environnement : {env_type}")
logger.info(f"Variable PYTHON_ENV : {os.getenv('PYTHON_ENV')}")
logger.info(
f"Existence de /.dockerenv : {os.path.exists('/.dockerenv')}")
logger.info(f"Dossier courant : {os.getcwd()}")
try:
logger.info(f"Contenu du dossier courant : {os.listdir('.')}")
except Exception as e:
logger.error(
f"Erreur lors de la lecture du dossier courant: {str(e)}")
try:
if env_type == 'production':
base_path = paths['uml_path']
url_prefix = paths['uml_url_prefix']
elif env_type == 'test':
base_path = paths['uml_path']
url_prefix = paths['uml_url_prefix']
else:
base_path = local_paths['uml_path']
url_prefix = local_paths['uml_url_prefix']
logger.info(
f"Chemin de base dans le système de fichiers: {base_path}")
logger.info(f"Préfixe URL pour les images: {url_prefix}")
logger.info(
f"Existence du chemin de base: {os.path.exists(base_path)}")
# Catégories de diagrammes à chercher
categories = [
'architecture', 'classes', 'sequences', 'composants',
'deploiements', 'gestion_projet'
]
# Dictionnaire pour stocker les chemins des images par catégorie
image_paths = {}
for category in categories:
try:
category_path = os.path.join(base_path, category)
logger.info(f"Vérification du dossier: {category_path}")
logger.info(
f"Existence du dossier {category}: {os.path.exists(category_path)}"
)
if os.path.exists(category_path):
try:
# Liste tous les fichiers PNG dans le dossier
png_files = [
f for f in os.listdir(category_path)
if f.lower().endswith('.png')
]
# Construit les chemins complets pour chaque image
image_paths[category] = [
f'{url_prefix}/{category}/{png}'
for png in png_files
]
logger.info(
f"Images trouvées pour {category}: {png_files}"
)
except Exception as e:
logger.error(
f"Erreur lors du traitement des fichiers PNG pour {category}: {str(e)}"
)
image_paths[category] = []
else:
logger.warning(f"Dossier non trouvé: {category_path}")
image_paths[category] = []
except Exception as e:
logger.error(
f"Erreur lors du traitement de la catégorie {category}: {str(e)}"
)
image_paths[category] = []
logger.info(
f"Chemins des images UML et gestion projet finaux : {image_paths}"
)
return render_template('uml.html', image_paths=image_paths)
except Exception as e:
logger.error(
f"Erreur lors du traitement des chemins UML: {str(e)}")
raise
except Exception as e:
logger.error(f"Erreur lors du chargement des diagrammes UML: {str(e)}")
logger.error(f"Traceback complet: {traceback.format_exc()}")
return render_template(
'error.html',
error_message="Erreur lors du chargement des diagrammes UML",
error_details=str(e),
suggestions=[
"Vérifiez que les images UML existent aux emplacements suivants :",
f"- Environnement détecté : {env_type if 'env_type' in locals() else 'inconnu'}",
f"- Chemin de base utilisé : {base_path if 'base_path' in locals() else 'inconnu'}",
f"- Préfixe URL : {url_prefix if 'url_prefix' in locals() else 'inconnu'}"
])
[docs]
@main_bp.route('/graphiques')
def graphiques():
"""Route pour afficher la page des graphiques"""
return render_template('graph.html')
[docs]
@main_bp.route('/api/plot-data')
def get_plot_data():
"""Route API pour récupérer les données des graphiques"""
try:
env = get_environment_type()
paths, local_paths = get_base_paths()
logger.info(f"Environnement détecté : {env}")
# Sélectionner les chemins selon l'environnement
if env == 'production':
file_path = os.path.join(paths['data_path'],
'environment_data_analysis.csv')
logger.info("Utilisation des chemins de production")
elif env == 'test':
file_path = os.path.join(paths['data_path'],
'environment_data_analysis.csv')
logger.info("Utilisation des chemins de test")
else:
file_path = os.path.join(local_paths['data_path'],
'environment_data_analysis.csv')
logger.info("Utilisation des chemins locaux")
logger.info(f"Tentative d'accès au fichier : {file_path}")
# Vérifier le contenu du répertoire pour le debug
try:
dir_path = os.path.dirname(file_path)
logger.info(
f"Contenu du répertoire {dir_path}: {os.listdir(dir_path)}")
except Exception as e:
logger.error(f"Erreur lors de la lecture du répertoire: {str(e)}")
if not os.path.exists(file_path):
logger.error(f"Le fichier n'existe pas : {file_path}")
if env == 'production':
message = (
'Fichier de données non trouvé dans le conteneur. '
'Vérifiez que le fichier est bien présent dans /testihm/data/datasets/ '
'sur le serveur et que les permissions sont correctes (chmod 666).'
)
elif env == 'test':
message = (
'Fichier de données non trouvé dans le conteneur. '
'Vérifiez que le fichier est bien présent dans /testihm/data/datasets/ '
'sur le serveur et que les permissions sont correctes (chmod 666).'
)
else:
message = (
'Fichier de données non trouvé. '
'Vérifiez que le fichier est présent dans src/database/')
return jsonify({'error': message}), 404
# Charger les données
df = pd.read_csv(file_path)
logger.info(f"Données brutes chargées : {df.shape[0]} lignes")
# Nettoyage des données
# 1. Supprimer les lignes avec des valeurs manquantes
df = df.dropna()
logger.info(f"Après suppression des NA : {df.shape[0]} lignes")
# 2. Convertir les dates
df['date'] = pd.to_datetime(df[['year', 'month',
'day']].astype(str).agg('-'.join,
axis=1))
# 3. Supprimer les doublons en gardant la première occurrence
df = df.drop_duplicates(subset=[
'date', 'pm2.5', 'TEMP', 'DEWP', 'PRES', 'Iws', 'Is', 'Ir'
],
keep='first')
logger.info(f"Après suppression des doublons : {df.shape[0]} lignes")
# Agréger les données par jour
numeric_columns = ['pm2.5', 'TEMP', 'DEWP', 'PRES', 'Iws', 'Is', 'Ir']
daily_df = df.groupby(df['date'])[numeric_columns].mean().reset_index()
logger.info(f"Données agrégées : {daily_df.shape[0]} lignes")
# Calculer les corrélations
correlations = df[numeric_columns].corr()['pm2.5'].sort_values(
ascending=False)
logger.info(f"Corrélations calculées : \n{correlations}")
# Préparer la réponse
response_data = {
'dates':
daily_df['date'].dt.strftime('%Y-%m-%d').tolist(),
'pm25': [float(x) for x in daily_df['pm2.5']],
'correlated_vars': [{
'name': var,
'label': {
'TEMP': 'Température',
'DEWP': 'Point de Rosée',
'PRES': 'Pression',
'Iws': 'Vitesse du Vent',
'Is': "Heures d'Ensoleillement",
'Ir': 'Précipitations'
}.get(var, var)
} for var in correlations[1:6].index]
}
# Ajouter les données normalisées
for var in correlations[1:6].index:
values = daily_df[var]
normalized = (values - values.mean()) / values.std()
response_data[var] = [float(x) for x in normalized]
# Ajouter les valeurs de corrélation
response_data['correlation_values'] = [
float(x) for x in correlations[1:6].values
]
logger.info("Données préparées avec succès")
return jsonify(response_data)
except Exception as e:
logger.error(f"Erreur lors de la récupération des données : {str(e)}")
logger.error(f"Traceback : {traceback.format_exc()}")
return jsonify({'error': str(e)}), 500
[docs]
@main_bp.route('/testihm/services/api_ihm/uml/schemas/<path:filename>')
def serve_uml_file(filename):
"""Route pour servir les fichiers UML en environnement de production."""
try:
logger.info(f"=== Début serve_uml_file pour {filename} ===")
env_type = get_environment_type()
paths, local_paths = get_base_paths()
logger.info(f"Environnement détecté : {env_type}")
logger.debug(f"Chemins disponibles : {json.dumps(paths, indent=2)}")
if env_type in ['production', 'test']:
try:
# Utiliser le chemin physique dans Docker
if os.path.exists('/.dockerenv'):
file_path = os.path.join(paths['physical_uml_path'],
filename)
else:
# Sur la VM
file_path = os.path.join(
'/testihm/services/api_ihm/uml/schemas',
filename)
logger.info(f"Tentative d'accès au fichier: {file_path}")
logger.debug(f"Le fichier existe: {os.path.exists(file_path)}")
if os.path.exists(file_path) and os.path.isfile(file_path):
logger.info(f"✓ Fichier trouvé et envoyé: {file_path}")
return send_file(file_path)
else:
logger.error(f"✗ Fichier non trouvé: {file_path}")
return jsonify({"error": "Fichier non trouvé"}), 404
except Exception as e:
logger.error(f"Erreur lors de l'accès au fichier: {str(e)}")
logger.debug(f"Traceback: {traceback.format_exc()}")
return jsonify({"error": str(e)}), 500
else:
logger.warning("Accès non autorisé en environnement local")
return jsonify({"error":
"Non autorisé en environnement local"}), 403
except Exception as e:
logger.error(f"Erreur générale dans serve_uml_file: {str(e)}")
logger.error(f"Traceback complet: {traceback.format_exc()}")
return jsonify({"error": str(e)}), 500
# @main_bp.route('/testihm/uml/schemas/<path:filename>')
# def serve_gestion_projet_file(filename):
# """
# Route pour servir les fichiers gestion de projet en environnement de test.
# """
# try:
# logger.info(f"Demande de fichier Gestion de Projet (test): {filename}")
# if is_test_environment():
# base_path = '/app/uml/schemas'
# file_path = os.path.join(base_path, filename)
# logger.info(f"Chemin complet du fichier: {file_path}")
# if os.path.exists(file_path) and os.path.isfile(file_path):
# logger.info(f"Fichier trouvé: {file_path}")
# return send_file(file_path)
# else:
# logger.error(f"Fichier non trouvé: {file_path}")
# return jsonify({"error": "Fichier non trouvé"}), 404
# else:
# return jsonify({"error":
# "Non autorisé en environnement local"}), 403
# except Exception as e:
# logger.error(
# f"Erreur lors de l'accès au fichier UML {filename}: {str(e)}")
# return jsonify({"error": str(e)}), 500
[docs]
@main_bp.route('/gestion-projet', methods=['GET'])
def gestion_projet():
"""
Affiche la page de gestion de projet avec le diagramme de Gantt et la matrice RACI.
Gère l'affichage des diagrammes depuis les chemins appropriés selon l'environnement.
Version: 1.1.0 (2024-01-05)
"""
try:
logger.info("=== Début du traitement de la route /gestion-projet ===")
env_type = get_environment_type()
paths, local_paths = get_base_paths() # Récupération correcte du tuple
logger.info(f"Environnement détecté : {env_type}")
logger.debug(
f"Chemins disponibles - paths: {json.dumps(paths, indent=2)}")
logger.debug(
f"Chemins disponibles - local_paths: {json.dumps(local_paths, indent=2)}"
)
try:
# Construire les chemins des images selon l'environnement
if env_type in ['production', 'test']:
if os.path.exists('/.dockerenv'):
# Dans Docker
base_url = '/testihm/services/api_ihm/uml/schemas'
base_path = paths['physical_uml_path']
logger.info(
f"Utilisation des chemins Docker - base_path: {base_path}"
)
else:
# Sur la VM
base_url = '/testihm/services/api_ihm/uml/schemas'
base_path = '/testihm/services/api_ihm/uml/schemas'
logger.info(
f"Utilisation des chemins VM - base_path: {base_path}")
else:
# En local
base_url = '/docs/uml/schemas'
base_path = os.path.join(project_root, 'docs', 'uml',
'schemas')
logger.info(
f"Utilisation des chemins locaux - base_path: {base_path}")
# Vérifier l'existence des fichiers
gantt_path = os.path.join(base_path, 'gestion_projet',
'gantt_diagram.png')
raci_path = os.path.join(base_path, 'gestion_projet',
'raci_matrix.png')
logger.info("Vérification des fichiers :")
logger.info(
f"- Gantt : {gantt_path} (existe: {os.path.exists(gantt_path)})"
)
logger.info(
f"- RACI : {raci_path} (existe: {os.path.exists(raci_path)})")
# Construire les URLs
image_paths = {
'gantt': [f'{base_url}/gestion_projet/gantt_diagram.png'],
'raci': [f'{base_url}/gestion_projet/raci_matrix.png']
}
logger.info(
f"URLs des images configurées : {json.dumps(image_paths, indent=2)}"
)
logger.info(
"=== Tentative de rendu du template gestion_projet.html ===")
return render_template('gestion_projet.html',
image_paths=image_paths)
except Exception as e:
logger.error(
f"Erreur lors de la configuration des chemins: {str(e)}")
logger.debug(f"Traceback: {traceback.format_exc()}")
raise
except Exception as e:
logger.error(
f"Erreur lors du chargement des diagrammes de gestion de projet: {str(e)}"
)
logger.error(f"Traceback complet: {traceback.format_exc()}")
return render_template(
'error.html',
error_message=
"Erreur lors du chargement des diagrammes de gestion de projet",
error_details=str(e),
suggestions=[
"Vérifiez que les images existent aux emplacements suivants :",
f"- Environnement détecté : {env_type}",
f"- URL de base : {base_url if 'base_url' in locals() else 'non définie'}",
f"- Chemin physique : {base_path if 'base_path' in locals() else 'non défini'}/gestion_projet/",
"- Fichiers attendus : gantt_diagram.png et raci_matrix.png"
])
[docs]
@main_bp.route('/gestion-projet/gantt')
def gantt():
"""
Affiche le diagramme de Gantt interactif.
Version: 1.0.0 (2024-01-05)
"""
try:
return render_template('gantt.html')
except Exception as e:
logger.error(f"Erreur lors de l'affichage du Gantt: {str(e)}")
return render_template(
'error.html',
error_message="Erreur lors de l'affichage du Gantt",
error_details=str(e))
[docs]
@main_bp.route('/gestion-projet/pert')
def pert():
"""
Affiche le diagramme PERT interactif.
Version: 1.0.0 (2024-01-05)
"""
try:
return render_template('pert.html')
except Exception as e:
logger.error(f"Erreur lors de l'affichage du PERT: {str(e)}")
return render_template(
'error.html',
error_message="Erreur lors de l'affichage du PERT",
error_details=str(e))
[docs]
@main_bp.route('/gestion-projet/raci')
def raci():
"""
Affiche la matrice RACI interactive.
Version: 1.0.0 (2024-01-05)
"""
try:
return render_template('raci.html')
except Exception as e:
logger.error(f"Erreur lors de l'affichage du RACI: {str(e)}")
return render_template(
'error.html',
error_message="Erreur lors de l'affichage du RACI",
error_details=str(e))
[docs]
@main_bp.route('/delete-prediction/<int:prediction_id>', methods=['POST'])
@login_required
def delete_prediction(prediction_id):
"""
Supprime une prédiction de l'historique.
Seuls les admins peuvent supprimer n'importe quelle prédiction.
Les utilisateurs ne peuvent supprimer que leurs propres prédictions.
"""
try:
prediction = Simulation.query.get_or_404(prediction_id)
# Vérifier les permissions
if not current_user.is_admin and prediction.user_id != current_user.id:
flash(
'Vous n\'avez pas la permission de supprimer cette prédiction.',
'error')
return jsonify({
'success': False,
'message': 'Permission refusée'
}), 403
# Supprimer la prédiction
db.session.delete(prediction)
db.session.commit()
flash('Prédiction supprimée avec succès.', 'success')
return jsonify({'success': True})
except Exception as e:
db.session.rollback()
logger.error(
f"Erreur lors de la suppression de la prédiction {prediction_id}: {str(e)}"
)
return jsonify({'success': False, 'message': str(e)}), 500