"""Services métier communs aux compteurs et à leurs relevés.""" from datetime import datetime, timezone from math import isfinite from ..models.college import Building, HousingUnit, Room, Zone from ..models.equipment import Equipment from ..models.planning import Meter, MeterReading, MeterReadingCorrection from ...extensions import db class MeterDomainError(ValueError): """Erreur de validation métier d'un compteur ou d'un relevé.""" SCOPE_MODELS = { "equipment": (Equipment, "equipment_id"), "building": (Building, "building_id"), "zone": (Zone, "zone_id"), "room": (Room, "room_id"), "housing_unit": (HousingUnit, "housing_unit_id"), } def _scope_values(**scope_ids): values = {name: value for name, value in scope_ids.items() if value is not None} if len(values) != 1: raise MeterDomainError("Un compteur doit avoir exactement un rattachement patrimoine.") return values def _validate_parent(meter, parent): if parent is None: return if parent.id == meter.id: raise MeterDomainError("Un compteur ne peut pas être son propre parent.") cursor = parent seen = set() while cursor is not None: if cursor.id in seen: raise MeterDomainError("La hiérarchie des compteurs contient déjà un cycle.") seen.add(cursor.id) if cursor.id == meter.id: raise MeterDomainError("Cette relation créerait un cycle de compteurs.") cursor = cursor.parent def validate_meter_scope(meter): """Valide le rattachement principal et la hiérarchie d'un compteur.""" with db.session.no_autoflush: _scope_values( equipment_id=meter.equipment_id, building_id=meter.building_id, zone_id=meter.zone_id, room_id=meter.room_id, housing_unit_id=meter.housing_unit_id, ) _validate_parent(meter, meter.parent) return meter def create_meter(*, name, meter_type="hours", unit="h", usage=None, initial_value=0, current_value=None, parent=None, remainder_label=None, status="active", replaces_meter_id=None, warning_threshold=None, critical_threshold=None, maintenance_interval=None, last_maintenance_value=None, reading_interval_days=None, commit=True, **scope_ids): """Crée un compteur avec les mêmes règles pour toutes les routes.""" name = (name or "").strip() if not name: raise MeterDomainError("Le nom du compteur est obligatoire.") scope = _scope_values(**scope_ids) model = Meter(name=name, meter_type=meter_type or "hours", unit=unit or "h", usage=usage, initial_value=initial_value or 0, current_value=initial_value if current_value is None else current_value, remainder_label=remainder_label, status=status or "active", replaces_meter_id=replaces_meter_id, warning_threshold=warning_threshold, critical_threshold=critical_threshold, maintenance_interval=maintenance_interval, last_maintenance_value=last_maintenance_value, reading_interval_days=reading_interval_days, **scope) if parent is not None: model.parent = parent validate_meter_scope(model) db.session.add(model) if commit: db.session.commit() return model def _latest_reading(meter): return (MeterReading.query.filter_by(meter_id=meter.id) .order_by(MeterReading.reading_date.desc(), MeterReading.id.desc()) .first()) def record_meter_reading(*, meter, value, user_id, reading_date=None, notes=None, is_reset=False, reset_reason=None, photo_filename=None, photo_path=None, commit=True): """Enregistre une vraie mesure et synchronise l'index courant.""" if meter.status == "replaced" or not meter.is_active: raise MeterDomainError( "Ce compteur est inactif ou remplacé. Les nouveaux relevés doivent être saisis sur le compteur actif." ) try: value = float(value) except (TypeError, ValueError): raise MeterDomainError("La valeur du relevé est invalide.") if not isfinite(value): raise MeterDomainError("La valeur du relevé est invalide.") reset_reason = (reset_reason or "").strip() if is_reset and not reset_reason: raise MeterDomainError("Le motif de remise à zéro est obligatoire.") if value < (meter.current_value or 0) and not is_reset: raise MeterDomainError("Un relevé inférieur exige une remise à zéro explicite et justifiée.") reading_date = reading_date or datetime.now(timezone.utc) reading = MeterReading( meter=meter, value=value, reading_date=reading_date, read_by_id=user_id, notes=notes, is_reset=bool(is_reset), reset_reason=reset_reason or None, photo_filename=photo_filename, photo_path=photo_path, ) meter.current_value = value meter.last_reading_date = reading_date.date() if is_reset: meter.initial_value = value meter.last_maintenance_value = value db.session.add(reading) if commit: db.session.commit() return reading def correct_meter_reading(*, reading, new_value, user_id, reason, commit=True): """Corrige un relevé en conservant une trace relationnelle complète.""" reason = (reason or "").strip() if not reason: raise MeterDomainError("La justification de correction est obligatoire.") try: new_value = float(new_value) except (TypeError, ValueError): raise MeterDomainError("La valeur corrigée est invalide.") if not isfinite(new_value): raise MeterDomainError("La valeur corrigée est invalide.") old_value = reading.value if old_value == new_value: raise MeterDomainError("La correction doit modifier la valeur.") db.session.add(MeterReadingCorrection( reading=reading, old_value=old_value, new_value=new_value, corrected_by_id=user_id, reason=reason, )) reading.value = new_value meter = reading.meter latest = _latest_reading(meter) if latest is not None: meter.current_value = latest.value meter.last_reading_date = latest.reading_date.date() if commit: db.session.commit() return reading def replace_meter(*, old_meter, name, user_id=None, commit=True, **kwargs): """Crée un nouveau compteur et clôt explicitement l'ancien.""" if old_meter.status == "replaced": raise MeterDomainError("Ce compteur est déjà remplacé.") scope = { key: getattr(old_meter, key) for key in ("equipment_id", "building_id", "zone_id", "room_id", "housing_unit_id") } new_meter = create_meter( name=name, meter_type=kwargs.pop("meter_type", old_meter.meter_type), unit=kwargs.pop("unit", old_meter.unit), usage=kwargs.pop("usage", old_meter.usage), initial_value=kwargs.pop("initial_value", 0), current_value=kwargs.pop("current_value", 0), replaces_meter_id=old_meter.id, commit=False, **scope, ) old_meter.status = "replaced" old_meter.is_active = False # C2 : les échéances ouvertes sont clôturées dans la même transaction que # le remplacement ; l'historique reste consultable. from .meter_reading_planning import cancel_open_occurrences_for_meter cancel_open_occurrences_for_meter(old_meter) db.session.add(new_meter) if commit: db.session.commit() return new_meter