From 59e3f6d8c1ad0bd9ddbfea6e443e8dd27a41f249 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 24 Aug 2026 22:02:59 +0000 Subject: [PATCH] fix(c4): integrate scheduled readings and copier meter roles --- app_new/core/services/c4_service.py | 69 +++++- app_new/core/services/meter_service.py | 2 + app_new/meters/routes.py | 5 + app_new/meters/templates/meters/edit.html | 10 + app_new/meters/templates/meters/new.html | 1 + tests/integration/test_checkpoint_c4_final.py | 201 ++++++++++++++++++ 6 files changed, 280 insertions(+), 8 deletions(-) create mode 100644 tests/integration/test_checkpoint_c4_final.py diff --git a/app_new/core/services/c4_service.py b/app_new/core/services/c4_service.py index ee3d422..c7f5966 100644 --- a/app_new/core/services/c4_service.py +++ b/app_new/core/services/c4_service.py @@ -10,6 +10,7 @@ from ..models import ( ConsumableUsage, ContractEquipmentAssignment, Equipment, EquipmentConsumable, Meter, MeterReading, PhotocopierContract, PhotocopierContractPeriod, ) +from ..models.planning import MeterReadingOccurrence, MeterReadingSchedule from ...extensions import db @@ -108,23 +109,64 @@ def _meter_role(meter): return {"COPIER_BW": "bw", "COPIER_COLOR": "color"}.get(meter.contract_role) +def classify_legacy_copier_meter(meter, *, commit=True): + """Classifie prudemment un ancien compteur photocopieur. + + Cette aide est volontairement explicite et ponctuelle : le calcul C4 ne + l'appelle jamais comme repli implicite. Les noms ambigus restent NONE. + """ + if meter.contract_role != "NONE" or meter.meter_type != "photocopieur": + return meter.contract_role + text = f"{meter.name or ''} {meter.usage or ''}".casefold() + import re + if re.search(r"\b(noir\s*(et|&)\s*blanc|n\s*[&/]\s*b|\bnb\b)\b", text): + meter.contract_role = "COPIER_BW" + elif re.search(r"\b(couleur|color)\b", text): + meter.contract_role = "COPIER_COLOR" + if commit: + db.session.commit() + return meter.contract_role + + +def _c2_reference_ids(meter_id, target_date, reading_ids): + """Retourne les relevés liés à l'échéance C2 de ``target_date``.""" + if not reading_ids: + return set() + rows = (db.session.query(MeterReadingOccurrence.meter_reading_id) + .join(MeterReadingOccurrence.schedule) + .filter(MeterReadingSchedule.meter_id == meter_id, + MeterReadingOccurrence.target_date == target_date, + MeterReadingOccurrence.meter_reading_id.in_(reading_ids)) + .all()) + return {row[0] for row in rows} + + def _meter_consumption(meter, start_date, end_date): readings = (MeterReading.query.filter( MeterReading.meter_id == meter.id, MeterReading.reading_date <= datetime.combine(end_date, datetime.max.time()), ).order_by(MeterReading.reading_date, MeterReading.id).all()) - return _meter_consumption_from_rows(readings, start_date, end_date, meter=meter) + exact_ids = _c2_reference_ids(meter.id, start_date, [row.id for row in readings]) + return _meter_consumption_from_rows(readings, start_date, end_date, meter=meter, + exact_reference_ids=exact_ids) -def _meter_consumption_from_rows(rows, start_date, end_date, meter=None): +def _meter_consumption_from_rows(rows, start_date, end_date, meter=None, exact_reference_ids=None): """Même calcul que ``_meter_consumption`` à partir d'un lot déjà chargé.""" start_dt = datetime.combine(start_date, datetime.min.time()) end_dt = datetime.combine(end_date, datetime.max.time()) relevant = [row for row in rows if row.reading_date <= end_dt] before = next((row for row in reversed(relevant) if row.reading_date < start_dt), None) inside = [row for row in relevant if start_dt <= row.reading_date <= end_dt] - exact_start = bool(inside and inside[0].reading_date.date() == start_date) - ordered = inside + exact_reference_ids = exact_reference_ids or set() + linked_reference = next((row for row in relevant + if row.id in exact_reference_ids and row.reading_date <= end_dt), None) + exact_start = bool(linked_reference or (inside and inside[0].reading_date.date() == start_date)) + if linked_reference: + ordered = [linked_reference] + [row for row in inside if row.id != linked_reference.id] + ordered.sort(key=lambda row: (row.reading_date, row.id)) + else: + ordered = inside total = 0.0 previous = None for reading in ordered: @@ -140,11 +182,18 @@ def _meter_consumption_from_rows(rows, start_date, end_date, meter=None): prior_rates = [] for left, right in zip(prior, prior[1:]): if not left.is_reset and not right.is_reset and right.value >= left.value: - days = max((right.reading_date.date() - left.reading_date.date()).days, 1) - prior_rates.append((right.value - left.value) / days) + left_date = left.reading_date.date() + right_date = right.reading_date.date() + if (meter is not None and meter.equipment is not None + and getattr(meter.equipment, "device_type", "") in {"photocopier", "photocopieur", "imprimante", "printer"}): + days = _school_days(left_date, right_date - timedelta(days=1)) + else: + days = max((right_date - left_date).days, 1) + if days > 0: + prior_rates.append((right.value - left.value) / days) if first and prior_rates: missing_days = max((first.reading_date.date() - start_date).days, 0) - if meter is not None and meter.equipment is not None and getattr(meter.equipment, "device_type", "") in {"photocopier", "imprimante", "printer"}: + if meter is not None and meter.equipment is not None and getattr(meter.equipment, "device_type", "") in {"photocopier", "photocopieur", "imprimante", "printer"}: comparable_days = _school_days(start_date, first.reading_date.date() - timedelta(days=1)) missing_days = comparable_days or missing_days total += mean(prior_rates) * missing_days @@ -180,7 +229,11 @@ def calculate_contract_usage(period): role = _meter_role(meter) if role is None: continue - amount, quality, _ = _meter_consumption_from_rows(readings_by_meter.get(meter.id, []), start, end, meter=meter) + meter_rows = readings_by_meter.get(meter.id, []) + exact_ids = _c2_reference_ids(meter.id, start, [row.id for row in meter_rows]) + amount, quality, _ = _meter_consumption_from_rows( + meter_rows, start, end, meter=meter, exact_reference_ids=exact_ids + ) result[role] += amount rank = {"EXACT": 0, "ESTIMATED": 1, "PARTIAL": 2} if rank[quality] > rank[result["quality"]]: diff --git a/app_new/core/services/meter_service.py b/app_new/core/services/meter_service.py index 2480926..8bc48d7 100644 --- a/app_new/core/services/meter_service.py +++ b/app_new/core/services/meter_service.py @@ -62,6 +62,7 @@ def validate_meter_scope(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, + contract_role="NONE", warning_threshold=None, critical_threshold=None, maintenance_interval=None, last_maintenance_value=None, reading_interval_days=None, commit=True, **scope_ids): @@ -78,6 +79,7 @@ def create_meter(*, name, meter_type="hours", unit="h", usage=None, warning_threshold=warning_threshold, critical_threshold=critical_threshold, maintenance_interval=maintenance_interval, last_maintenance_value=last_maintenance_value, reading_interval_days=reading_interval_days, + contract_role=contract_role if contract_role in {"NONE", "COPIER_BW", "COPIER_COLOR"} else "NONE", **scope) if parent is not None: model.parent = parent diff --git a/app_new/meters/routes.py b/app_new/meters/routes.py index d4df451..4a96332 100644 --- a/app_new/meters/routes.py +++ b/app_new/meters/routes.py @@ -18,6 +18,8 @@ from app_new.core.services.meter_service import ( replace_meter, ) +CONTRACT_ROLES = {"NONE", "COPIER_BW", "COPIER_COLOR"} + meters_bp = Blueprint('meters', __name__, template_folder='templates') @@ -63,6 +65,7 @@ def new(): 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, + contract_role=request.form.get('contract_role', 'NONE'), **{scope_key: scope_id}, ) flash(f'Compteur « {meter.name} » créé.', 'success') @@ -171,6 +174,8 @@ def edit(meter_id): 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 + contract_role = request.form.get('contract_role', 'NONE') + meter.contract_role = contract_role if contract_role in CONTRACT_ROLES else 'NONE' meter.status = request.form.get('status') or meter.status try: from app_new.core.services.meter_service import validate_meter_scope diff --git a/app_new/meters/templates/meters/edit.html b/app_new/meters/templates/meters/edit.html index eab6f93..bae2643 100644 --- a/app_new/meters/templates/meters/edit.html +++ b/app_new/meters/templates/meters/edit.html @@ -84,6 +84,16 @@ value="{{ meter.unit }}" placeholder="Ex: h, km, m³"> + +
+ + + Rôle explicite utilisé par les quotas contractuels. Les compteurs ambigus restent « Aucun ». +
diff --git a/app_new/meters/templates/meters/new.html b/app_new/meters/templates/meters/new.html index 7eaaddd..ed21f44 100644 --- a/app_new/meters/templates/meters/new.html +++ b/app_new/meters/templates/meters/new.html @@ -10,6 +10,7 @@
+
diff --git a/tests/integration/test_checkpoint_c4_final.py b/tests/integration/test_checkpoint_c4_final.py new file mode 100644 index 0000000..7ca0eb3 --- /dev/null +++ b/tests/integration/test_checkpoint_c4_final.py @@ -0,0 +1,201 @@ +from datetime import date, datetime, time, timezone +import re + +from app_new.extensions import db +from app_new.core.models import ( + CollegeClosure, Company, ContractEquipmentAssignment, Equipment, Meter, + MeterReading, MeterReadingOccurrence, User, WorkSchedule, +) +from app_new.core.models.planning import MeterReadingSchedule +from app_new.core.services.c4_service import ( + _school_days, calculate_contract_usage, classify_legacy_copier_meter, create_contract, +) +from app_new.core.services.meter_service import create_meter +from app_new.core.services.meter_reading_planning import ( + create_or_update_schedule, generate_occurrences, record_occurrence_reading, +) + + +def _supplier(name): + row = Company(name=name) + db.session.add(row) + db.session.flush() + return row + + +def _machine(name): + row = Equipment(name=name, device_type="photocopier", + serial_number=f"TEST_UI_C4_FINAL_{name}") + db.session.add(row) + db.session.flush() + return row + + +def _meter(machine, name, role="COPIER_BW", current=0): + row = Meter(equipment=machine, name=name, meter_type="photocopieur", + unit="pages", contract_role=role, current_value=current) + db.session.add(row) + db.session.flush() + return row + + +def _reading(meter, value, day): + row = MeterReading( + meter=meter, value=value, + reading_date=datetime.combine(day, time.min, tzinfo=timezone.utc), + ) + db.session.add(row) + db.session.flush() + return row + + +def test_c4_final_c2_anticipated_reading_is_exact_contract_reference(app, admin_user): + with app.app_context(): + supplier = _supplier("TEST_UI_C4_FINAL_C2_SUPPLIER") + contract = create_contract( + supplier_id=supplier.id, name="TEST_UI_C4_FINAL_C2_CONTRACT", + start_date=date(2028, 10, 1), end_date=date(2029, 9, 30), + anniversary_month=10, anniversary_day=1, + ) + period = contract.periods[0] + machine = _machine("TEST_UI_C4_FINAL_C2_MACHINE") + db.session.add(ContractEquipmentAssignment( + contract=contract, equipment=machine, entry_date=period.start_date, + )) + meter = _meter(machine, "N&B", current=1000) + _reading(meter, 900, date(2028, 9, 1)) + _reading(meter, 1000, date(2028, 9, 15)) + + schedule = create_or_update_schedule( + meter=meter, frequency="FIXED_ANNUAL_DATE", + reference_date=date(2028, 10, 1), assigned_to_id=admin_user["id"], + fixed_month=10, fixed_day=1, commit=False, + ) + # Le calendrier C2 connaît le vendredi comme journée travaillée. + 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(2028, 10, 1), end_date=date(2028, 10, 1), + schedule_ids=[schedule.id], + ) + occurrence = MeterReadingOccurrence.query.filter_by( + schedule_id=schedule.id, target_date=date(2028, 10, 1), + ).one() + assert occurrence.operational_date == date(2028, 9, 29) + + processed = record_occurrence_reading( + occurrence_id=occurrence.id, value=1100, user_id=admin_user["id"], + reading_date=datetime(2028, 9, 29, tzinfo=timezone.utc), + ) + db.session.expire_all() + assert processed.meter_reading.reading_date.date() == date(2028, 9, 29) + assert processed.target_date == date(2028, 10, 1) + _reading(meter, 1200, date(2028, 10, 15)) + usage = calculate_contract_usage(period) + assert usage["quality"] == "EXACT" + assert usage["bw"] == 100 + + +def test_c4_final_early_spontaneous_reading_is_not_exact(app): + with app.app_context(): + supplier = _supplier("TEST_UI_C4_FINAL_SPONTANEOUS_SUPPLIER") + contract = create_contract( + supplier_id=supplier.id, name="TEST_UI_C4_FINAL_SPONTANEOUS", + start_date=date(2028, 10, 1), end_date=date(2029, 9, 30), + anniversary_month=10, anniversary_day=1, + ) + period = contract.periods[0] + machine = _machine("TEST_UI_C4_FINAL_SPONTANEOUS_MACHINE") + db.session.add(ContractEquipmentAssignment( + contract=contract, equipment=machine, entry_date=period.start_date, + )) + meter = _meter(machine, "N&B", current=1000) + _reading(meter, 900, date(2028, 9, 1)) + _reading(meter, 1000, date(2028, 9, 15)) + _reading(meter, 1100, date(2028, 9, 29)) + _reading(meter, 1200, date(2028, 10, 15)) + usage = calculate_contract_usage(period) + assert usage["quality"] in {"ESTIMATED", "PARTIAL"} + + +def test_c4_final_legacy_roles_are_safe_and_explicit(app): + with app.app_context(): + machine = _machine("TEST_UI_C4_FINAL_LEGACY_MACHINE") + bw = _meter(machine, "N&B", role="NONE") + color = _meter(machine, "Couleur", role="NONE") + ambiguous = _meter(machine, "Heures fonctionnement", role="NONE") + assert classify_legacy_copier_meter(bw, commit=False) == "COPIER_BW" + assert classify_legacy_copier_meter(color, commit=False) == "COPIER_COLOR" + assert classify_legacy_copier_meter(ambiguous, commit=False) == "NONE" + assert bw.contract_role == "COPIER_BW" + assert color.contract_role == "COPIER_COLOR" + assert ambiguous.contract_role == "NONE" + + +def test_c4_final_meter_contract_role_can_be_edited_http(app, admin_user, authenticated_client): + with app.app_context(): + machine = _machine("TEST_UI_C4_FINAL_HTTP_MACHINE") + meter = create_meter(name="Compteur historique", meter_type="photocopieur", + unit="pages", contract_role="NONE", equipment_id=machine.id) + meter_id = meter.id + machine_id = machine.id + db.session.commit() + assert db.session.get(Meter, meter_id) is not None + with app.app_context(): + token_page = authenticated_client.get(f'/meters/{meter_id}/edit') + assert token_page.status_code == 200 + token = re.search(r'name="csrf_token" value="([^"]+)"', + token_page.get_data(as_text=True)).group(1) + response = authenticated_client.post( + f"/meters/{meter_id}/edit", data={ + "name": "Compteur historique", "meter_type": "photocopieur", "unit": "pages", + "scope_type": "equipment", "scope_id": machine_id, + "contract_role": "COPIER_BW", "parent_id": "", "csrf_token": token, + }, follow_redirects=False, + ) + assert response.status_code == 302, response.get_data(as_text=True) + assert db.session.get(Meter, meter_id).contract_role == "COPIER_BW" + response = authenticated_client.post( + f"/meters/{meter_id}/edit", data={ + "name": "Compteur historique", "meter_type": "photocopieur", "unit": "pages", + "scope_type": "equipment", "scope_id": machine_id, + "contract_role": "NONE", "parent_id": "", "csrf_token": token, + }, follow_redirects=False, + ) + assert response.status_code == 302 + assert db.session.get(Meter, meter_id).contract_role == "NONE" + + +def test_c4_final_copier_estimation_uses_school_days_not_calendar_days(app): + with app.app_context(): + supplier = _supplier("TEST_UI_C4_FINAL_SCHOOL_SUPPLIER") + contract = create_contract( + supplier_id=supplier.id, name="TEST_UI_C4_FINAL_SCHOOL", + start_date=date(2026, 2, 9), end_date=date(2026, 12, 31), + anniversary_month=2, anniversary_day=1, + ) + period = contract.periods[0] + machine = _machine("TEST_UI_C4_FINAL_SCHOOL_MACHINE") + db.session.add(ContractEquipmentAssignment( + contract=contract, equipment=machine, entry_date=period.start_date, + )) + meter = _meter(machine, "N&B", current=2600) + # 30 jours calendaires, dont 7 jours de fermeture en semaine : + # 1 500 pages / 15 jours scolaires = 100 pages/jour scolaire. + db.session.add(CollegeClosure( + name="TEST_UI_C4_FINAL_VACANCES", start_date=date(2026, 1, 5), + end_date=date(2026, 1, 13), closure_type="vacances", work_hours_type="none", + )) + db.session.flush() + assert _school_days(date(2026, 1, 1), date(2026, 1, 30)) == 15 + assert _school_days(date(2026, 2, 9), date(2026, 2, 14)) == 5 + _reading(meter, 1000, date(2026, 1, 1)) + _reading(meter, 2500, date(2026, 1, 31)) + _reading(meter, 2600, date(2026, 2, 12)) + usage = calculate_contract_usage(period) + # 3 jours scolaires manquants en février : 300 pages estimées. + assert usage["quality"] == "ESTIMATED" + assert 290 <= usage["bw"] <= 310