feat(planning): add scheduled meter reading work
This commit is contained in:
root
parent
3c1a6ea6c1
commit
601775891f
16 changed files with 1021 additions and 4 deletions
|
|
@ -12,7 +12,9 @@ from .maintenance import (
|
|||
from .company import Company, Part, PartStockMovement, Alert, Service
|
||||
from .planning import (
|
||||
WorkSchedule, WorkScheduleTemplate, AnnualTimeConfig, TimeEntry, CollegeClosure, ClosureSchedule, ClosureWorkDay, PersonalLeave, Training, TrainingParticipant,
|
||||
Meter, MeterReading, MeterReadingCorrection, Consumable, ConsumableUsage, EquipmentConsumable,
|
||||
Meter, MeterReading, MeterReadingCorrection, MeterReadingSchedule, MeterReadingOccurrence,
|
||||
MeterReadingRound, MeterReadingRoundMember, MeterReadingRoundOccurrence,
|
||||
Consumable, ConsumableUsage, EquipmentConsumable,
|
||||
PreventiveTask, PreventiveTaskConsumable, ScheduledTask, TaskExecutionSegment,
|
||||
TechnicianAvailability, AdminTask, ZoneAccessRule
|
||||
)
|
||||
|
|
@ -36,7 +38,9 @@ __all__ = [
|
|||
'Lot', 'LotTask', 'LotService', 'WorkRequest', 'WorkRequestQuote', 'WorkRequestEvent',
|
||||
'Company', 'Part', 'PartStockMovement', 'InterventionPart', 'Alert', 'Service',
|
||||
'WorkSchedule', 'WorkScheduleTemplate', 'AnnualTimeConfig', 'TimeEntry', 'CollegeClosure', 'ClosureSchedule', 'ClosureWorkDay', 'PersonalLeave', 'Training', 'TrainingParticipant',
|
||||
'Meter', 'MeterReading', 'MeterReadingCorrection', 'Consumable', 'ConsumableUsage', 'EquipmentConsumable',
|
||||
'Meter', 'MeterReading', 'MeterReadingCorrection', 'MeterReadingSchedule', 'MeterReadingOccurrence',
|
||||
'MeterReadingRound', 'MeterReadingRoundMember', 'MeterReadingRoundOccurrence',
|
||||
'Consumable', 'ConsumableUsage', 'EquipmentConsumable',
|
||||
'PreventiveTask', 'PreventiveTaskConsumable', 'ScheduledTask', 'TaskExecutionSegment',
|
||||
'TechnicianAvailability', 'AdminTask', 'ZoneAccessRule',
|
||||
'AppSettings', 'AuditLog', 'TemplateAuditMark',
|
||||
|
|
|
|||
|
|
@ -371,6 +371,125 @@ class MeterReadingCorrection(db.Model):
|
|||
corrected_by = db.relationship("User")
|
||||
|
||||
|
||||
class MeterReadingSchedule(db.Model):
|
||||
"""Règle de planification des relevés d'un compteur."""
|
||||
__tablename__ = "meter_reading_schedules"
|
||||
__table_args__ = (
|
||||
db.UniqueConstraint("meter_id", name="uq_meter_reading_schedule_meter"),
|
||||
db.CheckConstraint(
|
||||
"frequency in ('WEEKLY', 'MONTHLY', 'SEMI_ANNUAL', 'ANNUAL', 'FIXED_ANNUAL_DATE')",
|
||||
name="ck_meter_reading_schedule_frequency",
|
||||
),
|
||||
)
|
||||
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
meter_id = db.Column(db.Integer, db.ForeignKey("meters.id", ondelete="CASCADE"), nullable=False, index=True)
|
||||
is_active = db.Column(db.Boolean, nullable=False, default=True, server_default="1")
|
||||
frequency = db.Column(db.String(30), nullable=False)
|
||||
reference_date = db.Column(db.Date, nullable=False)
|
||||
target_time = db.Column(db.Time, nullable=True)
|
||||
duration_minutes = db.Column(db.Integer, nullable=False, default=5)
|
||||
assigned_to_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
|
||||
closed_day_policy = db.Column(db.String(30), nullable=False, default="PREVIOUS_WORKING_DAY")
|
||||
calendar_scope = db.Column(db.String(20), nullable=False, default="school")
|
||||
fixed_month = db.Column(db.Integer, nullable=True)
|
||||
fixed_day = db.Column(db.Integer, nullable=True)
|
||||
created_at = db.Column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
updated_at = db.Column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
|
||||
|
||||
meter = db.relationship("Meter", backref=db.backref("reading_schedule", uselist=False))
|
||||
assigned_to = db.relationship("User")
|
||||
occurrences = db.relationship(
|
||||
"MeterReadingOccurrence", back_populates="schedule", cascade="all, delete-orphan",
|
||||
order_by="MeterReadingOccurrence.target_date",
|
||||
)
|
||||
|
||||
|
||||
class MeterReadingRound(db.Model):
|
||||
"""Définition d'une tournée de relevés."""
|
||||
__tablename__ = "meter_reading_rounds"
|
||||
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
name = db.Column(db.String(255), nullable=False)
|
||||
is_active = db.Column(db.Boolean, nullable=False, default=True, server_default="1")
|
||||
default_assigned_to_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
|
||||
estimated_duration_minutes = db.Column(db.Integer, nullable=True)
|
||||
created_at = db.Column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
updated_at = db.Column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
|
||||
|
||||
default_assigned_to = db.relationship("User")
|
||||
members = db.relationship(
|
||||
"MeterReadingRoundMember", back_populates="round", cascade="all, delete-orphan",
|
||||
order_by="MeterReadingRoundMember.position",
|
||||
)
|
||||
occurrences = db.relationship(
|
||||
"MeterReadingRoundOccurrence", back_populates="round", cascade="all, delete-orphan",
|
||||
)
|
||||
|
||||
|
||||
class MeterReadingRoundMember(db.Model):
|
||||
"""Compteur/règle et ordre d'une tournée."""
|
||||
__tablename__ = "meter_reading_round_members"
|
||||
__table_args__ = (
|
||||
db.UniqueConstraint("round_id", "schedule_id", name="uq_meter_round_schedule"),
|
||||
db.UniqueConstraint("round_id", "position", name="uq_meter_round_position"),
|
||||
)
|
||||
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
round_id = db.Column(db.Integer, db.ForeignKey("meter_reading_rounds.id", ondelete="CASCADE"), nullable=False, index=True)
|
||||
schedule_id = db.Column(db.Integer, db.ForeignKey("meter_reading_schedules.id", ondelete="CASCADE"), nullable=False, index=True)
|
||||
position = db.Column(db.Integer, nullable=False)
|
||||
|
||||
round = db.relationship("MeterReadingRound", back_populates="members")
|
||||
schedule = db.relationship("MeterReadingSchedule")
|
||||
|
||||
|
||||
class MeterReadingRoundOccurrence(db.Model):
|
||||
"""Occurrence opérationnelle d'une tournée à une date donnée."""
|
||||
__tablename__ = "meter_reading_round_occurrences"
|
||||
__table_args__ = (
|
||||
db.UniqueConstraint("round_id", "operational_date", name="uq_meter_round_occurrence_date"),
|
||||
)
|
||||
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
round_id = db.Column(db.Integer, db.ForeignKey("meter_reading_rounds.id", ondelete="CASCADE"), nullable=False, index=True)
|
||||
operational_date = db.Column(db.Date, nullable=False, index=True)
|
||||
status = db.Column(db.String(30), nullable=False, default="À FAIRE")
|
||||
created_at = db.Column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
|
||||
round = db.relationship("MeterReadingRound", back_populates="occurrences")
|
||||
occurrences = db.relationship("MeterReadingOccurrence", back_populates="round_occurrence")
|
||||
|
||||
|
||||
class MeterReadingOccurrence(db.Model):
|
||||
"""Une échéance unique, persistante et traitable."""
|
||||
__tablename__ = "meter_reading_occurrences"
|
||||
__table_args__ = (
|
||||
db.UniqueConstraint("schedule_id", "target_date", name="uq_meter_reading_occurrence_target"),
|
||||
)
|
||||
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
schedule_id = db.Column(db.Integer, db.ForeignKey("meter_reading_schedules.id", ondelete="CASCADE"), nullable=False, index=True)
|
||||
target_date = db.Column(db.Date, nullable=False, index=True)
|
||||
operational_date = db.Column(db.Date, nullable=False, index=True)
|
||||
status = db.Column(db.String(30), nullable=False, default="À FAIRE")
|
||||
assigned_to_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
|
||||
meter_reading_id = db.Column(db.Integer, db.ForeignKey("meter_readings.id", ondelete="SET NULL"), nullable=True, index=True)
|
||||
round_occurrence_id = db.Column(db.Integer, db.ForeignKey("meter_reading_round_occurrences.id", ondelete="SET NULL"), nullable=True, index=True)
|
||||
processed_by_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
|
||||
processed_at = db.Column(db.DateTime, nullable=True)
|
||||
no_reading_reason = db.Column(db.String(40), nullable=True)
|
||||
comment = db.Column(db.Text, nullable=True)
|
||||
created_at = db.Column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
updated_at = db.Column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
|
||||
|
||||
schedule = db.relationship("MeterReadingSchedule", back_populates="occurrences")
|
||||
assigned_to = db.relationship("User", foreign_keys=[assigned_to_id])
|
||||
processed_by = db.relationship("User", foreign_keys=[processed_by_id])
|
||||
meter_reading = db.relationship("MeterReading")
|
||||
round_occurrence = db.relationship("MeterReadingRoundOccurrence", back_populates="occurrences")
|
||||
|
||||
|
||||
class Consumable(db.Model):
|
||||
"""Consommable pour maintenance."""
|
||||
__tablename__ = "consumables"
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ TYPE_LABELS = {
|
|||
"admin": "Administratif",
|
||||
"external_company": "Entreprise extérieure",
|
||||
"emergency": "Urgence",
|
||||
"meter_reading": "Relevé compteur",
|
||||
}
|
||||
CONSTRAINT_LABELS = {
|
||||
"fixed": "Horaire fixe",
|
||||
|
|
@ -39,6 +40,7 @@ class PlanningCandidate:
|
|||
priority: int = 3
|
||||
constraint: str = "flexible"
|
||||
target_date: Optional[date] = None
|
||||
due_date: Optional[date] = None
|
||||
earliest_start: Optional[time] = None
|
||||
latest_end: Optional[time] = None
|
||||
duration_minutes: int = 30
|
||||
|
|
@ -272,12 +274,14 @@ class DayPlanner:
|
|||
|
||||
@classmethod
|
||||
def from_database(cls, day: date, user_id=None):
|
||||
"""Charge les sources accessibles sans créer de nouvelle table."""
|
||||
"""Charge les sources accessibles, dont les échéances de relevé C2."""
|
||||
from ..models.planning import ScheduledTask, AdminTask
|
||||
from ..models.maintenance import Intervention
|
||||
from ...contracts.models import ContractVisit
|
||||
from .meter_reading_planning import ensure_occurrences_for_day, list_open_occurrences
|
||||
|
||||
candidates = []
|
||||
ensure_occurrences_for_day(day, user_id=user_id)
|
||||
tasks = ScheduledTask.query.filter(
|
||||
ScheduledTask.scheduled_date == day,
|
||||
ScheduledTask.status.in_(("planned", "in_progress", "suspended", "postponed")),
|
||||
|
|
@ -366,4 +370,27 @@ class DayPlanner:
|
|||
fixed_start=fixed_start, fixed_end=fixed_end,
|
||||
explanation=visit.display_mode,
|
||||
))
|
||||
for occurrence in list_open_occurrences(day, user_id=user_id):
|
||||
meter = occurrence.schedule.meter
|
||||
scope = meter.equipment or meter.building or meter.zone or meter.room or meter.housing_unit
|
||||
room = meter.room or (meter.equipment.effective_room if meter.equipment else None)
|
||||
scope_name = getattr(scope, "name", None)
|
||||
title = f"Relevé compteur — {meter.name}"
|
||||
if scope_name:
|
||||
title += f" — {scope_name}"
|
||||
overdue = occurrence.operational_date < day
|
||||
candidates.append(PlanningCandidate(
|
||||
source_type="meter_reading_occurrence", source_id=occurrence.id,
|
||||
title=title, description=meter.usage or "",
|
||||
task_type="meter_reading", priority=2 if overdue else 3,
|
||||
constraint="deadline", target_date=day, due_date=occurrence.target_date,
|
||||
duration_minutes=occurrence.schedule.duration_minutes,
|
||||
room_id=room.id if room else None, room_name=room.name if room else None,
|
||||
zone_name=room.zone.name if room and room.zone else None,
|
||||
building_name=room.building.name if room and room.building else None,
|
||||
assigned_to=occurrence.assigned_to_id,
|
||||
status="en retard" if overdue else "à faire",
|
||||
explanation=(f"Échéance cible : {occurrence.target_date.strftime('%d/%m/%Y')}" +
|
||||
(" · date opérationnelle dépassée" if overdue else "")),
|
||||
))
|
||||
return cls.propose(day, candidates, user_id=user_id)
|
||||
|
|
|
|||
272
app_new/core/services/meter_reading_planning.py
Normal file
272
app_new/core/services/meter_reading_planning.py
Normal file
|
|
@ -0,0 +1,272 @@
|
|||
"""Planification opérationnelle des relevés de compteurs (Checkpoint C2)."""
|
||||
|
||||
from calendar import monthrange
|
||||
from datetime import date, datetime, time, timedelta
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import joinedload
|
||||
|
||||
from ...extensions import db
|
||||
from ..models.planning import (
|
||||
MeterReadingOccurrence, MeterReadingRound, MeterReadingRoundMember,
|
||||
MeterReadingRoundOccurrence, MeterReadingSchedule,
|
||||
)
|
||||
from .meter_service import MeterDomainError, record_meter_reading
|
||||
|
||||
|
||||
FREQUENCIES = {"WEEKLY", "MONTHLY", "SEMI_ANNUAL", "ANNUAL", "FIXED_ANNUAL_DATE"}
|
||||
NO_READING_REASONS = {
|
||||
"INACCESSIBLE": "Compteur inaccessible",
|
||||
"OCCUPANT_ABSENT": "Occupant absent",
|
||||
"OCCUPANT_REFUSED": "Occupant refuse de communiquer le relevé",
|
||||
"READING_UNKNOWN": "Relevé non connu",
|
||||
"OTHER": "Autre",
|
||||
}
|
||||
TODO = "À FAIRE"
|
||||
TREATED_READING = "TRAITÉE AVEC RELEVÉ"
|
||||
TREATED_WITHOUT_READING = "TRAITÉE SANS RELEVÉ"
|
||||
CANCELLED = "ANNULÉE"
|
||||
|
||||
|
||||
def _add_months(value, months):
|
||||
index = value.year * 12 + value.month - 1 + months
|
||||
year, month = divmod(index, 12)
|
||||
month += 1
|
||||
return date(year, month, min(value.day, monthrange(year, month)[1]))
|
||||
|
||||
|
||||
def _target_dates(schedule, start_date, end_date):
|
||||
reference = schedule.reference_date
|
||||
if schedule.frequency == "FIXED_ANNUAL_DATE":
|
||||
month = schedule.fixed_month or reference.month
|
||||
day = schedule.fixed_day or reference.day
|
||||
for year in range(start_date.year - 1, end_date.year + 2):
|
||||
candidate = date(year, month, min(day, monthrange(year, month)[1]))
|
||||
if start_date <= candidate <= end_date:
|
||||
yield candidate
|
||||
return
|
||||
|
||||
step = {"WEEKLY": 7, "MONTHLY": 1, "SEMI_ANNUAL": 6, "ANNUAL": 12}[schedule.frequency]
|
||||
cursor = reference
|
||||
if schedule.frequency == "WEEKLY":
|
||||
if cursor < start_date:
|
||||
cursor += timedelta(days=((start_date - cursor).days // 7) * 7)
|
||||
else:
|
||||
while cursor < start_date:
|
||||
cursor = _add_months(cursor, step)
|
||||
while cursor <= end_date:
|
||||
if cursor >= start_date:
|
||||
yield cursor
|
||||
cursor = cursor + timedelta(days=step) if schedule.frequency == "WEEKLY" else _add_months(cursor, step)
|
||||
|
||||
|
||||
def operational_date_for(schedule, target_date):
|
||||
"""Conserve la date cible et anticipe uniquement pour un calendrier scolaire."""
|
||||
if schedule.calendar_scope != "school" or schedule.closed_day_policy != "PREVIOUS_WORKING_DAY":
|
||||
return target_date
|
||||
from .planning_service import PlanningService
|
||||
# Sans horaires explicites pour le responsable, le moteur ne peut pas
|
||||
# déduire qu'un jour est travaillé. On conserve donc la date cible ; une
|
||||
# configuration d'horaires ultérieure permettra l'anticipation précise.
|
||||
from ..models.planning import WorkSchedule
|
||||
schedules = WorkSchedule.query.filter_by(is_active=True)
|
||||
if schedule.assigned_to_id is not None:
|
||||
schedules = schedules.filter_by(user_id=schedule.assigned_to_id)
|
||||
if schedules.first() is None:
|
||||
return target_date
|
||||
if PlanningService.get_working_hours(target_date, user_id=schedule.assigned_to_id):
|
||||
return target_date
|
||||
candidate = target_date - timedelta(days=1)
|
||||
for _ in range(366):
|
||||
if PlanningService.get_working_hours(candidate, user_id=schedule.assigned_to_id):
|
||||
return candidate
|
||||
candidate -= timedelta(days=1)
|
||||
raise MeterDomainError("Aucun jour opérationnel trouvé avant l'échéance.")
|
||||
|
||||
|
||||
def create_or_update_schedule(*, meter, frequency, reference_date, target_time=None,
|
||||
duration_minutes=5, assigned_to_id=None,
|
||||
closed_day_policy="PREVIOUS_WORKING_DAY",
|
||||
calendar_scope=None, fixed_month=None, fixed_day=None,
|
||||
is_active=True, commit=True):
|
||||
if frequency not in FREQUENCIES:
|
||||
raise MeterDomainError("Fréquence de relevé inconnue.")
|
||||
if not reference_date:
|
||||
raise MeterDomainError("Une date de référence est obligatoire.")
|
||||
if int(duration_minutes or 0) <= 0:
|
||||
raise MeterDomainError("La durée estimée doit être positive.")
|
||||
if meter.status == "replaced" or not meter.is_active:
|
||||
raise MeterDomainError("Un compteur remplacé ou inactif ne peut plus recevoir de règle de relevé.")
|
||||
if calendar_scope is None:
|
||||
calendar_scope = "housing" if meter.housing_unit_id else "school"
|
||||
schedule = meter.reading_schedule
|
||||
if schedule is None:
|
||||
schedule = MeterReadingSchedule(meter=meter)
|
||||
db.session.add(schedule)
|
||||
schedule.frequency = frequency
|
||||
schedule.reference_date = reference_date
|
||||
schedule.target_time = target_time
|
||||
schedule.duration_minutes = int(duration_minutes)
|
||||
schedule.assigned_to_id = assigned_to_id
|
||||
schedule.closed_day_policy = closed_day_policy
|
||||
schedule.calendar_scope = calendar_scope
|
||||
schedule.fixed_month = fixed_month
|
||||
schedule.fixed_day = fixed_day
|
||||
schedule.is_active = bool(is_active)
|
||||
if commit:
|
||||
db.session.commit()
|
||||
return schedule
|
||||
|
||||
|
||||
def _round_occurrence_for(schedule, operational_date):
|
||||
member = MeterReadingRoundMember.query.filter_by(schedule_id=schedule.id).first()
|
||||
if not member or not member.round.is_active:
|
||||
return None
|
||||
occurrence = MeterReadingRoundOccurrence.query.filter_by(
|
||||
round_id=member.round_id, operational_date=operational_date,
|
||||
).first()
|
||||
if occurrence is None:
|
||||
occurrence = MeterReadingRoundOccurrence(round_id=member.round_id, operational_date=operational_date)
|
||||
db.session.add(occurrence)
|
||||
db.session.flush()
|
||||
return occurrence
|
||||
|
||||
|
||||
def _refresh_round_status(round_occurrence):
|
||||
if not round_occurrence:
|
||||
return
|
||||
statuses = [item.status for item in round_occurrence.occurrences]
|
||||
if statuses and all(status != TODO for status in statuses):
|
||||
round_occurrence.status = "TERMINÉE"
|
||||
elif any(status != TODO for status in statuses):
|
||||
round_occurrence.status = "EN COURS"
|
||||
else:
|
||||
round_occurrence.status = TODO
|
||||
|
||||
|
||||
def generate_occurrences(*, start_date, end_date, schedule_ids=None, commit=True):
|
||||
"""Génère un horizon borné, avec unicité logique par règle/date."""
|
||||
query = MeterReadingSchedule.query.options(joinedload(MeterReadingSchedule.meter)).filter_by(is_active=True)
|
||||
if schedule_ids is not None:
|
||||
query = query.filter(MeterReadingSchedule.id.in_(schedule_ids))
|
||||
created = []
|
||||
for schedule in query.all():
|
||||
if schedule.meter.status == "replaced" or not schedule.meter.is_active:
|
||||
for old in MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id, status=TODO).all():
|
||||
old.status = CANCELLED
|
||||
old.no_reading_reason = "METER_REPLACED"
|
||||
old.comment = "Occurrence annulée automatiquement : compteur remplacé ou inactif."
|
||||
continue
|
||||
for target_date in _target_dates(schedule, start_date, end_date):
|
||||
occurrence = MeterReadingOccurrence.query.filter_by(
|
||||
schedule_id=schedule.id, target_date=target_date,
|
||||
).first()
|
||||
if occurrence is not None:
|
||||
continue
|
||||
operational_date = operational_date_for(schedule, target_date)
|
||||
occurrence = MeterReadingOccurrence(
|
||||
schedule=schedule, target_date=target_date, operational_date=operational_date,
|
||||
assigned_to_id=schedule.assigned_to_id,
|
||||
)
|
||||
db.session.add(occurrence)
|
||||
db.session.flush()
|
||||
occurrence.round_occurrence = _round_occurrence_for(schedule, operational_date)
|
||||
created.append(occurrence)
|
||||
if commit:
|
||||
db.session.commit()
|
||||
return created
|
||||
|
||||
|
||||
def ensure_occurrences_for_day(day, user_id=None, lookback_days=365, horizon_days=31):
|
||||
schedules = MeterReadingSchedule.query.filter_by(is_active=True)
|
||||
if user_id is not None:
|
||||
schedules = schedules.filter(MeterReadingSchedule.assigned_to_id == user_id)
|
||||
ids = [item.id for item in schedules.all()]
|
||||
return generate_occurrences(
|
||||
start_date=day - timedelta(days=lookback_days),
|
||||
end_date=day + timedelta(days=horizon_days),
|
||||
schedule_ids=ids,
|
||||
) if ids else []
|
||||
|
||||
|
||||
def list_open_occurrences(day, user_id=None):
|
||||
query = MeterReadingOccurrence.query.filter(
|
||||
MeterReadingOccurrence.status == TODO,
|
||||
MeterReadingOccurrence.operational_date <= day,
|
||||
).options(
|
||||
joinedload(MeterReadingOccurrence.schedule).joinedload(MeterReadingSchedule.meter),
|
||||
joinedload(MeterReadingOccurrence.assigned_to),
|
||||
)
|
||||
if user_id is not None:
|
||||
query = query.filter(MeterReadingOccurrence.assigned_to_id == user_id)
|
||||
return query.order_by(MeterReadingOccurrence.operational_date, MeterReadingOccurrence.id).all()
|
||||
|
||||
|
||||
def _locked_occurrence(occurrence_id):
|
||||
occurrence = db.session.execute(
|
||||
select(MeterReadingOccurrence).where(MeterReadingOccurrence.id == occurrence_id).with_for_update()
|
||||
).scalar_one()
|
||||
if occurrence.status != TODO:
|
||||
raise MeterDomainError("Cette échéance est déjà traitée ou annulée.")
|
||||
return occurrence
|
||||
|
||||
|
||||
def record_occurrence_reading(*, occurrence_id, value, user_id, notes=None,
|
||||
is_reset=False, reset_reason=None,
|
||||
photo_filename=None, photo_path=None, reading_date=None):
|
||||
occurrence = _locked_occurrence(occurrence_id)
|
||||
if occurrence.assigned_to_id not in (None, user_id):
|
||||
raise MeterDomainError("Cette échéance n'est pas attribuée à cet utilisateur.")
|
||||
reading = record_meter_reading(
|
||||
meter=occurrence.schedule.meter, value=value, user_id=user_id,
|
||||
notes=notes, is_reset=is_reset, reset_reason=reset_reason,
|
||||
photo_filename=photo_filename, photo_path=photo_path,
|
||||
reading_date=reading_date, commit=False,
|
||||
)
|
||||
occurrence.status = TREATED_READING
|
||||
occurrence.meter_reading = reading
|
||||
occurrence.processed_by_id = user_id
|
||||
occurrence.processed_at = datetime.now()
|
||||
_refresh_round_status(occurrence.round_occurrence)
|
||||
db.session.commit()
|
||||
return occurrence
|
||||
|
||||
|
||||
def record_occurrence_without_reading(*, occurrence_id, reason, user_id, comment=None):
|
||||
if reason not in NO_READING_REASONS:
|
||||
raise MeterDomainError("Motif de relevé non obtenu inconnu.")
|
||||
occurrence = _locked_occurrence(occurrence_id)
|
||||
if occurrence.assigned_to_id not in (None, user_id):
|
||||
raise MeterDomainError("Cette échéance n'est pas attribuée à cet utilisateur.")
|
||||
occurrence.status = TREATED_WITHOUT_READING
|
||||
occurrence.no_reading_reason = reason
|
||||
occurrence.comment = (comment or "").strip() or None
|
||||
occurrence.processed_by_id = user_id
|
||||
occurrence.processed_at = datetime.now()
|
||||
_refresh_round_status(occurrence.round_occurrence)
|
||||
db.session.commit()
|
||||
return occurrence
|
||||
|
||||
|
||||
def create_round(*, name, default_assigned_to_id=None, estimated_duration_minutes=None, commit=True):
|
||||
name = (name or "").strip()
|
||||
if not name:
|
||||
raise MeterDomainError("Le nom de la tournée est obligatoire.")
|
||||
round_ = MeterReadingRound(
|
||||
name=name, default_assigned_to_id=default_assigned_to_id,
|
||||
estimated_duration_minutes=estimated_duration_minutes,
|
||||
)
|
||||
db.session.add(round_)
|
||||
if commit:
|
||||
db.session.commit()
|
||||
return round_
|
||||
|
||||
|
||||
def add_round_member(*, round_, schedule, position, commit=True):
|
||||
if position is None or int(position) < 1:
|
||||
raise MeterDomainError("La position de tournée doit être positive.")
|
||||
member = MeterReadingRoundMember(round=round_, schedule=schedule, position=int(position))
|
||||
db.session.add(member)
|
||||
if commit:
|
||||
db.session.commit()
|
||||
return member
|
||||
184
app_new/planning/meter_readings.py
Normal file
184
app_new/planning/meter_readings.py
Normal file
|
|
@ -0,0 +1,184 @@
|
|||
"""Routes C2 : règles, échéances et tournées de relevés."""
|
||||
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
from flask import abort, current_app, flash, jsonify, redirect, render_template, request, url_for
|
||||
from flask_login import current_user, login_required
|
||||
from werkzeug.utils import secure_filename
|
||||
|
||||
from ..extensions import db
|
||||
from ..core.models.planning import (
|
||||
Meter, MeterReadingOccurrence, MeterReadingRound, MeterReadingSchedule,
|
||||
)
|
||||
from ..core.models.user import User
|
||||
from ..core.services.meter_reading_planning import (
|
||||
FREQUENCIES, NO_READING_REASONS, MeterDomainError, add_round_member, create_or_update_schedule,
|
||||
create_round, ensure_occurrences_for_day, generate_occurrences,
|
||||
record_occurrence_reading, record_occurrence_without_reading,
|
||||
)
|
||||
from .schedules import planning_bp
|
||||
|
||||
|
||||
def _admin_required():
|
||||
if not current_user.is_admin():
|
||||
abort(403)
|
||||
|
||||
|
||||
@planning_bp.route('/meter-schedules')
|
||||
@login_required
|
||||
def meter_schedules():
|
||||
_admin_required()
|
||||
schedules = MeterReadingSchedule.query.order_by(MeterReadingSchedule.id.desc()).all()
|
||||
return render_template('planning/meter_schedules.html', schedules=schedules)
|
||||
|
||||
|
||||
@planning_bp.route('/meter-schedules/new', methods=['GET', 'POST'])
|
||||
@login_required
|
||||
def new_meter_schedule():
|
||||
_admin_required()
|
||||
meters = Meter.query.filter(Meter.status != 'replaced', Meter.is_active.is_(True)).order_by(Meter.name).all()
|
||||
users = User.query.filter_by(is_active=True).order_by(User.full_name, User.username).all()
|
||||
if request.method == 'POST':
|
||||
meter = db.session.get(Meter, request.form.get('meter_id', type=int))
|
||||
if not meter:
|
||||
flash('Compteur invalide.', 'danger')
|
||||
else:
|
||||
try:
|
||||
reference_date = date.fromisoformat(request.form.get('reference_date'))
|
||||
target_time = datetime.strptime(request.form['target_time'], '%H:%M').time() if request.form.get('target_time') else None
|
||||
create_or_update_schedule(
|
||||
meter=meter, frequency=request.form.get('frequency'), reference_date=reference_date,
|
||||
target_time=target_time, duration_minutes=request.form.get('duration_minutes', type=int, default=5),
|
||||
assigned_to_id=request.form.get('assigned_to_id', type=int),
|
||||
calendar_scope=request.form.get('calendar_scope') or None,
|
||||
fixed_month=request.form.get('fixed_month', type=int), fixed_day=request.form.get('fixed_day', type=int),
|
||||
)
|
||||
flash('Règle de relevé enregistrée.', 'success')
|
||||
return redirect(url_for('planning.meter_schedules'))
|
||||
except (ValueError, TypeError) as exc:
|
||||
flash(str(exc) or 'Date ou heure invalide.', 'danger')
|
||||
return render_template('planning/meter_schedule_form.html', meters=meters, users=users, frequencies=sorted(FREQUENCIES))
|
||||
|
||||
|
||||
@planning_bp.route('/meter-occurrences/<int:occurrence_id>')
|
||||
@login_required
|
||||
def meter_occurrence_detail(occurrence_id):
|
||||
occurrence = MeterReadingOccurrence.query.get_or_404(occurrence_id)
|
||||
if occurrence.assigned_to_id not in (None, current_user.id) and not current_user.is_admin():
|
||||
abort(403)
|
||||
return render_template(
|
||||
'planning/meter_occurrence.html', occurrence=occurrence,
|
||||
no_reading_reasons=NO_READING_REASONS,
|
||||
)
|
||||
|
||||
|
||||
def _optional_photo():
|
||||
photo = request.files.get('photo')
|
||||
if not photo or not photo.filename:
|
||||
return None, None
|
||||
allowed = {'jpg', 'jpeg', 'png', 'gif'}
|
||||
if '.' not in photo.filename or photo.filename.rsplit('.', 1)[1].lower() not in allowed:
|
||||
raise ValueError('La photo doit être une image JPG, PNG ou GIF.')
|
||||
safe_name = secure_filename(photo.filename)
|
||||
upload_dir = Path(current_app.config.get('UPLOAD_FOLDER', 'uploads')) / 'meter-readings'
|
||||
upload_dir.mkdir(parents=True, exist_ok=True)
|
||||
path = upload_dir / f'{uuid4().hex}_{safe_name}'
|
||||
photo.save(path)
|
||||
return path.name, str(path)
|
||||
|
||||
|
||||
@planning_bp.route('/meter-occurrences/<int:occurrence_id>/reading', methods=['POST'])
|
||||
@login_required
|
||||
def meter_occurrence_reading(occurrence_id):
|
||||
photo_path = None
|
||||
try:
|
||||
photo_filename, photo_path = _optional_photo()
|
||||
record_occurrence_reading(
|
||||
occurrence_id=occurrence_id, value=request.form.get('value'), user_id=current_user.id,
|
||||
notes=request.form.get('notes'), is_reset=request.form.get('is_reset') == '1',
|
||||
reset_reason=request.form.get('reset_reason'), photo_filename=photo_filename, photo_path=photo_path,
|
||||
)
|
||||
flash('Relevé enregistré et échéance traitée.', 'success')
|
||||
except (ValueError, MeterDomainError) as exc:
|
||||
db.session.rollback()
|
||||
if photo_path:
|
||||
Path(photo_path).unlink(missing_ok=True)
|
||||
flash(str(exc), 'danger')
|
||||
return redirect(url_for('planning.meter_occurrence_detail', occurrence_id=occurrence_id))
|
||||
|
||||
|
||||
@planning_bp.route('/meter-occurrences/<int:occurrence_id>/without-reading', methods=['POST'])
|
||||
@login_required
|
||||
def meter_occurrence_without_reading(occurrence_id):
|
||||
try:
|
||||
record_occurrence_without_reading(
|
||||
occurrence_id=occurrence_id, reason=request.form.get('reason'),
|
||||
user_id=current_user.id, comment=request.form.get('comment'),
|
||||
)
|
||||
flash('Échéance traitée sans relevé.', 'success')
|
||||
except MeterDomainError as exc:
|
||||
db.session.rollback()
|
||||
flash(str(exc), 'danger')
|
||||
return redirect(url_for('planning.meter_occurrence_detail', occurrence_id=occurrence_id))
|
||||
|
||||
|
||||
@planning_bp.route('/meter-rounds')
|
||||
@login_required
|
||||
def meter_rounds():
|
||||
_admin_required()
|
||||
rounds = MeterReadingRound.query.order_by(MeterReadingRound.name).all()
|
||||
return render_template('planning/meter_rounds.html', rounds=rounds)
|
||||
|
||||
|
||||
@planning_bp.route('/meter-rounds/new', methods=['GET', 'POST'])
|
||||
@login_required
|
||||
def new_meter_round():
|
||||
_admin_required()
|
||||
schedules = MeterReadingSchedule.query.filter_by(is_active=True).order_by(MeterReadingSchedule.id).all()
|
||||
users = User.query.filter_by(is_active=True).order_by(User.full_name, User.username).all()
|
||||
if request.method == 'POST':
|
||||
try:
|
||||
round_ = create_round(
|
||||
name=request.form.get('name'), default_assigned_to_id=request.form.get('assigned_to_id', type=int),
|
||||
estimated_duration_minutes=request.form.get('estimated_duration_minutes', type=int),
|
||||
commit=False,
|
||||
)
|
||||
selected = request.form.getlist('schedule_ids')
|
||||
for position, schedule_id in enumerate(selected, 1):
|
||||
schedule = db.session.get(MeterReadingSchedule, int(schedule_id))
|
||||
if schedule:
|
||||
add_round_member(round_=round_, schedule=schedule, position=position, commit=False)
|
||||
db.session.commit()
|
||||
flash('Tournée enregistrée.', 'success')
|
||||
return redirect(url_for('planning.meter_round_detail', round_id=round_.id))
|
||||
except (ValueError, TypeError, MeterDomainError) as exc:
|
||||
db.session.rollback()
|
||||
flash(str(exc), 'danger')
|
||||
return render_template('planning/meter_round_form.html', schedules=schedules, users=users)
|
||||
|
||||
|
||||
@planning_bp.route('/meter-rounds/<int:round_id>')
|
||||
@login_required
|
||||
def meter_round_detail(round_id):
|
||||
round_ = MeterReadingRound.query.get_or_404(round_id)
|
||||
if not current_user.is_admin():
|
||||
assigned = {member.schedule.assigned_to_id for member in round_.members}
|
||||
if round_.default_assigned_to_id != current_user.id and current_user.id not in assigned:
|
||||
abort(403)
|
||||
target_date = request.args.get('date', date.today().isoformat())
|
||||
try:
|
||||
target_date = date.fromisoformat(target_date)
|
||||
except ValueError:
|
||||
target_date = date.today()
|
||||
generate_occurrences(start_date=target_date, end_date=target_date)
|
||||
round_occurrence = next((item for item in round_.occurrences if item.operational_date == target_date), None)
|
||||
members = []
|
||||
if round_occurrence:
|
||||
members = sorted(round_occurrence.occurrences, key=lambda item: item.schedule.meter.name)
|
||||
members.sort(key=lambda item: next((m.position for m in round_.members if m.schedule_id == item.schedule_id), 9999))
|
||||
return render_template(
|
||||
'planning/meter_round_detail.html', round=round_, round_occurrence=round_occurrence,
|
||||
members=members, target_date=target_date,
|
||||
)
|
||||
|
|
@ -6,6 +6,7 @@ from . import schedules
|
|||
from . import closures
|
||||
from . import time_tracking
|
||||
from . import admin_rules
|
||||
from . import meter_readings
|
||||
from .schedules import planning_bp
|
||||
|
||||
__all__ = ["planning_bp"]
|
||||
|
|
|
|||
|
|
@ -20,6 +20,8 @@
|
|||
<h2 class="mb-0 h5"><i class="bi bi-calendar3"></i> Planning</h2>
|
||||
<div class="btn-group btn-group-sm">
|
||||
<a href="{{ url_for('planning.availability') }}" class="btn btn-outline-primary"><i class="bi bi-clock"></i> <span class="d-none d-sm-inline">Horaires</span></a>
|
||||
{% if current_user.is_admin() %}<a href="{{ url_for('planning.meter_schedules') }}" class="btn btn-outline-primary"><i class="bi bi-speedometer2"></i> <span class="d-none d-sm-inline">Relevés</span></a>{% endif %}
|
||||
{% if current_user.is_admin() %}<a href="{{ url_for('planning.meter_rounds') }}" class="btn btn-outline-primary"><i class="bi bi-sign-turn-right"></i> <span class="d-none d-sm-inline">Tournées</span></a>{% endif %}
|
||||
<a href="{{ url_for('planning.help_page') }}" class="btn btn-outline-secondary"><i class="bi bi-question-circle"></i> <span class="d-none d-sm-inline">Aide</span></a>
|
||||
</div>
|
||||
</div>
|
||||
|
|
|
|||
7
app_new/templates/planning/meter_occurrence.html
Normal file
7
app_new/templates/planning/meter_occurrence.html
Normal file
|
|
@ -0,0 +1,7 @@
|
|||
{% extends "base.html" %}
|
||||
{% block title %}Relevé — {{ occurrence.schedule.meter.name }}{% endblock %}
|
||||
{% block content %}
|
||||
<div class="container py-3"><a href="{{ url_for('planning.my_day', date=occurrence.operational_date.isoformat()) }}">← Ma journée</a><div class="card mt-3"><div class="card-body"><h1 class="h4">{{ occurrence.schedule.meter.name }}</h1><p class="mb-1">Échéance cible : <strong>{{ occurrence.target_date.strftime('%d/%m/%Y') }}</strong></p><p class="mb-1">Date opérationnelle : {{ occurrence.operational_date.strftime('%d/%m/%Y') }}</p><p class="mb-1">Durée estimée : {{ occurrence.schedule.duration_minutes }} min</p><p class="text-muted">{{ occurrence.schedule.meter.usage or 'Usage non précisé' }}</p>
|
||||
{% if occurrence.status == 'À FAIRE' %}<form method="post" enctype="multipart/form-data" action="{{ url_for('planning.meter_occurrence_reading', occurrence_id=occurrence.id) }}" class="mt-3"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><label class="form-label" for="value">Nouvelle valeur ({{ occurrence.schedule.meter.unit }})</label><input class="form-control" type="number" step="0.01" inputmode="decimal" id="value" name="value" value="{{ occurrence.schedule.meter.current_value }}" required><label class="form-label mt-2" for="notes">Notes</label><textarea class="form-control" id="notes" name="notes"></textarea><label class="form-label mt-2" for="photo">Photo facultative</label><input class="form-control" type="file" id="photo" name="photo" accept="image/jpeg,image/png,image/gif"><button class="btn btn-primary mt-3">Valider le relevé</button></form><hr id="sans-releve"><form method="post" action="{{ url_for('planning.meter_occurrence_without_reading', occurrence_id=occurrence.id) }}" class="mt-3"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><label class="form-label" for="reason">Traiter sans relevé</label><select class="form-select" id="reason" name="reason" required>{% for key, label in no_reading_reasons.items() %}<option value="{{ key }}">{{ label }}</option>{% endfor %}</select><textarea class="form-control mt-2" name="comment" placeholder="Commentaire facultatif"></textarea><button class="btn btn-outline-secondary mt-2">Traiter sans relevé</button></form>{% else %}<div class="alert alert-info">{{ occurrence.status }}{% if occurrence.no_reading_reason %} — {{ no_reading_reasons.get(occurrence.no_reading_reason, occurrence.no_reading_reason) }}{% endif %}{% if occurrence.comment %}<br>{{ occurrence.comment }}{% endif %}</div>{% endif %}
|
||||
</div></div></div>
|
||||
{% endblock %}
|
||||
3
app_new/templates/planning/meter_round_detail.html
Normal file
3
app_new/templates/planning/meter_round_detail.html
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
{% extends "base.html" %}
|
||||
{% block title %}{{ round.name }}{% endblock %}
|
||||
{% block content %}<div class="container py-3"><a href="{{ url_for('planning.meter_rounds') }}">← Tournées</a><h1 class="h3 mt-2">{{ round.name }}</h1>{% set done = members|selectattr('status', 'ne', 'À FAIRE')|list|length %}<p>{{ done }} / {{ members|length }} relevé(s) traité(s) — {{ target_date.strftime('%d/%m/%Y') }}</p><div class="list-group">{% for occurrence in members %}<div class="list-group-item d-flex justify-content-between align-items-center"><div><strong>{{ loop.index }}. {{ occurrence.schedule.meter.name }}</strong><div class="small text-muted">{{ occurrence.schedule.meter.current_value }} {{ occurrence.schedule.meter.unit }} · {{ occurrence.schedule.duration_minutes }} min</div></div>{% if occurrence.status == 'À FAIRE' %}<a class="btn btn-sm btn-primary" href="{{ url_for('planning.meter_occurrence_detail', occurrence_id=occurrence.id) }}">Relever</a>{% else %}<span class="badge text-bg-success">{{ occurrence.status }}</span>{% endif %}</div>{% else %}<div class="alert alert-info">Aucune occurrence pour cette date.</div>{% endfor %}</div></div>{% endblock %}
|
||||
3
app_new/templates/planning/meter_round_form.html
Normal file
3
app_new/templates/planning/meter_round_form.html
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
{% extends "base.html" %}
|
||||
{% block title %}Nouvelle tournée de relevés{% endblock %}
|
||||
{% block content %}<div class="container py-3"><h1 class="h3">Nouvelle tournée de relevés</h1><form method="post"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><label class="form-label" for="name">Nom</label><input class="form-control" id="name" name="name" required><label class="form-label mt-2" for="assigned_to_id">Responsable par défaut</label><select class="form-select" id="assigned_to_id" name="assigned_to_id"><option value="">Aucun</option>{% for user in users %}<option value="{{ user.id }}">{{ user.full_name or user.username }}</option>{% endfor %}</select><h2 class="h5 mt-3">Membres dans l'ordre</h2>{% for schedule in schedules %}<div class="form-check"><input class="form-check-input" type="checkbox" name="schedule_ids" value="{{ schedule.id }}" id="schedule-{{ schedule.id }}"><label class="form-check-label" for="schedule-{{ schedule.id }}">{{ schedule.meter.name }}</label></div>{% endfor %}<button class="btn btn-primary mt-3">Créer la tournée</button></form></div>{% endblock %}
|
||||
3
app_new/templates/planning/meter_rounds.html
Normal file
3
app_new/templates/planning/meter_rounds.html
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
{% extends "base.html" %}
|
||||
{% block title %}Tournées de relevés{% endblock %}
|
||||
{% block content %}<div class="container-fluid py-3"><div class="d-flex justify-content-between align-items-center"><h1 class="h3">Tournées de relevés</h1><a class="btn btn-primary" href="{{ url_for('planning.new_meter_round') }}">Nouvelle tournée</a></div><div class="list-group mt-3">{% for round in rounds %}<a class="list-group-item list-group-item-action" href="{{ url_for('planning.meter_round_detail', round_id=round.id) }}"><strong>{{ round.name }}</strong><span class="float-end">{{ round.members|length }} compteur(s)</span></a>{% else %}<div class="text-muted">Aucune tournée.</div>{% endfor %}</div></div>{% endblock %}
|
||||
17
app_new/templates/planning/meter_schedule_form.html
Normal file
17
app_new/templates/planning/meter_schedule_form.html
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
{% extends "base.html" %}
|
||||
{% block title %}Nouvelle règle de relevé{% endblock %}
|
||||
{% block content %}
|
||||
<div class="container py-3"><h1 class="h3">Nouvelle règle de relevé</h1>
|
||||
<form method="post" class="row g-3">
|
||||
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
|
||||
<div class="col-md-6"><label class="form-label" for="meter_id">Compteur</label><select class="form-select" id="meter_id" name="meter_id" required>{% for meter in meters %}<option value="{{ meter.id }}">{{ meter.name }} — {{ meter.unit }}</option>{% endfor %}</select></div>
|
||||
<div class="col-md-6"><label class="form-label" for="frequency">Fréquence</label><select class="form-select" id="frequency" name="frequency" required>{% for frequency in frequencies %}<option value="{{ frequency }}">{{ frequency }}</option>{% endfor %}</select></div>
|
||||
<div class="col-md-4"><label class="form-label" for="reference_date">Date de début/référence</label><input class="form-control" type="date" id="reference_date" name="reference_date" required></div>
|
||||
<div class="col-md-4"><label class="form-label" for="target_time">Heure cible facultative</label><input class="form-control" type="time" id="target_time" name="target_time"></div>
|
||||
<div class="col-md-4"><label class="form-label" for="duration_minutes">Durée estimée (minutes)</label><input class="form-control" type="number" min="1" id="duration_minutes" name="duration_minutes" value="5" required></div>
|
||||
<div class="col-md-4"><label class="form-label" for="assigned_to_id">Responsable</label><select class="form-select" id="assigned_to_id" name="assigned_to_id"><option value="">Non attribué</option>{% for user in users %}<option value="{{ user.id }}">{{ user.full_name or user.username }}</option>{% endfor %}</select></div>
|
||||
<div class="col-md-4"><label class="form-label" for="calendar_scope">Calendrier</label><select class="form-select" id="calendar_scope" name="calendar_scope"><option value="school">Collège</option><option value="housing">Logement de fonction</option></select></div>
|
||||
<div class="col-md-4"><label class="form-label" for="fixed_month">Date annuelle fixe (mois/jour)</label><div class="input-group"><input class="form-control" type="number" min="1" max="12" id="fixed_month" name="fixed_month" placeholder="Mois"><input class="form-control" type="number" min="1" max="31" name="fixed_day" placeholder="Jour"></div></div>
|
||||
<div class="col-12"><button class="btn btn-primary">Enregistrer</button> <a class="btn btn-outline-secondary" href="{{ url_for('planning.meter_schedules') }}">Annuler</a></div>
|
||||
</form></div>
|
||||
{% endblock %}
|
||||
6
app_new/templates/planning/meter_schedules.html
Normal file
6
app_new/templates/planning/meter_schedules.html
Normal file
|
|
@ -0,0 +1,6 @@
|
|||
{% extends "base.html" %}
|
||||
{% block title %}Règles de relevé{% endblock %}
|
||||
{% block content %}
|
||||
<div class="container-fluid py-3"><div class="d-flex justify-content-between align-items-center mb-3"><h1 class="h3">Règles de relevé</h1><a class="btn btn-primary" href="{{ url_for('planning.new_meter_schedule') }}">Nouvelle règle</a></div>
|
||||
<div class="table-responsive"><table class="table table-hover"><thead><tr><th>Compteur</th><th>Fréquence</th><th>Référence</th><th>Calendrier</th><th>Responsable</th><th>Durée</th><th>État</th></tr></thead><tbody>{% for schedule in schedules %}<tr><td>{{ schedule.meter.name }}</td><td>{{ schedule.frequency }}</td><td>{{ schedule.reference_date.strftime('%d/%m/%Y') }}</td><td>{{ 'Logement' if schedule.calendar_scope == 'housing' else 'Collège' }}</td><td>{{ schedule.assigned_to.full_name if schedule.assigned_to else 'Non attribué' }}</td><td>{{ schedule.duration_minutes }} min</td><td>{{ 'Active' if schedule.is_active else 'Inactive' }}</td></tr>{% else %}<tr><td colspan="7" class="text-muted">Aucune règle configurée.</td></tr>{% endfor %}</tbody></table></div></div>
|
||||
{% endblock %}
|
||||
|
|
@ -44,6 +44,11 @@
|
|||
</div>
|
||||
<h2 class="h5 mb-1 mt-1">{{ item.title }}</h2>
|
||||
<div class="small text-muted">{{ item.location_label }}</div>
|
||||
{% if item.source_type == 'meter_reading_occurrence' and item.due_date %}
|
||||
<div class="small {% if item.due_date < target_date %}text-danger{% else %}text-muted{% endif %}">
|
||||
Échéance cible : {{ item.due_date.strftime('%d/%m/%Y') }}{% if item.due_date < target_date %} · en retard{% endif %}
|
||||
</div>
|
||||
{% endif %}
|
||||
</div>
|
||||
<div class="text-end">
|
||||
<span class="badge text-bg-secondary">{{ item.type_label }}</span>
|
||||
|
|
@ -52,6 +57,12 @@
|
|||
</div>
|
||||
</div>
|
||||
{% if item.explanation %}<p class="mb-0 mt-2 small"><i class="bi bi-info-circle"></i> {{ item.explanation }}</p>{% endif %}
|
||||
{% if item.source_type == 'meter_reading_occurrence' %}
|
||||
<div class="d-flex flex-wrap gap-2 mt-3">
|
||||
<a class="btn btn-sm btn-primary" href="{{ url_for('planning.meter_occurrence_detail', occurrence_id=item.source_id) }}">Saisir le relevé</a>
|
||||
<a class="btn btn-sm btn-outline-secondary" href="{{ url_for('planning.meter_occurrence_detail', occurrence_id=item.source_id) }}#sans-releve">Traiter sans relevé</a>
|
||||
</div>
|
||||
{% endif %}
|
||||
{% if item.source_type == 'scheduled_task' and execution_states.get(item.source_id) %}
|
||||
{% set execution = execution_states.get(item.source_id) %}
|
||||
<div class="small mt-2">Temps réalisé : {{ execution.realized }} min · restant : {{ execution.remaining }} min</div>
|
||||
|
|
|
|||
129
migrations/versions/o0d1e2f3g4h5_meter_reading_c2.py
Normal file
129
migrations/versions/o0d1e2f3g4h5_meter_reading_c2.py
Normal file
|
|
@ -0,0 +1,129 @@
|
|||
"""Checkpoint C2: règles, occurrences et tournées de relevés."""
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
revision = "o0d1e2f3g4h5"
|
||||
down_revision = "n9c0d1e2f3g4"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade():
|
||||
op.create_table(
|
||||
"meter_reading_schedules",
|
||||
sa.Column("id", sa.Integer(), primary_key=True),
|
||||
sa.Column("meter_id", sa.Integer(), nullable=False),
|
||||
sa.Column("is_active", sa.Boolean(), nullable=False, server_default="1"),
|
||||
sa.Column("frequency", sa.String(30), nullable=False),
|
||||
sa.Column("reference_date", sa.Date(), nullable=False),
|
||||
sa.Column("target_time", sa.Time(), nullable=True),
|
||||
sa.Column("duration_minutes", sa.Integer(), nullable=False, server_default="5"),
|
||||
sa.Column("assigned_to_id", sa.Integer(), nullable=True),
|
||||
sa.Column("closed_day_policy", sa.String(30), nullable=False, server_default="PREVIOUS_WORKING_DAY"),
|
||||
sa.Column("calendar_scope", sa.String(20), nullable=False, server_default="school"),
|
||||
sa.Column("fixed_month", sa.Integer(), nullable=True),
|
||||
sa.Column("fixed_day", sa.Integer(), nullable=True),
|
||||
sa.Column("created_at", sa.DateTime(), nullable=False),
|
||||
sa.Column("updated_at", sa.DateTime(), nullable=False),
|
||||
sa.ForeignKeyConstraint(["meter_id"], ["meters.id"], ondelete="CASCADE"),
|
||||
sa.ForeignKeyConstraint(["assigned_to_id"], ["users.id"], ondelete="SET NULL"),
|
||||
sa.UniqueConstraint("meter_id", name="uq_meter_reading_schedule_meter"),
|
||||
sa.CheckConstraint("frequency in ('WEEKLY', 'MONTHLY', 'SEMI_ANNUAL', 'ANNUAL', 'FIXED_ANNUAL_DATE')", name="ck_meter_reading_schedule_frequency"),
|
||||
)
|
||||
op.create_index("ix_meter_reading_schedules_meter_id", "meter_reading_schedules", ["meter_id"])
|
||||
op.create_index("ix_meter_reading_schedules_assigned_to_id", "meter_reading_schedules", ["assigned_to_id"])
|
||||
|
||||
op.create_table(
|
||||
"meter_reading_rounds",
|
||||
sa.Column("id", sa.Integer(), primary_key=True),
|
||||
sa.Column("name", sa.String(255), nullable=False),
|
||||
sa.Column("is_active", sa.Boolean(), nullable=False, server_default="1"),
|
||||
sa.Column("default_assigned_to_id", sa.Integer(), nullable=True),
|
||||
sa.Column("estimated_duration_minutes", sa.Integer(), nullable=True),
|
||||
sa.Column("created_at", sa.DateTime(), nullable=False),
|
||||
sa.Column("updated_at", sa.DateTime(), nullable=False),
|
||||
sa.ForeignKeyConstraint(["default_assigned_to_id"], ["users.id"], ondelete="SET NULL"),
|
||||
)
|
||||
op.create_index("ix_meter_reading_rounds_default_assigned_to_id", "meter_reading_rounds", ["default_assigned_to_id"])
|
||||
|
||||
op.create_table(
|
||||
"meter_reading_round_members",
|
||||
sa.Column("id", sa.Integer(), primary_key=True),
|
||||
sa.Column("round_id", sa.Integer(), nullable=False),
|
||||
sa.Column("schedule_id", sa.Integer(), nullable=False),
|
||||
sa.Column("position", sa.Integer(), nullable=False),
|
||||
sa.ForeignKeyConstraint(["round_id"], ["meter_reading_rounds.id"], ondelete="CASCADE"),
|
||||
sa.ForeignKeyConstraint(["schedule_id"], ["meter_reading_schedules.id"], ondelete="CASCADE"),
|
||||
sa.UniqueConstraint("round_id", "schedule_id", name="uq_meter_round_schedule"),
|
||||
sa.UniqueConstraint("round_id", "position", name="uq_meter_round_position"),
|
||||
)
|
||||
op.create_index("ix_meter_reading_round_members_round_id", "meter_reading_round_members", ["round_id"])
|
||||
op.create_index("ix_meter_reading_round_members_schedule_id", "meter_reading_round_members", ["schedule_id"])
|
||||
|
||||
op.create_table(
|
||||
"meter_reading_round_occurrences",
|
||||
sa.Column("id", sa.Integer(), primary_key=True),
|
||||
sa.Column("round_id", sa.Integer(), nullable=False),
|
||||
sa.Column("operational_date", sa.Date(), nullable=False),
|
||||
sa.Column("status", sa.String(30), nullable=False, server_default="À FAIRE"),
|
||||
sa.Column("created_at", sa.DateTime(), nullable=False),
|
||||
sa.ForeignKeyConstraint(["round_id"], ["meter_reading_rounds.id"], ondelete="CASCADE"),
|
||||
sa.UniqueConstraint("round_id", "operational_date", name="uq_meter_round_occurrence_date"),
|
||||
)
|
||||
op.create_index("ix_meter_reading_round_occurrences_round_id", "meter_reading_round_occurrences", ["round_id"])
|
||||
op.create_index("ix_meter_reading_round_occurrences_operational_date", "meter_reading_round_occurrences", ["operational_date"])
|
||||
|
||||
op.create_table(
|
||||
"meter_reading_occurrences",
|
||||
sa.Column("id", sa.Integer(), primary_key=True),
|
||||
sa.Column("schedule_id", sa.Integer(), nullable=False),
|
||||
sa.Column("target_date", sa.Date(), nullable=False),
|
||||
sa.Column("operational_date", sa.Date(), nullable=False),
|
||||
sa.Column("status", sa.String(30), nullable=False, server_default="À FAIRE"),
|
||||
sa.Column("assigned_to_id", sa.Integer(), nullable=True),
|
||||
sa.Column("meter_reading_id", sa.Integer(), nullable=True),
|
||||
sa.Column("round_occurrence_id", sa.Integer(), nullable=True),
|
||||
sa.Column("processed_by_id", sa.Integer(), nullable=True),
|
||||
sa.Column("processed_at", sa.DateTime(), nullable=True),
|
||||
sa.Column("no_reading_reason", sa.String(40), nullable=True),
|
||||
sa.Column("comment", sa.Text(), nullable=True),
|
||||
sa.Column("created_at", sa.DateTime(), nullable=False),
|
||||
sa.Column("updated_at", sa.DateTime(), nullable=False),
|
||||
sa.ForeignKeyConstraint(["schedule_id"], ["meter_reading_schedules.id"], ondelete="CASCADE"),
|
||||
sa.ForeignKeyConstraint(["assigned_to_id"], ["users.id"], ondelete="SET NULL"),
|
||||
sa.ForeignKeyConstraint(["meter_reading_id"], ["meter_readings.id"], ondelete="SET NULL"),
|
||||
sa.ForeignKeyConstraint(["round_occurrence_id"], ["meter_reading_round_occurrences.id"], ondelete="SET NULL"),
|
||||
sa.ForeignKeyConstraint(["processed_by_id"], ["users.id"], ondelete="SET NULL"),
|
||||
sa.UniqueConstraint("schedule_id", "target_date", name="uq_meter_reading_occurrence_target"),
|
||||
)
|
||||
for name, column in (
|
||||
("schedule_id", "schedule_id"), ("target_date", "target_date"),
|
||||
("operational_date", "operational_date"), ("assigned_to_id", "assigned_to_id"),
|
||||
("meter_reading_id", "meter_reading_id"), ("round_occurrence_id", "round_occurrence_id"),
|
||||
("processed_by_id", "processed_by_id"),
|
||||
):
|
||||
op.create_index(f"ix_meter_reading_occurrences_{name}", "meter_reading_occurrences", [column])
|
||||
|
||||
|
||||
def downgrade():
|
||||
for name in (
|
||||
"ix_meter_reading_occurrences_processed_by_id", "ix_meter_reading_occurrences_round_occurrence_id",
|
||||
"ix_meter_reading_occurrences_meter_reading_id", "ix_meter_reading_occurrences_assigned_to_id",
|
||||
"ix_meter_reading_occurrences_operational_date", "ix_meter_reading_occurrences_target_date",
|
||||
"ix_meter_reading_occurrences_schedule_id",
|
||||
):
|
||||
op.drop_index(name, table_name="meter_reading_occurrences")
|
||||
op.drop_table("meter_reading_occurrences")
|
||||
op.drop_index("ix_meter_reading_round_occurrences_operational_date", table_name="meter_reading_round_occurrences")
|
||||
op.drop_index("ix_meter_reading_round_occurrences_round_id", table_name="meter_reading_round_occurrences")
|
||||
op.drop_table("meter_reading_round_occurrences")
|
||||
op.drop_index("ix_meter_reading_round_members_schedule_id", table_name="meter_reading_round_members")
|
||||
op.drop_index("ix_meter_reading_round_members_round_id", table_name="meter_reading_round_members")
|
||||
op.drop_table("meter_reading_round_members")
|
||||
op.drop_index("ix_meter_reading_rounds_default_assigned_to_id", table_name="meter_reading_rounds")
|
||||
op.drop_table("meter_reading_rounds")
|
||||
op.drop_index("ix_meter_reading_schedules_assigned_to_id", table_name="meter_reading_schedules")
|
||||
op.drop_index("ix_meter_reading_schedules_meter_id", table_name="meter_reading_schedules")
|
||||
op.drop_table("meter_reading_schedules")
|
||||
229
tests/integration/test_meter_checkpoint_c2.py
Normal file
229
tests/integration/test_meter_checkpoint_c2.py
Normal file
|
|
@ -0,0 +1,229 @@
|
|||
from datetime import date, time
|
||||
from time import perf_counter
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import event
|
||||
|
||||
from app_new import db
|
||||
from app_new.core.models.college import Building, HousingUnit
|
||||
from app_new.core.models.planning import (
|
||||
Meter, MeterReading, MeterReadingOccurrence, MeterReadingRoundOccurrence,
|
||||
MeterReadingSchedule, WorkSchedule,
|
||||
)
|
||||
from app_new.core.services.meter_reading_planning import (
|
||||
CANCELLED, TREATED_READING, TREATED_WITHOUT_READING, add_round_member,
|
||||
create_or_update_schedule, create_round, generate_occurrences,
|
||||
operational_date_for, record_occurrence_reading, record_occurrence_without_reading,
|
||||
)
|
||||
from app_new.core.services.meter_service import create_meter, replace_meter
|
||||
|
||||
|
||||
def _meter(name, building=None, housing=None):
|
||||
if building is None:
|
||||
building = Building(name=f"TEST_UI_C2_{name}_BUILDING")
|
||||
db.session.add(building)
|
||||
db.session.flush()
|
||||
return create_meter(name=f"TEST_UI_C2_{name}", building_id=building.id if housing is None else None, housing_unit_id=housing.id if housing else None)
|
||||
|
||||
|
||||
def _schedule(meter, admin_user, frequency, reference, **kwargs):
|
||||
return create_or_update_schedule(
|
||||
meter=meter, frequency=frequency, reference_date=reference,
|
||||
assigned_to_id=admin_user["id"], **kwargs,
|
||||
)
|
||||
|
||||
|
||||
def test_c2_frequencies_and_generation_are_idempotent(app, admin_user):
|
||||
with app.app_context():
|
||||
frequencies = {
|
||||
"WEEKLY": date(2026, 10, 5), "MONTHLY": date(2026, 10, 1),
|
||||
"SEMI_ANNUAL": date(2026, 10, 1), "ANNUAL": date(2026, 10, 1),
|
||||
"FIXED_ANNUAL_DATE": date(2026, 10, 1),
|
||||
}
|
||||
schedules = []
|
||||
for frequency, reference in frequencies.items():
|
||||
meter = _meter(f"FREQ_{frequency}")
|
||||
schedules.append(_schedule(meter, admin_user, frequency, reference, fixed_month=10, fixed_day=1))
|
||||
db.session.commit()
|
||||
first = generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2027, 10, 10))
|
||||
counts = {schedule.id: MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).count() for schedule in schedules}
|
||||
second = generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2027, 10, 10))
|
||||
assert first
|
||||
assert second == []
|
||||
assert all(MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).count() == counts[schedule.id] for schedule in schedules)
|
||||
assert counts[schedules[0].id] == 53
|
||||
assert counts[schedules[1].id] == 13
|
||||
assert counts[schedules[2].id] == 3
|
||||
assert counts[schedules[3].id] == 2
|
||||
assert counts[schedules[4].id] == 2
|
||||
|
||||
|
||||
def test_c2_target_and_operational_dates_and_housing_calendar(app, admin_user):
|
||||
with app.app_context():
|
||||
building = Building(name="TEST_UI_C2_CALENDAR_BUILDING")
|
||||
housing = HousingUnit(name="TEST_UI_C2_HOUSING", building=building)
|
||||
db.session.add_all([building, housing])
|
||||
db.session.flush()
|
||||
school_meter = _meter("SCHOOL_DATE", building=building)
|
||||
schedule = _schedule(school_meter, admin_user, "FIXED_ANNUAL_DATE", date(2026, 10, 3))
|
||||
db.session.add(WorkSchedule(user_id=admin_user["id"], day_of_week=4, start_time=time(8), end_time=time(17), is_active=True))
|
||||
db.session.flush()
|
||||
generate_occurrences(start_date=date(2026, 10, 3), end_date=date(2026, 10, 3), schedule_ids=[schedule.id])
|
||||
occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).one()
|
||||
assert occurrence.target_date == date(2026, 10, 3)
|
||||
assert occurrence.operational_date == date(2026, 10, 2)
|
||||
|
||||
housing_meter = _meter("HOUSING_DATE", housing=housing)
|
||||
housing_schedule = _schedule(housing_meter, admin_user, "ANNUAL", date(2026, 10, 3))
|
||||
generate_occurrences(start_date=date(2026, 10, 3), end_date=date(2026, 10, 3), schedule_ids=[housing_schedule.id])
|
||||
housing_occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=housing_schedule.id).one()
|
||||
assert housing_occurrence.operational_date == date(2026, 10, 3)
|
||||
|
||||
|
||||
def test_c2_processing_with_reading_and_without_reading_never_creates_zero(app, admin_user):
|
||||
with app.app_context():
|
||||
meter = _meter("PROCESSING")
|
||||
schedule = _schedule(meter, admin_user, "MONTHLY", date(2026, 10, 1))
|
||||
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2026, 10, 1), schedule_ids=[schedule.id])
|
||||
occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).one()
|
||||
processed = record_occurrence_reading(occurrence_id=occurrence.id, value=42, user_id=admin_user["id"])
|
||||
assert processed.status == TREATED_READING
|
||||
assert MeterReading.query.filter_by(meter_id=meter.id).count() == 1
|
||||
|
||||
other = _meter("WITHOUT")
|
||||
other_schedule = _schedule(other, admin_user, "MONTHLY", date(2026, 10, 1))
|
||||
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2026, 10, 1), schedule_ids=[other_schedule.id])
|
||||
other_occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=other_schedule.id).one()
|
||||
without = record_occurrence_without_reading(
|
||||
occurrence_id=other_occurrence.id, reason="OCCUPANT_REFUSED", user_id=admin_user["id"], comment="TEST_UI_C2 refus"
|
||||
)
|
||||
assert without.status == TREATED_WITHOUT_READING
|
||||
assert without.comment == "TEST_UI_C2 refus"
|
||||
assert MeterReading.query.filter_by(meter_id=other.id).count() == 0
|
||||
|
||||
|
||||
def test_c2_http_occurrence_actions_use_shared_service(app, admin_user, authenticated_client):
|
||||
with app.app_context():
|
||||
meter = _meter("HTTP_READING")
|
||||
schedule = _schedule(meter, admin_user, "MONTHLY", date(2026, 10, 1))
|
||||
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2026, 10, 1), schedule_ids=[schedule.id])
|
||||
occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).one()
|
||||
|
||||
response = authenticated_client.post(
|
||||
f"/planning/meter-occurrences/{occurrence.id}/reading",
|
||||
data={"value": "123.5", "notes": "TEST_UI_C2_HTTP"},
|
||||
)
|
||||
assert response.status_code == 302
|
||||
db.session.refresh(occurrence)
|
||||
assert occurrence.status == TREATED_READING
|
||||
assert MeterReading.query.filter_by(meter_id=meter.id).count() == 1
|
||||
|
||||
other = _meter("HTTP_WITHOUT")
|
||||
other_schedule = _schedule(other, admin_user, "MONTHLY", date(2026, 10, 1))
|
||||
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2026, 10, 1), schedule_ids=[other_schedule.id])
|
||||
other_occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=other_schedule.id).one()
|
||||
response = authenticated_client.post(
|
||||
f"/planning/meter-occurrences/{other_occurrence.id}/without-reading",
|
||||
data={"reason": "READING_UNKNOWN", "comment": "TEST_UI_C2_HTTP inconnu"},
|
||||
)
|
||||
assert response.status_code == 302
|
||||
db.session.refresh(other_occurrence)
|
||||
assert other_occurrence.status == TREATED_WITHOUT_READING
|
||||
assert other_occurrence.no_reading_reason == "READING_UNKNOWN"
|
||||
assert MeterReading.query.filter_by(meter_id=other.id).count() == 0
|
||||
|
||||
|
||||
def test_c2_open_occurrence_stays_visible_as_overdue_in_my_day(app, admin_user, authenticated_client):
|
||||
with app.app_context():
|
||||
target = date(2026, 10, 5)
|
||||
meter = _meter("MY_DAY")
|
||||
_schedule(meter, admin_user, "WEEKLY", date(2026, 9, 28), duration_minutes=5)
|
||||
db.session.add(WorkSchedule(user_id=admin_user["id"], day_of_week=0, start_time=time(8), end_time=time(17), is_active=True))
|
||||
db.session.commit()
|
||||
response = authenticated_client.get(f"/planning/my-day?date={target.isoformat()}")
|
||||
assert response.status_code == 200
|
||||
assert b"Relev\xc3\xa9 compteur" in response.data
|
||||
assert b"Saisir le relev\xc3\xa9" in response.data
|
||||
|
||||
|
||||
def test_c2_replaced_meter_cancels_open_occurrences(app, admin_user):
|
||||
with app.app_context():
|
||||
meter = _meter("REPLACED")
|
||||
schedule = _schedule(meter, admin_user, "MONTHLY", date(2026, 10, 1))
|
||||
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2026, 10, 1), schedule_ids=[schedule.id])
|
||||
old_occurrence = MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id).one()
|
||||
replace_meter(old_meter=meter, name="TEST_UI_C2_REPLACEMENT", user_id=admin_user["id"])
|
||||
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2027, 1, 1))
|
||||
db.session.refresh(old_occurrence)
|
||||
assert old_occurrence.status == CANCELLED
|
||||
assert MeterReadingOccurrence.query.filter_by(schedule_id=schedule.id, status="À FAIRE").count() == 0
|
||||
|
||||
|
||||
def test_c2_round_order_progress_and_resume(app, admin_user, authenticated_client):
|
||||
with app.app_context():
|
||||
first = _meter("ROUND_1")
|
||||
second = _meter("ROUND_2")
|
||||
first_schedule = _schedule(first, admin_user, "MONTHLY", date(2026, 10, 1))
|
||||
second_schedule = _schedule(second, admin_user, "MONTHLY", date(2026, 10, 1))
|
||||
round_ = create_round(name="TEST_UI_C2_TOURNEE", default_assigned_to_id=admin_user["id"])
|
||||
add_round_member(round_=round_, schedule=first_schedule, position=1, commit=False)
|
||||
add_round_member(round_=round_, schedule=second_schedule, position=2, commit=False)
|
||||
db.session.commit()
|
||||
generate_occurrences(start_date=date(2026, 10, 1), end_date=date(2026, 10, 1))
|
||||
round_occurrence = MeterReadingRoundOccurrence.query.filter_by(round_id=round_.id, operational_date=date(2026, 10, 1)).one()
|
||||
assert [item.schedule.meter.name for item in sorted(round_occurrence.occurrences, key=lambda item: item.schedule.meter.name)]
|
||||
response = authenticated_client.get(f"/planning/meter-rounds/{round_.id}?date=2026-10-01")
|
||||
assert response.status_code == 200
|
||||
assert b"TEST_UI_C2_TOURNEE" in response.data
|
||||
assert b"0 / 2" in response.data
|
||||
record_occurrence_without_reading(
|
||||
occurrence_id=round_occurrence.occurrences[0].id, reason="READING_UNKNOWN", user_id=admin_user["id"]
|
||||
)
|
||||
db.session.refresh(round_occurrence)
|
||||
assert round_occurrence.status == "EN COURS"
|
||||
|
||||
|
||||
def test_c2_permissions_keep_configuration_admin_only(app, client):
|
||||
response = client.get('/planning/meter-schedules/new')
|
||||
assert response.status_code in (302, 403)
|
||||
|
||||
|
||||
def test_c2_planning_sql_smoke_is_measured(app, admin_user, authenticated_client):
|
||||
with app.app_context():
|
||||
schedules = []
|
||||
for index in range(3):
|
||||
meter = _meter(f"PERF_{index}")
|
||||
schedules.append(_schedule(meter, admin_user, "MONTHLY", date(2026, 10, 1)))
|
||||
round_ = create_round(name="TEST_UI_C2_PERF_ROUND", default_assigned_to_id=admin_user["id"])
|
||||
for position, schedule in enumerate(schedules, 1):
|
||||
add_round_member(round_=round_, schedule=schedule, position=position, commit=False)
|
||||
db.session.commit()
|
||||
|
||||
query_count = 0
|
||||
|
||||
def count_queries(*_args, **_kwargs):
|
||||
nonlocal query_count
|
||||
query_count += 1
|
||||
|
||||
event.listen(db.engine, "before_cursor_execute", count_queries)
|
||||
try:
|
||||
start = perf_counter()
|
||||
daily = authenticated_client.get("/planning/my-day?date=2026-10-01")
|
||||
daily_seconds = perf_counter() - start
|
||||
daily_queries = query_count
|
||||
query_count = 0
|
||||
start = perf_counter()
|
||||
round_response = authenticated_client.get(
|
||||
f"/planning/meter-rounds/{round_.id}?date=2026-10-01"
|
||||
)
|
||||
round_seconds = perf_counter() - start
|
||||
round_queries = query_count
|
||||
finally:
|
||||
event.remove(db.engine, "before_cursor_execute", count_queries)
|
||||
|
||||
print(
|
||||
f"C2_PERF my_day_seconds={daily_seconds:.4f} my_day_queries={daily_queries} "
|
||||
f"round_seconds={round_seconds:.4f} round_queries={round_queries}"
|
||||
)
|
||||
assert daily.status_code == 200
|
||||
assert round_response.status_code == 200
|
||||
Loading…
Reference in a new issue