179 lines
7 KiB
Python
179 lines
7 KiB
Python
|
"""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."""
|
||
|
|
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
|
||
|
|
db.session.add(new_meter)
|
||
|
|
if commit:
|
||
|
|
db.session.commit()
|
||
|
|
return new_meter
|