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