""" Meters Routes - GMAO Collège Gestion des compteurs """ from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify, current_app, send_file, abort from flask_login import login_required, current_user from datetime import datetime from pathlib import Path from uuid import uuid4 from werkzeug.utils import secure_filename from sqlalchemy.orm import joinedload from app_new.extensions import db from app_new.core.models.planning import Meter, MeterReading from app_new.core.models.equipment import Equipment from app_new.core.models.college import Building, HousingUnit, Room, Zone from app_new.core.services.meter_service import ( MeterDomainError, correct_meter_reading, create_meter, record_meter_reading, replace_meter, ) meters_bp = Blueprint('meters', __name__, template_folder='templates') @meters_bp.route('/') @login_required def index(): """Liste de tous les compteurs.""" meters = Meter.query.order_by(Meter.name).all() # Statistiques total_count = len(meters) warning_count = sum(1 for m in meters if m.warning_level == 'warning') critical_count = sum(1 for m in meters if m.warning_level == 'critical') maintenance_count = sum(1 for m in meters if m.warning_level == 'maintenance') return render_template('meters/index.html', meters=meters, total_count=total_count, warning_count=warning_count, critical_count=critical_count, maintenance_count=maintenance_count) @meters_bp.route('/new', methods=['GET', 'POST']) @login_required def new(): """Créer un compteur rattaché à un élément patrimoine.""" if request.method == 'POST': try: scope_type = request.form.get('scope_type') scope_id = request.form.get('scope_id', type=int) scope_key = { 'equipment': 'equipment_id', 'building': 'building_id', 'zone': 'zone_id', 'room': 'room_id', 'housing_unit': 'housing_unit_id', }.get(scope_type) if not scope_key or not scope_id: raise MeterDomainError('Un rattachement patrimoine est obligatoire.') meter = create_meter( name=(request.form.get('name') or '').strip(), meter_type=request.form.get('meter_type') or 'hours', unit=(request.form.get('unit') or '').strip() or 'h', usage=(request.form.get('usage') or '').strip() or None, initial_value=request.form.get('initial_value', type=float, default=0), parent=Meter.query.get(request.form.get('parent_id', type=int)) if request.form.get('parent_id', type=int) else None, remainder_label=(request.form.get('remainder_label') or '').strip() or None, **{scope_key: scope_id}, ) flash(f'Compteur « {meter.name} » créé.', 'success') return redirect(url_for('meters.detail', meter_id=meter.id)) except MeterDomainError as exc: db.session.rollback() flash(str(exc), 'danger') return render_template( 'meters/new.html', equipments=Equipment.query.filter_by(is_group=False).order_by(Equipment.name).all(), buildings=Building.query.order_by(Building.name).all(), zones=Zone.query.order_by(Zone.name).all(), rooms=Room.query.order_by(Room.name).all(), housing_units=HousingUnit.query.order_by(HousingUnit.name).all(), parents=Meter.query.filter(Meter.status != 'replaced').order_by(Meter.name).all(), ) @meters_bp.route('/') @login_required def detail(meter_id): """Détails d'un compteur avec historique des relevés.""" meter = Meter.query.options( joinedload(Meter.equipment), joinedload(Meter.building), joinedload(Meter.zone), joinedload(Meter.room), joinedload(Meter.housing_unit), joinedload(Meter.replaced_meter), joinedload(Meter.replacement), ).get_or_404(meter_id) readings = MeterReading.query.options(joinedload(MeterReading.read_by)).filter_by(meter_id=meter_id)\ .order_by(MeterReading.reading_date.desc()).limit(50).all() return render_template('meters/detail.html', meter=meter, readings=readings) @meters_bp.route('//reading', methods=['POST']) @login_required def add_reading(meter_id): """Ajouter un relevé à un compteur.""" meter = Meter.query.get_or_404(meter_id) is_reset = request.form.get('is_reset') == '1' reset_reason = (request.form.get('reset_reason') or '').strip() photo_filename = photo_path = None photo = request.files.get('photo') if photo and photo.filename: allowed = {'jpg', 'jpeg', 'png', 'gif'} if '.' not in photo.filename or photo.filename.rsplit('.', 1)[1].lower() not in allowed: flash('La photo doit être une image JPG, PNG ou GIF.', 'danger') return redirect(url_for('meters.detail', meter_id=meter_id)) photo_filename = secure_filename(photo.filename) stored_name = f'{uuid4().hex}_{photo_filename}' upload_dir = Path(current_app.config.get('UPLOAD_FOLDER', 'uploads')) / 'meter-readings' upload_dir.mkdir(parents=True, exist_ok=True) photo_path = str(upload_dir / stored_name) photo.save(photo_path) photo_filename = stored_name try: reading = record_meter_reading( meter=meter, value=request.form.get('value'), user_id=current_user.id, notes=request.form.get('notes', ''), is_reset=is_reset, reset_reason=reset_reason, photo_filename=photo_filename, photo_path=photo_path, ) except MeterDomainError as exc: db.session.rollback() if photo_path: Path(photo_path).unlink(missing_ok=True) flash(str(exc), 'danger') return redirect(url_for('meters.detail', meter_id=meter_id)) flash(f'Relevé enregistré: {reading.value} {meter.unit}', 'success') return redirect(url_for('meters.detail', meter_id=meter_id)) @meters_bp.route('//edit', methods=['GET', 'POST']) @login_required def edit(meter_id): """Modifier un compteur.""" from datetime import datetime, timedelta, date meter = Meter.query.get_or_404(meter_id) equipments = Equipment.query.filter_by(is_group=False).order_by(Equipment.name).all() if request.method == 'POST': meter.name = request.form.get('name') meter.meter_type = request.form.get('meter_type', 'hours') meter.unit = request.form.get('unit', 'h') meter.warning_threshold = request.form.get('warning_threshold', type=float) meter.critical_threshold = request.form.get('critical_threshold', type=float) meter.maintenance_interval = request.form.get('maintenance_interval', type=float) meter.last_maintenance_value = request.form.get('last_maintenance_value', type=float) meter.reading_interval_days = request.form.get('reading_interval_days', type=int) # L'index courant est piloté exclusivement par les relevés. Les # corrections passent par l'historique afin de rester auditées. # Date du dernier relevé last_reading_date = request.form.get('last_reading_date') if last_reading_date: meter.last_reading_date = datetime.strptime(last_reading_date, '%Y-%m-%d').date() else: meter.last_reading_date = None # Équipement scope_type = request.form.get('scope_type') scope_id = request.form.get('scope_id', type=int) if scope_type and scope_id: scope_keys = ('equipment_id', 'building_id', 'zone_id', 'room_id', 'housing_unit_id') scope_key = {'equipment': 'equipment_id', 'building': 'building_id', 'zone': 'zone_id', 'room': 'room_id', 'housing_unit': 'housing_unit_id'}.get(scope_type) if scope_key: for key in scope_keys: setattr(meter, key, scope_id if key == scope_key else None) meter.parent_id = request.form.get('parent_id', type=int) meter.usage = (request.form.get('usage') or '').strip() or None meter.remainder_label = (request.form.get('remainder_label') or '').strip() or None meter.status = request.form.get('status') or meter.status try: from app_new.core.services.meter_service import validate_meter_scope validate_meter_scope(meter) except MeterDomainError as exc: db.session.rollback() flash(str(exc), 'danger') return redirect(url_for('meters.edit', meter_id=meter_id)) db.session.commit() flash('Compteur modifié avec succès', 'success') return redirect(url_for('meters.detail', meter_id=meter_id)) return render_template('meters/edit.html', meter=meter, equipments=equipments, buildings=Building.query.order_by(Building.name).all(), zones=Zone.query.order_by(Zone.name).all(), rooms=Room.query.order_by(Room.name).all(), housing_units=HousingUnit.query.order_by(HousingUnit.name).all(), meters=Meter.query.filter(Meter.id != meter_id).order_by(Meter.name).all(), timedelta=timedelta, today=date.today()) @meters_bp.route('//readings//correct', methods=['POST']) @login_required def correct_reading(meter_id, reading_id): reading = MeterReading.query.filter_by(id=reading_id, meter_id=meter_id).first_or_404() try: correct_meter_reading( reading=reading, new_value=request.form.get('value'), user_id=current_user.id, reason=request.form.get('reason'), ) flash('Relevé corrigé et correction auditée.', 'success') except MeterDomainError as exc: db.session.rollback() flash(str(exc), 'danger') return redirect(url_for('meters.detail', meter_id=meter_id)) @meters_bp.route('//readings//photo') @login_required def reading_photo(meter_id, reading_id): reading = MeterReading.query.filter_by(id=reading_id, meter_id=meter_id).first_or_404() upload_root = Path(current_app.config.get('UPLOAD_FOLDER', 'uploads')).resolve() photo_path = Path(reading.photo_path or '').resolve() if not reading.photo_path or upload_root not in photo_path.parents or not photo_path.is_file(): abort(404) return send_file(photo_path, download_name=reading.photo_filename or photo_path.name) @meters_bp.route('//replace', methods=['POST']) @login_required def replace(meter_id): old_meter = Meter.query.get_or_404(meter_id) try: new_meter = replace_meter( old_meter=old_meter, name=(request.form.get('name') or '').strip(), user_id=current_user.id, meter_type=request.form.get('meter_type'), unit=request.form.get('unit'), usage=request.form.get('usage'), ) flash(f'Compteur remplacé par « {new_meter.name} ».', 'success') except MeterDomainError as exc: db.session.rollback() flash(str(exc), 'danger') return redirect(url_for('meters.detail', meter_id=meter_id))